zettelkasten.synapse.trace_surface

zettelkasten.synapse.trace_surface

Deterministic, redacted reasoning trace for grounded navigation.

The trace is projected only from parsed MemDSL structures and router-owned verdicts. It deliberately excludes source-node bodies, raw model output, filesystem paths, and verifier payloads. Any structural disagreement raises TraceSurfaceIntegrityError so callers can withhold the trace without weakening or rewriting the independently adjudicated navigation verdict.

TraceSurfaceIntegrityError

Bases: ValueError

The trusted trace cannot be projected without an integrity mismatch.

Source code in zettelkasten/synapse/trace_surface.py

class TraceSurfaceIntegrityError(ValueError):
    """The trusted trace cannot be projected without an integrity mismatch."""

build_trace_surface

build_trace_surface(nav_result: 'NavResult') -> dict[str, Any]

Project a version-1 redacted trace from one completed navigation run.

Source code in zettelkasten/synapse/trace_surface.py

def build_trace_surface(nav_result: "NavResult") -> dict[str, Any]:
    """Project a version-1 redacted trace from one completed navigation run."""

    decision = nav_result.decision
    if decision.verdict not in _VERDICTS:
        raise TraceSurfaceIntegrityError("trace verdict is not canonical")

    claims: list[dict[str, Any]] = []
    citations: list[dict[str, Any]] = []
    assertions: list[dict[str, str]] = []
    conclusion_id = decision.conclusion or ""

    if nav_result.output is not None and nav_result.output.reasoning is not None:
        dag = nav_result.output.reasoning
        try:
            dag.validate()
        except Exception as exc:
            raise TraceSurfaceIntegrityError("trace reasoning DAG is invalid") from exc
        if not conclusion_id or conclusion_id != dag.conclusion:
            raise TraceSurfaceIntegrityError("trace conclusion disagrees with the parsed DAG")
        closure = _claim_closure(dag, conclusion_id)
        by_id = {claim.id: claim for claim in dag.claims}
        cite_to_address, citations = _citation_map(nav_result, closure)
        for claim_id in closure:
            claim = by_id[claim_id]
            verdict = decision.claim_verdicts.get(claim_id)
            if verdict is None:
                raise TraceSurfaceIntegrityError("trace claim has no router-owned status")
            claims.append(
                {
                    "id": claim.id,
                    "text": claim.text,
                    "inference_form": claim.form.value,
                    "premise_claim_ids": sorted(claim.premises),
                    "status": verdict.wire().value,
                    "flags": _flags(verdict),
                    "citation_addresses": sorted(
                        {
                            cite_to_address[cite]
                            for cite in claim.cites
                            if cite in cite_to_address
                        }
                    ),
                }
            )
        for binding in sorted(dag.prose_citations, key=lambda item: item.span):
            if len(binding.claim_ids) != 1:
                raise TraceSurfaceIntegrityError(
                    "trace prose assertion must bind exactly one claim"
                )
            claim_id = binding.claim_ids[0]
            if claim_id not in closure:
                raise TraceSurfaceIntegrityError(
                    "trace prose assertion cites outside the conclusion closure"
                )
            start, end = binding.span
            if not dag.prose or start < 0 or end > len(dag.prose) or start >= end:
                raise TraceSurfaceIntegrityError("trace prose assertion span is invalid")
            assertions.append({"text": dag.prose[start:end], "claim_id": claim_id})
    elif decision.conclusion:
        raise TraceSurfaceIntegrityError("trace conclusion exists without parsed reasoning")

    status = decision.status.value if decision.status is not None else None
    if decision.verdict == ANSWER and status not in _WIRE_STATUSES:
        raise TraceSurfaceIntegrityError("answer trace has no canonical status")
    if decision.verdict == ABSTAIN and status is not None:
        raise TraceSurfaceIntegrityError("abstention trace must not assert answer status")

    disclosures = [_DETAIL_OUTDATED] if decision.outdated else []
    return {
        "version": TRACE_SURFACE_VERSION,
        "verdict": decision.verdict,
        "status": status,
        "conclusion_claim_id": conclusion_id or None,
        "claims": claims,
        "citations": citations,
        "assertions": assertions,
        "disclosures": disclosures,
        "abstention": (
            {
                "reasons": list(decision.reasons),
                "halted_reason": nav_result.halted_reason,
                "detail": decision.detail,
            }
            if decision.verdict == ABSTAIN
            else None
        ),
        "telemetry": {
            "hops": nav_result.hops,
            "renavigations": nav_result.renavigations,
            "parse_repairs": nav_result.parse_repairs,
            "tokens_spent": nav_result.tokens_spent,
            "loop_break": nav_result.loop_break,
        },
    }

validate_trace_surface_payload

validate_trace_surface_payload(surface: Any) -> str | None

Validate the redacted public trace shape; return the first error.

Source code in zettelkasten/synapse/trace_surface.py

