Task seed resolution

akms.task_context.resolve

Deterministic task inputs to advisory and exact knowledge seeds.

Exact code-mirror and required-route selection is intentionally independent from tag derivation: code-mirror nodes are tagless by schema, while required routes bypass advisory query thresholds and caps.

TaskSeeds dataclass

TaskSeeds(
    scope: tuple[str, ...] = (),
    deliverables: tuple[str, ...] = (),
    changed_files: tuple[str, ...] = (),
    symbols: tuple[str, ...] = (),
    advisory_tags: tuple[str, ...] = (),
    title: str = "",
    objective: str = "",
    implementation_steps: tuple[str, ...] = (),
)

Retrieval-relevant task inputs before graph-bound resolution.

from_task classmethod

from_task(
    task: Mapping[str, Any],
    *,
    changed_files: Sequence[str] | None = None,
) -> TaskSeeds

Extract retrieval inputs from a task JSON-compatible mapping.

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

@classmethod
def from_task(
    cls,
    task: Mapping[str, Any],
    *,
    changed_files: Sequence[str] | None = None,
) -> TaskSeeds:
    """Extract retrieval inputs from a task JSON-compatible mapping."""

    task_changed = list(_text_tuple(task.get("changed_files")))
    task_changed.extend(_text_tuple(changed_files))
    return cls(
        scope=_text_tuple(task.get("scope")),
        deliverables=_text_tuple(task.get("deliverables")),
        changed_files=tuple(task_changed),
        symbols=_text_tuple(task.get("symbols")),
        advisory_tags=_text_tuple(task.get("akms_tags")),
        title=str(task.get("title", "")),
        objective=str(task.get("objective", "")),
        implementation_steps=_text_tuple(task.get("implementation_steps")),
    )

tag_task

tag_task() -> dict[str, object]

Return the legacy derive_tags input without changing its API.

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

def tag_task(self) -> dict[str, object]:
    """Return the legacy ``derive_tags`` input without changing its API."""

    return {
        "akms_tags": list(self.advisory_tags),
        "scope": list(self.scope),
        "title": self.title,
        "objective": self.objective,
        "implementation_steps": list(self.implementation_steps),
    }

ResolvedSeeds dataclass

ResolvedSeeds(
    advisory_tags: tuple[str, ...] = (),
    exact_mirror_node_ids: tuple[str, ...] = (),
    required_route_node_ids: tuple[str, ...] = (),
    reasons: dict[str, tuple[str, ...]] = dict(),
)

Canonical advisory tags and uncapped exact node seeds.

all_exact_node_ids property

all_exact_node_ids: tuple[str, ...]

Return the sorted union of mirror and required-route node IDs.

TaskPathSpec dataclass

TaskPathSpec(
    value: str,
    kind: Literal["exact", "directory", "glob"],
    source: Literal["scope", "deliverable", "changed_file"],
)

One canonical repository path input used by task resolution.

canonicalize_task_path_specs

canonicalize_task_path_specs(
    seeds: TaskSeeds,
) -> tuple[TaskPathSpec, ...]

Return normalized, filtered path specs exactly as resolution consumes.

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

def canonicalize_task_path_specs(seeds: TaskSeeds) -> tuple[TaskPathSpec, ...]:
    """Return normalized, filtered path specs exactly as resolution consumes."""

    if not isinstance(seeds, TaskSeeds):
        raise TypeError("seeds must be TaskSeeds")
    specs: set[TaskPathSpec] = set()
    sources: tuple[tuple[_PathSpecSource, Iterable[str]], ...] = (
        ("scope", seeds.scope),
        ("deliverable", seeds.deliverables),
        ("changed_file", seeds.changed_files),
    )
    for source, values in sources:
        for value in values:
            spec = _path_spec(value, source=source)
            if spec is not None:
                specs.add(spec)
    return tuple(sorted(specs, key=lambda spec: (spec.source, spec.kind, spec.value)))

resolve_task_seeds

