Deterministic public answer projection for grounded navigation.
The navigation engine already has the only inputs that may license a public
answer: a router-owned ANSWER decision, the parsed MemDSL reasoning DAG, and
the final render envelope. This module projects those objects into a small,
versioned JSON surface without asking a model to restate anything and without
copying source bodies.
The projection is deliberately fail-closed. Any disagreement between the
router decision, conclusion claim, claim verdicts, or citation references raises
:class:AnswerSurfaceIntegrityError; the public navigate wrapper converts that
condition to a labeled abstention instead of exposing a partial answer.
AnswerSurfaceIntegrityError
Bases: ValueError
The accepted answer cannot be projected without an integrity gap.
Source code in zettelkasten/synapse/answer_surface.py
| class AnswerSurfaceIntegrityError(ValueError):
"""The accepted answer cannot be projected without an integrity gap."""
|
build_answer_surface
build_answer_surface(nav_result: 'NavResult') -> 'dict[str, Any] | None'
Build the version-1 answer surface from a completed navigation result.
Returns None for every non-answer result. For an answer, all data comes
from the accepted :class:RouterDecision, parsed :class:ReasoningDAG, and
final :class:Envelope carried by nav_result:
- answer text is the router-approved OUTPUT v2 prose when present, otherwise
the v1 conclusion claim text;
- status comes only from the router decision;
- premises are the deterministic transitive claim closure;
- citations are grouped by canonical source :class:
Address token and carry
the citing claim ids plus router-derived node status; and
- outdated disclosure reuses the single canonical render constant.
Source node bodies, titles, provenance strings, and model-authored rewrites
are intentionally absent.
Source code in zettelkasten/synapse/answer_surface.py
| def build_answer_surface(nav_result: "NavResult") -> "dict[str, Any] | None":
"""Build the version-1 answer surface from a completed navigation result.
Returns ``None`` for every non-answer result. For an answer, all data comes
from the accepted :class:`RouterDecision`, parsed :class:`ReasoningDAG`, and
final :class:`Envelope` carried by ``nav_result``:
* answer text is the router-approved OUTPUT v2 prose when present, otherwise
the v1 conclusion claim text;
* status comes only from the router decision;
* premises are the deterministic transitive claim closure;
* citations are grouped by canonical source :class:`Address` token and carry
the citing claim ids plus router-derived node status; and
* outdated disclosure reuses the single canonical render constant.
Source node bodies, titles, provenance strings, and model-authored rewrites
are intentionally absent.
"""
decision = nav_result.decision
if decision.verdict != ANSWER:
return None
if not isinstance(decision.status, EpistemicStatus):
raise AnswerSurfaceIntegrityError("answer has no router-owned status")
if nav_result.output is None or nav_result.output.reasoning is None:
raise AnswerSurfaceIntegrityError("answer has no parsed reasoning DAG")
if nav_result.envelope is None:
raise AnswerSurfaceIntegrityError("answer has no final render envelope")
dag = nav_result.output.reasoning
try:
dag.validate()
except Exception as exc:
raise AnswerSurfaceIntegrityError("answer reasoning DAG is invalid") from exc
by_id = {claim.id: claim for claim in dag.claims}
conclusion_id = decision.conclusion
if not conclusion_id or conclusion_id != dag.conclusion or conclusion_id not in by_id:
raise AnswerSurfaceIntegrityError("router conclusion does not match the parsed DAG")
conclusion = by_id[conclusion_id]
if not isinstance(conclusion.text, str) or not conclusion.text:
raise AnswerSurfaceIntegrityError("router conclusion has no answer text")
if not isinstance(conclusion.form, InferenceForm):
raise AnswerSurfaceIntegrityError("router conclusion has no canonical inference form")
conclusion_verdict = decision.claim_verdicts.get(conclusion_id)
if conclusion_verdict is None or conclusion_verdict.wire() is not decision.status:
raise AnswerSurfaceIntegrityError("router conclusion status mapping is inconsistent")
if bool(conclusion_verdict.outdated) != bool(decision.outdated):
raise AnswerSurfaceIntegrityError("router outdated mapping is inconsistent")
if decision.outdated and not decision.detail.endswith(f"; {_DETAIL_OUTDATED}"):
raise AnswerSurfaceIntegrityError("router answer lacks the canonical outdated disclosure")
premise_ids = _premise_order(dag, conclusion_id)
claim_order = [*premise_ids, conclusion_id]
for claim_id in claim_order:
if claim_id not in decision.claim_verdicts:
raise AnswerSurfaceIntegrityError("premise claim has no router-owned status")
answer_text = conclusion.text
if dag.prose:
if decision.traceability is None or not decision.traceability.ok:
raise AnswerSurfaceIntegrityError("licensed prose lacks an approved traceability verdict")
bound_claim_ids = {
claim_id
for citation in dag.prose_citations
for claim_id in citation.claim_ids
}
if not bound_claim_ids or not bound_claim_ids.issubset(set(claim_order)):
raise AnswerSurfaceIntegrityError(
"licensed prose cites a claim outside the router conclusion closure"
)
answer_text = dag.prose
ref_to_address: dict[str, str] = {}
addresses: set[str] = set()
for node in nav_result.envelope.nodes:
address = node.address.to_token()
if node.ref in ref_to_address or address in addresses:
raise AnswerSurfaceIntegrityError("final envelope has ambiguous citation mapping")
ref_to_address[node.ref] = address
addresses.add(address)
citations_by_address: dict[str, dict[str, Any]] = {}
for claim_id in claim_order:
claim = by_id[claim_id]
for cite in sorted(set(claim.cites)):
if cite in ref_to_address:
address = ref_to_address[cite]
else:
try:
address = Address.parse(cite).to_token()
except Exception as exc:
raise AnswerSurfaceIntegrityError(
"claim citation is neither a final-envelope ref nor an Address"
) from exc
if address not in addresses:
raise AnswerSurfaceIntegrityError(
"claim citation does not resolve in the final envelope"
)
verdict = _node_verdict(nav_result, cite=cite, address=address)
row = citations_by_address.get(address)
if row is None:
row = {
"address": address,
"claim_ids": [],
"status": verdict.wire().value,
"outdated": bool(verdict.outdated),
}
citations_by_address[address] = row
elif row["status"] != verdict.wire().value or row["outdated"] != bool(verdict.outdated):
raise AnswerSurfaceIntegrityError("grouped citation statuses are inconsistent")
if claim_id not in row["claim_ids"]:
row["claim_ids"].append(claim_id)
if not citations_by_address:
raise AnswerSurfaceIntegrityError("answer premise closure has no resolved citations")
if any(row["outdated"] for row in citations_by_address.values()) != bool(decision.outdated):
raise AnswerSurfaceIntegrityError("citation outdated mapping is inconsistent")
return {
"version": ANSWER_SURFACE_VERSION,
"text": answer_text,
"conclusion_claim_id": conclusion.id,
"inference_form": conclusion.form.value,
"status": decision.status.value,
"premise_claim_ids": premise_ids,
"citations": [citations_by_address[address] for address in sorted(citations_by_address)],
"outdated": bool(decision.outdated),
"disclosure": _DETAIL_OUTDATED if decision.outdated else None,
}
|
validate_answer_surface_payload
validate_answer_surface_payload(surface: Any) -> 'str | None'
Validate the self-contained public shape; return the first error, if any.
This is intentionally narrower than :func:build_answer_surface: a wire
consumer cannot re-check the hidden DAG/envelope mapping, but it can enforce
the version, field types, canonical Address/status vocabulary, uniqueness,
and inseparable outdated disclosure. The benchmark uses this code-defined
contract without adding fields to the frozen corpus.
Source code in zettelkasten/synapse/answer_surface.py
| def validate_answer_surface_payload(surface: Any) -> "str | None":
"""Validate the self-contained public shape; return the first error, if any.
This is intentionally narrower than :func:`build_answer_surface`: a wire
consumer cannot re-check the hidden DAG/envelope mapping, but it can enforce
the version, field types, canonical Address/status vocabulary, uniqueness,
and inseparable outdated disclosure. The benchmark uses this code-defined
contract without adding fields to the frozen corpus.
"""
if not isinstance(surface, dict):
return "answer_surface must be an object"
if set(surface) != _SURFACE_FIELDS:
return "answer_surface fields do not match the version-1 contract"
if surface.get("version") != ANSWER_SURFACE_VERSION:
return f"answer_surface.version must be {ANSWER_SURFACE_VERSION}"
for field in ("text", "conclusion_claim_id", "inference_form", "status"):
if not isinstance(surface.get(field), str) or not surface[field]:
return f"answer_surface.{field} must be a non-empty string"
if surface["status"] not in _WIRE_STATUSES:
return f"answer_surface.status is not a canonical wire status: {surface['status']!r}"
if surface["inference_form"] not in _INFERENCE_FORMS:
return (
"answer_surface.inference_form is not canonical: "
f"{surface['inference_form']!r}"
)
premise_ids = surface.get("premise_claim_ids")
if (
not isinstance(premise_ids, list)
or any(not isinstance(item, str) or not item for item in premise_ids)
or len(premise_ids) != len(set(premise_ids))
):
return "answer_surface.premise_claim_ids must be a unique list of non-empty strings"
citations = surface.get("citations")
if not isinstance(citations, list) or not citations:
return "answer_surface.citations must be a non-empty list"
seen_addresses: set[str] = set()
for index, citation in enumerate(citations):
if not isinstance(citation, dict):
return f"answer_surface.citations[{index}] must be an object"
if set(citation) != _CITATION_FIELDS:
return f"answer_surface.citations[{index}] fields do not match version 1"
address = citation.get("address")
if not isinstance(address, str) or not address:
return f"answer_surface.citations[{index}].address must be a non-empty string"
try:
canonical = Address.parse(address).to_token()
except Exception:
return f"answer_surface.citations[{index}].address is invalid"
if canonical != address or address in seen_addresses:
return "answer_surface citation addresses must be canonical and unique"
seen_addresses.add(address)
claim_ids = citation.get("claim_ids")
if (
not isinstance(claim_ids, list)
or not claim_ids
or any(not isinstance(item, str) or not item for item in claim_ids)
or len(claim_ids) != len(set(claim_ids))
):
return (
f"answer_surface.citations[{index}].claim_ids must be a unique, "
"non-empty list of strings"
)
if citation.get("status") not in _WIRE_STATUSES:
return f"answer_surface.citations[{index}].status is not canonical"
if not isinstance(citation.get("outdated"), bool):
return f"answer_surface.citations[{index}].outdated must be a boolean"
outdated = surface.get("outdated")
if not isinstance(outdated, bool):
return "answer_surface.outdated must be a boolean"
expected_disclosure = _DETAIL_OUTDATED if outdated else None
if surface.get("disclosure") != expected_disclosure:
return "answer_surface disclosure does not match canonical outdated semantics"
return None
|