Task-knowledge query

akms.task_context.query

Compose uncapped exact task requirements with ordinary advisory queries.

SelectionClass

Bases: str, Enum

Why a node was selected, in deterministic precedence order.

NodeSelection dataclass

NodeSelection(
    node_id: str,
    selection_class: SelectionClass,
    node_data: dict[str, Any],
    reasons: tuple[str, ...],
)

One selected graph node with its class, data, and audit reasons.

TaskKnowledgeQueryResult dataclass

TaskKnowledgeQueryResult(
    selections: tuple[NodeSelection, ...],
)

Canonical required, coactivated, then advisory node selections.

RequiredNodeUnavailableError

RequiredNodeUnavailableError(issues: Mapping[str, str])

Bases: ValueError

Raised before advisory querying when exact requirements cannot load.

Source code in packages/akms/src/akms/task_context/query.py

def __init__(self, issues: Mapping[str, str]):
    if not issues:
        raise ValueError("RequiredNodeUnavailableError requires issues")
    self.issues = {
        str(node_id): str(reason) for node_id, reason in sorted(issues.items())
    }
    self.node_ids = tuple(self.issues)
    detail = "; ".join(
        f"{node_id}: {reason}" for node_id, reason in self.issues.items()
    )
    super().__init__(f"Required task nodes are unavailable: {detail}")

query_task_knowledge

query_task_knowledge(
    graph: DiGraph,
    seeds: ResolvedSeeds,
    agent_role: AgentRole | str,
    *,
    config: PropagationConfig | None = None,
    max_depth: int = 2,
) -> TaskKnowledgeQueryResult

Compose exact requirements with the unchanged advisory query path.

Required nodes are validated and inserted independently of ordinary role, status, confidence, ranking, and loadout-cap filters. One-hop load_with targets are classified separately as coactivated. Remaining ordinary query_subgraph results retain their existing order and node data.

Source code in packages/akms/src/akms/task_context/query.py

def query_task_knowledge(
    graph: nx.DiGraph,
    seeds: ResolvedSeeds,
    agent_role: AgentRole | str,
    *,
    config: PropagationConfig | None = None,
    max_depth: int = 2,
) -> TaskKnowledgeQueryResult:
    """Compose exact requirements with the unchanged advisory query path.

    Required nodes are validated and inserted independently of ordinary role,
    status, confidence, ranking, and loadout-cap filters. One-hop ``load_with``
    targets are classified separately as coactivated. Remaining ordinary
    ``query_subgraph`` results retain their existing order and node data.
    """

    if not isinstance(seeds, ResolvedSeeds):
        raise TypeError("seeds must be ResolvedSeeds")

    required_node_ids = seeds.all_exact_node_ids
    issues = _required_issues(graph, required_node_ids)
    if issues:
        raise RequiredNodeUnavailableError(issues)

    ordinary = query_subgraph(
        graph,
        list(seeds.advisory_tags),
        agent_role,
        config=config,
        max_depth=max_depth,
    )
    ordinary_base_ids = tuple(
        node_id for node_id, node_data in ordinary if not node_data.get("_coactivated")
    )
    required_set = frozenset(required_node_ids)
    coactivation_reasons = _load_with_targets(
        graph,
        required_node_ids + ordinary_base_ids,
        required_set,
    )
    coactivated_node_ids = tuple(sorted(coactivation_reasons))
    coactivated_set = frozenset(coactivated_node_ids)

    selections: list[NodeSelection] = []
    for node_id in required_node_ids:
        node_data = dict(graph.nodes[node_id])
        node_data.pop("_coactivated", None)
        selections.append(
            NodeSelection(
                node_id=node_id,
                selection_class=SelectionClass.REQUIRED,
                node_data=node_data,
                reasons=seeds.reasons[node_id],
            )
        )

    for node_id in coactivated_node_ids:
        node_data = dict(graph.nodes[node_id])
        node_data["_coactivated"] = True
        selections.append(
            NodeSelection(
                node_id=node_id,
                selection_class=SelectionClass.COACTIVATED,
                node_data=node_data,
                reasons=tuple(coactivation_reasons[node_id]),
            )
        )

    advisory_reason = (
        "advisory tag query: " + ", ".join(seeds.advisory_tags)
        if seeds.advisory_tags
        else "advisory tag query: empty"
    )
    for node_id, node_data in ordinary:
        if node_id in required_set or node_id in coactivated_set:
            continue
        selections.append(
            NodeSelection(
                node_id=node_id,
                selection_class=SelectionClass.ADVISORY,
                node_data=node_data,
                reasons=(advisory_reason,),
            )
        )

    return TaskKnowledgeQueryResult(selections=tuple(selections))