def validate_trace_surface_payload(surface: Any) -> str | None:
    """Validate the redacted public trace shape; return the first error."""

    if not isinstance(surface, dict):
        return "trace_surface must be an object"
    expected = {
        "version",
        "verdict",
        "status",
        "conclusion_claim_id",
        "claims",
        "citations",
        "assertions",
        "disclosures",
        "abstention",
        "telemetry",
    }
    if set(surface) != expected or surface.get("version") != TRACE_SURFACE_VERSION:
        return "trace_surface fields/version do not match the version-1 contract"
    if surface.get("verdict") not in _VERDICTS:
        return "trace_surface.verdict is not canonical"
    status = surface.get("status")
    if status is not None and status not in _WIRE_STATUSES:
        return "trace_surface.status is not canonical"
    if not isinstance(surface.get("claims"), list):
        return "trace_surface.claims must be a list"
    if not isinstance(surface.get("citations"), list):
        return "trace_surface.citations must be a list"
    if not isinstance(surface.get("assertions"), list):
        return "trace_surface.assertions must be a list"
    if not isinstance(surface.get("disclosures"), list):
        return "trace_surface.disclosures must be a list"

    claim_ids: set[str] = set()
    for claim in surface["claims"]:
        if not isinstance(claim, dict) or set(claim) != {
            "id",
            "text",
            "inference_form",
            "premise_claim_ids",
            "status",
            "flags",
            "citation_addresses",
        }:
            return "trace_surface claim fields are invalid"
        if not isinstance(claim["id"], str) or not claim["id"] or claim["id"] in claim_ids:
            return "trace_surface claim ids must be unique non-empty strings"
        claim_ids.add(claim["id"])
        if not isinstance(claim["text"], str) or not claim["text"]:
            return "trace_surface claim text must be non-empty"
        if claim["inference_form"] not in _INFERENCE_FORMS:
            return "trace_surface claim inference form is not canonical"
        if claim["status"] not in _WIRE_STATUSES:
            return "trace_surface claim status is not canonical"
        flags = claim["flags"]
        if not isinstance(flags, dict) or set(flags) != set(_FLAG_FIELDS):
            return "trace_surface claim flags are invalid"
        if any(not isinstance(flags[field], bool) for field in _FLAG_FIELDS):
            return "trace_surface claim flags must be booleans"
        for field in ("premise_claim_ids", "citation_addresses"):
            if not isinstance(claim[field], list) or any(
                not isinstance(item, str) or not item for item in claim[field]
            ):
                return f"trace_surface claim {field} is invalid"

    for claim in surface["claims"]:
        if any(item not in claim_ids for item in claim["premise_claim_ids"]):
            return "trace_surface premise references an unknown claim"
    conclusion = surface.get("conclusion_claim_id")
    if conclusion is not None and conclusion not in claim_ids:
        return "trace_surface conclusion references an unknown claim"

    seen_addresses: set[str] = set()
    for citation in surface["citations"]:
        if not isinstance(citation, dict) or set(citation) != {
            "address",
            "claim_ids",
            "status",
            "flags",
        }:
            return "trace_surface citation fields are invalid"
        address = citation["address"]
        try:
            canonical = Address.parse(address).to_token()
        except Exception:
            return "trace_surface citation address is invalid"
        if canonical != address or address in seen_addresses:
            return "trace_surface citation addresses must be canonical and unique"
        seen_addresses.add(address)
        if citation["status"] not in _WIRE_STATUSES:
            return "trace_surface citation status is not canonical"
        if (
            not isinstance(citation["claim_ids"], list)
            or not citation["claim_ids"]
            or any(item not in claim_ids for item in citation["claim_ids"])
        ):
            return "trace_surface citation claim ids are invalid"
        flags = citation["flags"]
        if not isinstance(flags, dict) or set(flags) != set(_FLAG_FIELDS):
            return "trace_surface citation flags are invalid"

    for assertion in surface["assertions"]:
        if (
            not isinstance(assertion, dict)
            or set(assertion) != {"text", "claim_id"}
            or not isinstance(assertion["text"], str)
            or not assertion["text"]
            or assertion["claim_id"] not in claim_ids
        ):
            return "trace_surface assertion is invalid"

    abstention = surface.get("abstention")
    if surface["verdict"] == ABSTAIN:
        if not isinstance(abstention, dict) or set(abstention) != {
            "reasons",
            "halted_reason",
            "detail",
        }:
            return "trace_surface abstention fields are invalid"
    elif abstention is not None:
        return "answer trace must not carry an abstention"

    telemetry = surface.get("telemetry")
    if not isinstance(telemetry, dict) or set(telemetry) != {
        "hops",
        "renavigations",
        "parse_repairs",
        "tokens_spent",
        "loop_break",
    }:
        return "trace_surface telemetry fields are invalid"
    if any(
        isinstance(telemetry[field], bool)
        or not isinstance(telemetry[field], int)
        or telemetry[field] < 0
        for field in ("hops", "renavigations", "parse_repairs", "tokens_spent")
    ):
        return "trace_surface telemetry counters must be non-negative integers"
    if not isinstance(telemetry["loop_break"], str):
        return "trace_surface telemetry loop_break must be a string"
    return None