Resolution manifests

akms.task_context.manifest

Deterministic audit manifests for task-knowledge resolution.

The fingerprint boundary is intentionally narrower than the task JSON. Only fields consumed by :class:~akms.task_context.resolve.TaskSeeds, current changed paths, the query role, and resolver/index/graph versions participate. Execution state such as routing evidence, review results, and completion metadata must not invalidate retrieval caches.

ResolutionPathSpec

Bases: BaseModel

One normalized path spec exactly as consumed by the resolver.

DeclaredTaskPaths

Bases: BaseModel

Task fields that declare repository paths for exact retrieval.

RetrievalTaskFields

Bases: BaseModel

Non-path task fields consumed by task seed derivation.

ResolutionInputs

Bases: BaseModel

All and only inputs that own a resolution fingerprint.

ResolvedSeedsManifest

Bases: BaseModel

Canonical resolved seed details retained for audit.

ResolutionSelectedNode

Bases: BaseModel

Manifest-safe selection metadata, excluding volatile node data.

ResolutionManifest

Bases: BaseModel

Deterministic record of one task-knowledge resolution.

StaleResolutionManifestError

StaleResolutionManifestError(
    *, manifest_fingerprint: str, current_fingerprint: str
)

Bases: ValueError

Raised when current retrieval inputs no longer match a manifest.

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

def __init__(
    self,
    *,
    manifest_fingerprint: str,
    current_fingerprint: str,
):
    self.manifest_fingerprint = manifest_fingerprint
    self.current_fingerprint = current_fingerprint
    super().__init__(
        "Resolution manifest is stale: "
        f"stored={manifest_fingerprint}, current={current_fingerprint}"
    )

canonicalize_resolution_inputs

canonicalize_resolution_inputs(
    *,
    task: TaskSeeds | Mapping[str, Any],
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
) -> ResolutionInputs

Extract the explicit retrieval-owned subset of current task inputs.

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

def canonicalize_resolution_inputs(
    *,
    task: TaskSeeds | Mapping[str, Any],
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
) -> ResolutionInputs:
    """Extract the explicit retrieval-owned subset of current task inputs."""

    seeds = _task_seeds(task, changed_paths=changed_paths)
    specs = canonicalize_task_path_specs(seeds)

    def specs_for(source: str) -> tuple[ResolutionPathSpec, ...]:
        return tuple(
            ResolutionPathSpec(value=spec.value, kind=spec.kind)
            for spec in specs
            if spec.source == source
        )

    return ResolutionInputs(
        declared_paths=DeclaredTaskPaths(
            scope=specs_for("scope"),
            deliverables=specs_for("deliverable"),
        ),
        changed_paths=specs_for("changed_file"),
        task_fields=RetrievalTaskFields(
            title=seeds.title,
            objective=seeds.objective,
            implementation_steps=seeds.implementation_steps,
            symbols=seeds.symbols,
            akms_tags=seeds.advisory_tags,
        ),
        role=agent_role,
        graph_version=graph_version,
        route_index_hash=route_index_hash,
        resolver_version=resolver_version,
    )

fingerprint_resolution_inputs

fingerprint_resolution_inputs(
    inputs: ResolutionInputs,
) -> str

Return the SHA-256 fingerprint of canonical retrieval inputs.

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

def fingerprint_resolution_inputs(inputs: ResolutionInputs) -> str:
    """Return the SHA-256 fingerprint of canonical retrieval inputs."""

    if not isinstance(inputs, ResolutionInputs):
        raise TypeError("inputs must be ResolutionInputs")
    payload = inputs.model_dump(mode="json")
    return hashlib.sha256(_canonical_json_bytes(payload)).hexdigest()

compute_resolution_fingerprint

compute_resolution_fingerprint(
    *,
    task: TaskSeeds | Mapping[str, Any],
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
) -> str

Canonicalize current inputs and return their stable fingerprint.

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

def compute_resolution_fingerprint(
    *,
    task: TaskSeeds | Mapping[str, Any],
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
) -> str:
    """Canonicalize current inputs and return their stable fingerprint."""

    inputs = canonicalize_resolution_inputs(
        task=task,
        changed_paths=changed_paths,
        agent_role=agent_role,
        graph_version=graph_version,
        route_index_hash=route_index_hash,
        resolver_version=resolver_version,
    )
    return fingerprint_resolution_inputs(inputs)

create_resolution_manifest

create_resolution_manifest(
    *,
    task: TaskSeeds | Mapping[str, Any],
    resolved_seeds: ResolvedSeeds,
    query_result: TaskKnowledgeQueryResult,
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
    generated_at: datetime | str | None = None,
) -> ResolutionManifest

Build a validated manifest from resolver and query results.

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