resolve_task_seeds(
    graph: DiGraph,
    seeds: TaskSeeds | Mapping[str, Any],
    *,
    route_index: TaskRouteIndex
    | Mapping[str, Any]
    | str
    | Path
    | None = None,
    changed_files: Sequence[str] | None = None,
) -> ResolvedSeeds

Resolve advisory tags, exact mirror IDs, route IDs, and reasons.

Plain paths match exactly. A trailing slash declares a directory input and glob metacharacters declare a glob input. changed_files are always exact paths, even if a filename contains glob metacharacters.

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

def resolve_task_seeds(
    graph: nx.DiGraph,
    seeds: TaskSeeds | Mapping[str, Any],
    *,
    route_index: TaskRouteIndex | Mapping[str, Any] | str | Path | None = None,
    changed_files: Sequence[str] | None = None,
) -> ResolvedSeeds:
    """Resolve advisory tags, exact mirror IDs, route IDs, and reasons.

    Plain paths match exactly. A trailing slash declares a directory input and
    glob metacharacters declare a glob input. ``changed_files`` are always
    exact paths, even if a filename contains glob metacharacters.
    """

    if isinstance(seeds, Mapping):
        task_seeds = TaskSeeds.from_task(seeds, changed_files=changed_files)
    elif isinstance(seeds, TaskSeeds):
        if changed_files:
            task_seeds = TaskSeeds(
                scope=seeds.scope,
                deliverables=seeds.deliverables,
                changed_files=seeds.changed_files + tuple(changed_files),
                symbols=seeds.symbols,
                advisory_tags=seeds.advisory_tags,
                title=seeds.title,
                objective=seeds.objective,
                implementation_steps=seeds.implementation_steps,
            )
        else:
            task_seeds = seeds
    else:
        raise TypeError("seeds must be TaskSeeds or a task mapping")

    specs = canonicalize_task_path_specs(task_seeds)
    if _is_documentation_only(task_seeds, specs):
        specs = ()
    reasons: dict[str, set[str]] = {}
    mirror_ids: set[str] = set()
    route_ids: set[str] = set()

    for node_id, data in graph.nodes(data=True):
        if not _is_code_mirror(data):
            continue
        source_file = data.get("source_file")
        if not isinstance(source_file, str):
            continue
        try:
            repository_path = normalize_repository_path(source_file)
        except ValueError:
            continue
        for spec in specs:
            if spec.matches(repository_path):
                node_key = str(node_id)
                mirror_ids.add(node_key)
                reasons.setdefault(node_key, set()).add(
                    _match_reason(
                        subject="mirror source_file",
                        repository_path=repository_path,
                        spec=spec,
                    )
                )

    if route_index is not None:
        index = parse_route_index(route_index, graph=graph)
        for repository_path, records in index.by_path.items():
            for spec in specs:
                if not spec.matches(repository_path):
                    continue
                for record in records:
                    route_ids.add(record.node_id)
                    reasons.setdefault(record.node_id, set()).add(
                        (
                            _match_reason(
                                subject="required route",
                                repository_path=repository_path,
                                spec=spec,
                            )
                            + f": {record.reason}"
                            + f" [provenance={_route_provenance(record)}]"
                        )
                    )
        for symbol in task_seeds.symbols:
            for record in index.by_symbol.get(symbol, ()):
                route_ids.add(record.node_id)
                reasons.setdefault(record.node_id, set()).add(
                    f"required symbol route '{symbol}': {record.reason} "
                    f"[provenance={_route_provenance(record)}]"
                )

    advisory_tags = tuple(sorted(set(derive_tags(graph, task_seeds.tag_task()))))
    return ResolvedSeeds(
        advisory_tags=advisory_tags,
        exact_mirror_node_ids=tuple(mirror_ids),
        required_route_node_ids=tuple(route_ids),
        reasons={
            node_id: tuple(node_reasons) for node_id, node_reasons in reasons.items()
        },
    )