def create_resolution_manifest(
    *,
    task: TaskSeeds | Mapping[str, Any],
    resolved_seeds: ResolvedSeeds,
    query_result: TaskKnowledgeQueryResult,
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
    generated_at: datetime | str | None = None,
) -> ResolutionManifest:
    """Build a validated manifest from resolver and query results."""

    if not isinstance(query_result, TaskKnowledgeQueryResult):
        raise TypeError("query_result must be TaskKnowledgeQueryResult")
    inputs = canonicalize_resolution_inputs(
        task=task,
        changed_paths=changed_paths,
        agent_role=agent_role,
        graph_version=graph_version,
        route_index_hash=route_index_hash,
        resolver_version=resolver_version,
    )
    selected_nodes = tuple(
        ResolutionSelectedNode(
            node_id=selection.node_id,
            selection_class=selection.selection_class,
            reasons=selection.reasons,
        )
        for selection in query_result.selections
    )
    return ResolutionManifest(
        # The parameter accepts a str for caller convenience; the model field is
        # a tz-aware datetime and pydantic coerces on the way in.
        generated_at=(
            datetime.now(UTC)
            if generated_at is None
            else cast("datetime", generated_at)
        ),
        fingerprint=fingerprint_resolution_inputs(inputs),
        inputs=inputs,
        resolved_seeds=ResolvedSeedsManifest.from_resolved(resolved_seeds),
        selected_nodes=selected_nodes,
    )

canonical_manifest_bytes

canonical_manifest_bytes(
    manifest: ResolutionManifest | Mapping[str, Any],
) -> bytes

Serialize a manifest as compact, sorted-key UTF-8 JSON.

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

def canonical_manifest_bytes(
    manifest: ResolutionManifest | Mapping[str, Any],
) -> bytes:
    """Serialize a manifest as compact, sorted-key UTF-8 JSON."""

    validated = _revalidate_resolution_manifest(manifest)
    return _canonical_json_bytes(validated.model_dump(mode="json"))

write_resolution_manifest

write_resolution_manifest(
    path: str | Path,
    manifest: ResolutionManifest | Mapping[str, Any],
) -> Path

Atomically replace path with a complete canonical manifest.

The temporary file is created in the destination directory, flushed and fsynced before :func:os.replace, and removed on every failure path. The destination directory is fsynced after replacement where supported.

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

def write_resolution_manifest(
    path: str | Path,
    manifest: ResolutionManifest | Mapping[str, Any],
) -> Path:
    """Atomically replace ``path`` with a complete canonical manifest.

    The temporary file is created in the destination directory, flushed and
    fsynced before :func:`os.replace`, and removed on every failure path. The
    destination directory is fsynced after replacement where supported.
    """

    destination = Path(path)
    destination.parent.mkdir(parents=True, exist_ok=True)
    payload = canonical_manifest_bytes(manifest)
    descriptor, temporary_name = tempfile.mkstemp(
        prefix=f".{destination.name}.",
        suffix=".tmp",
        dir=str(destination.parent),
    )
    temporary = Path(temporary_name)
    try:
        with os.fdopen(descriptor, "wb") as handle:
            descriptor = -1
            handle.write(payload)
            handle.flush()
            os.fsync(handle.fileno())
        os.replace(temporary, destination)
        _fsync_parent_directory(destination.parent)
    except BaseException:
        if descriptor >= 0:
            os.close(descriptor)
        try:
            temporary.unlink()
        except FileNotFoundError:
            pass
        raise
    return destination

load_resolution_manifest

load_resolution_manifest(
    path: str | Path,
) -> ResolutionManifest

Read and validate one manifest from disk.

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

def load_resolution_manifest(path: str | Path) -> ResolutionManifest:
    """Read and validate one manifest from disk."""

    payload = json.loads(Path(path).read_text(encoding="utf-8"))
    return ResolutionManifest.model_validate(payload)

validate_resolution_manifest

validate_resolution_manifest(
    manifest: ResolutionManifest | Mapping[str, Any],
    *,
    task: TaskSeeds | Mapping[str, Any],
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
) -> ResolutionManifest

Return manifest or raise when current retrieval inputs are stale.

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

def validate_resolution_manifest(
    manifest: ResolutionManifest | Mapping[str, Any],
    *,
    task: TaskSeeds | Mapping[str, Any],
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
) -> ResolutionManifest:
    """Return ``manifest`` or raise when current retrieval inputs are stale."""

    validated = _revalidate_resolution_manifest(manifest)
    current_fingerprint = compute_resolution_fingerprint(
        task=task,
        changed_paths=changed_paths,
        agent_role=agent_role,
        graph_version=graph_version,
        route_index_hash=route_index_hash,
        resolver_version=resolver_version,
    )
    if validated.fingerprint != current_fingerprint:
        raise StaleResolutionManifestError(
            manifest_fingerprint=validated.fingerprint,
            current_fingerprint=current_fingerprint,
        )
    return validated

resolution_manifest_is_stale

resolution_manifest_is_stale(
    manifest: ResolutionManifest | Mapping[str, Any],
    *,
    task: TaskSeeds | Mapping[str, Any],
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
) -> bool

Return whether current retrieval inputs invalidate manifest.

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

def resolution_manifest_is_stale(
    manifest: ResolutionManifest | Mapping[str, Any],
    *,
    task: TaskSeeds | Mapping[str, Any],
    agent_role: AgentRole | str,
    graph_version: str,
    route_index_hash: str,
    changed_paths: Sequence[str] | None = None,
    resolver_version: str = RESOLUTION_RESOLVER_VERSION,
) -> bool:
    """Return whether current retrieval inputs invalidate ``manifest``."""

    try:
        validate_resolution_manifest(
            manifest,
            task=task,
            changed_paths=changed_paths,
            agent_role=agent_role,
            graph_version=graph_version,
            route_index_hash=route_index_hash,
            resolver_version=resolver_version,
        )
    except StaleResolutionManifestError:
        return True
    return False