From 41664ce60d20889c39d3183eabb534fa8acbbbef Mon Sep 17 00:00:00 2001 From: Shay Date: Wed, 10 Jun 2026 10:53:54 -0700 Subject: [PATCH] feat(ask): add carried-handle acquisition seam for served ASK MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The missing-boundary doc (#682) established that chat/runtime.py has no lawful in-turn provider for a QUESTION_NEEDED ContemplationResult: the only producer is the off-serving contemplation pass, and scanning teaching/questions/ would be new, unspecified acquisition logic. This slice builds the recommended unblocking seam instead of the forbidden runtime shortcut: carried-handle acquisition. core/epistemic_disclosure/ask_handle.py: - AskArtifactHandle — an explicit, caller-carried reference to one already-produced Q1-D DeliveredQuestion artifact. Provenance is the producer's own content address (sha256 of blocking_reason:slot_name: text, also the artifact filename stem) — the only identity field the Q1-D artifact carries today. No turn/case provenance exists yet; that boundary is documented, not invented. - resolve_served_ask_handle — gate-first (zero filesystem access while ask_serving_enabled is dark), structural handle validation, single explicit-path artifact read (no glob/iterdir/scan), and content-hash re-derivation so a stale/replaced/tampered artifact fails closed. Emits a ResolvedAskCandidate duck-typed to what evaluate_served_ask reads (terminal/question_path/proposal_path) — no question prose is ever constructed or carried. - acquire_served_ask_from_handle — composes resolution into the existing acquire_served_ask_candidate / evaluate_served_ask stack. All Q1-D artifact policy (status/review/served/answer-binding/text/ slot/collision) stays delegated; the seam owns identity only. Not changed: chat/runtime.py, chat(...) signature, ChatResponse, TurnEvent, telemetry schema, Q1B_ASK_CARVE_OUT, proposal_allowed, VERIFIED serving, CLAIMS/metrics/lane pins. The seam imports neither generate.contemplation (pass_manager) nor core.epistemic_questions render/delivery — enforced by AST tests. Tests (tests/test_ask_handle.py, 18 cases): gate-dark no-I/O proof, fallback preservation, valid resolution served end-to-end through the existing stack, proposal-artifact / missing / malformed / invalid-Q1-D / path-collision / stale-hash / non-addressed-filename / structurally- invalid-handle fail-closed paths, no-discovery-without-exact-handle, and AST guards for scan primitives, forbidden imports, and prose. Verification: focused ASK+governance suites 396 passed; architectural invariants 56 passed; smoke-equivalent suite 90 passed; lane SHAs 8/9 content-match (the 9th is the documented local public_demo wall-clock flake, 48.3s vs the 30s ADR-0099 budget — no content drift, no re-pin). --- core/epistemic_disclosure/__init__.py | 17 + core/epistemic_disclosure/ask_handle.py | 251 +++++++++++++++ tests/test_ask_handle.py | 397 ++++++++++++++++++++++++ 3 files changed, 665 insertions(+) create mode 100644 core/epistemic_disclosure/ask_handle.py create mode 100644 tests/test_ask_handle.py diff --git a/core/epistemic_disclosure/__init__.py b/core/epistemic_disclosure/__init__.py index 142332c5..d8dbbe5d 100644 --- a/core/epistemic_disclosure/__init__.py +++ b/core/epistemic_disclosure/__init__.py @@ -25,6 +25,11 @@ Shipped so far (all off-serving — nothing here imports ``generate.derivation`` * :mod:`~core.epistemic_disclosure.ask_acquisition` — a serving-safe acquisition seam that gate-checks before provider execution and delegates candidate validation to the ASK artifact adapter. It does not call pass_manager directly. + * :mod:`~core.epistemic_disclosure.ask_handle` — the carried-handle acquisition + seam: resolves an explicit, content-addressed reference to one already-produced + Q1-D artifact into an acquisition-compatible candidate. It never produces, + renders, or scans; identity checks only — artifact policy stays delegated to + the adapter. * :mod:`~core.epistemic_disclosure.verified_contract` (P1-A) — the VERIFIED contract: the obligation, the proof shape, the validator, and the single sanctioned route to ``EpistemicState.VERIFIED`` / ``DisclosureClaim.VERIFIED``. Contract only — no @@ -38,6 +43,13 @@ from core.epistemic_disclosure.ask_acquisition import ( ContemplationProvider, acquire_served_ask_candidate, ) +from core.epistemic_disclosure.ask_handle import ( + AskArtifactHandle, + AskHandleResolution, + ResolvedAskCandidate, + acquire_served_ask_from_handle, + resolve_served_ask_handle, +) from core.epistemic_disclosure.ask_serving import ( ServedAskDecision, evaluate_served_ask, @@ -75,8 +87,11 @@ __all__ = [ "Q1B_ASK_CARVE_OUT", "VERIFICATION_OBLIGATION", "AskAcquisitionDecision", + "AskArtifactHandle", + "AskHandleResolution", "ContemplationProvider", "DisclosureClaim", + "ResolvedAskCandidate", "LimitationAssessment", "LimitationKind", "MissingSlot", @@ -88,11 +103,13 @@ __all__ = [ "VerificationResult", "VerificationVerdict", "acquire_served_ask_candidate", + "acquire_served_ask_from_handle", "assess_from_attempt", "assess_from_family", "choose_served_disposition", "disclosure_for_verification", "evaluate_served_ask", "evaluate_verification", + "resolve_served_ask_handle", "terminal_for_action", ] diff --git a/core/epistemic_disclosure/ask_handle.py b/core/epistemic_disclosure/ask_handle.py new file mode 100644 index 00000000..f7299da1 --- /dev/null +++ b/core/epistemic_disclosure/ask_handle.py @@ -0,0 +1,251 @@ +"""Carried-handle acquisition seam for served ASK (Stage 2). + +The missing-boundary analysis +(``docs/analysis/ask-runtime-wiring-missing-boundary-2026-06-10.md``) established +that the serving turn has no lawful way to obtain a ``QUESTION_NEEDED`` +``ContemplationResult``: the only producer is the off-serving contemplation pass, +and scanning the ``teaching/questions`` sink would be new, unspecified acquisition +logic. This module is the recommended unblocking slice — **carried-handle +acquisition**: an explicit, caller-carried reference to one already-produced Q1-D +``DeliveredQuestion`` artifact is *resolved* (never produced, never rendered, +never discovered by scan) into an acquisition-compatible candidate for the +existing ASK serving stack. + +The handle's provenance is the Q1-D artifact's own content address. The producer +(:mod:`core.epistemic_questions.delivery`) writes each artifact to +``{content_hash}.json`` where ``content_hash`` is +``sha256(f"{blocking_reason}:{slot_name}:{text}")`` over the artifact's own +fields. Resolution therefore verifies three independent things: + +1. the handle is structurally valid (non-empty path, 64-hex content hash, and the + path's filename is exactly the producer's content-addressed name); +2. the referenced artifact exists and parses as a JSON object; +3. the artifact's body re-hashes to the handle's content hash — a stale, + replaced, or tampered artifact fails closed here. + +Everything else — ``status == "question_only"``, ``requires_review``, +``served is False``, ``answer_binding`` absence, question text/slot validity, the +``question_path != proposal_path`` collision — is **delegated** to +:func:`core.epistemic_disclosure.ask_serving.evaluate_served_ask` via +:func:`core.epistemic_disclosure.ask_acquisition.acquire_served_ask_candidate`. +This seam owns handle resolution only; it duplicates no artifact serving policy. + +**Trust boundary.** A handle is trusted plumbing state, never user text: it must +be constructed by a reviewed production path (a future producer→serving plumbing +slice), not parsed out of a chat turn. The Q1-D artifact carries no turn/case +provenance fields today, so the handle is *content-addressed*, not *turn-stamped* +— turn-addressability is exactly the explicitness of carrying the handle into the +turn. If a future slice adds turn provenance to the artifact, resolution should +check it here. + +**Forbidden moves this module never makes:** call +``generate.contemplation.pass_manager.contemplate``; import or call +``core.epistemic_questions.render`` / ``render_question`` / +``deliver_ask``; construct question prose; scan ``teaching/questions`` or any +other sink (no glob/iterdir/listdir — the only filesystem access is reading the +single explicitly-named artifact, and only after the default-dark +``ask_serving_enabled`` gate passes). +""" + +from __future__ import annotations + +import hashlib +import json +import re +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from core.epistemic_disclosure.ask_acquisition import ( + AskAcquisitionDecision, + acquire_served_ask_candidate, +) +from core.epistemic_questions.serving_gate import ask_serving_enabled + +#: The producer's content address is a sha256 hex digest — nothing else is a +#: structurally valid handle hash. +_CONTENT_HASH_RE = re.compile(r"^[0-9a-f]{64}$") + +#: String form of ``Terminal.QUESTION_NEEDED``. Deliberately a literal: importing +#: ``generate.contemplation`` from this serving-side seam would import the +#: off-serving pass manager package, which the boundary forbids. +#: ``evaluate_served_ask`` compares the stringified terminal, so the literal is +#: contract-equivalent. +_QUESTION_NEEDED = "QUESTION_NEEDED" + + +@dataclass(frozen=True, slots=True) +class AskArtifactHandle: + """An explicit, carried reference to one already-produced Q1-D artifact. + + ``question_path`` names the single artifact file; ``content_hash`` is the + producer's content address for it (also its filename stem). ``proposal_path`` + is carried through so the downstream adapter can enforce the + question/proposal collision rule — it is ``None`` for a pure ASK delivery + (the Q1-D producer never co-assigns the two). + """ + + question_path: str + content_hash: str + proposal_path: str | None = None + + +@dataclass(frozen=True, slots=True) +class ResolvedAskCandidate: + """An acquisition-compatible candidate produced by handle resolution. + + Duck-types exactly the fields ``evaluate_served_ask`` reads off a + ``ContemplationResult`` (``terminal`` / ``question_path`` / + ``proposal_path``); ``terminal`` is always the ``QUESTION_NEEDED`` string. It + carries no question text — the served surface is read from the artifact by + the adapter, never constructed here. + """ + + terminal: str + question_path: str + proposal_path: str | None = None + + +@dataclass(frozen=True, slots=True) +class AskHandleResolution: + """The typed outcome of resolving one carried handle. + + ``reason`` is a stable snake_case token naming the verification step that + rejected the handle (or ``"resolved"``). ``candidate`` is populated iff + ``resolved`` is True. + """ + + resolved: bool + reason: str + candidate: ResolvedAskCandidate | None = None + + +def _rejected(reason: str) -> AskHandleResolution: + return AskHandleResolution(resolved=False, reason=reason, candidate=None) + + +def _recompute_content_hash(payload: Any) -> str | None: + """Recompute the producer's content address from the artifact body. + + Mirrors ``DeliveredQuestion.content_hash`` exactly: the digest payload is + ``f"{blocking_reason}:{slot_name}:{text}"`` with a slot-less artifact + contributing the empty string for ``slot_name`` (the producer serializes that + case as ``null``). Returns ``None`` when the fields needed to re-derive the + address are absent or mistyped — a malformed body, by construction. + """ + if not isinstance(payload, dict): + return None + blocking_reason = payload.get("blocking_reason") + question = payload.get("question") + if not isinstance(blocking_reason, str) or not isinstance(question, dict): + return None + slot_name = question.get("slot_name") + if slot_name is None: + slot_name = "" + text = question.get("text") + if not isinstance(slot_name, str) or not isinstance(text, str): + return None + digest_payload = f"{blocking_reason}:{slot_name}:{text}" + return hashlib.sha256(digest_payload.encode("utf-8")).hexdigest() + + +def resolve_served_ask_handle(config: Any, handle: Any) -> AskHandleResolution: + """Resolve an explicit carried handle into an acquisition-compatible candidate. + + Gate-first: while ``ask_serving_enabled`` is false (the default), this + function performs **no filesystem access at all** and rejects with + ``"gate_disabled"`` — the seam is side-effect free under default runtime + configuration. Past the gate, the handle is verified structurally, the named + artifact is read (one explicit path, never a scan), and the body must + re-hash to the handle's content address. Any failure rejects with a typed + reason; resolution never raises on bad input. + + The returned candidate has passed *identity* checks only. Q1-D field policy + (status/review/served/answer-binding/text/slot/collision) is still owned by + ``evaluate_served_ask`` downstream — pass the candidate through + :func:`acquire_served_ask_from_handle` or + ``acquire_served_ask_candidate(..., contemplation_result=candidate)``. + """ + if not ask_serving_enabled(config): + return _rejected("gate_disabled") + if handle is None: + return _rejected("missing_handle") + + question_path_value = getattr(handle, "question_path", None) + content_hash_value = getattr(handle, "content_hash", None) + proposal_path_value = getattr(handle, "proposal_path", None) + + if not isinstance(question_path_value, str) or not question_path_value.strip(): + return _rejected("malformed_handle") + if ( + not isinstance(content_hash_value, str) + or _CONTENT_HASH_RE.fullmatch(content_hash_value) is None + ): + return _rejected("malformed_handle") + if proposal_path_value is not None and not isinstance(proposal_path_value, str): + return _rejected("malformed_handle") + + question_path = Path(question_path_value) + if question_path.name != f"{content_hash_value}.json": + # Not the producer's content-addressed filename for this hash — the + # handle does not name a producer-written artifact identity. + return _rejected("handle_address_mismatch") + if proposal_path_value is not None and str(question_path) == str(proposal_path_value): + return _rejected("path_collision") + + if not question_path.is_file(): + return _rejected("missing_artifact") + try: + payload = json.loads(question_path.read_text(encoding="utf-8")) + except (OSError, json.JSONDecodeError): + return _rejected("malformed_artifact") + + recomputed = _recompute_content_hash(payload) + if recomputed is None: + return _rejected("malformed_artifact") + if recomputed != content_hash_value: + # The file at the named path is not the artifact the handle was issued + # for — stale, replaced, or tampered. Fail closed. + return _rejected("content_hash_mismatch") + + return AskHandleResolution( + resolved=True, + reason="resolved", + candidate=ResolvedAskCandidate( + terminal=_QUESTION_NEEDED, + question_path=str(question_path), + proposal_path=proposal_path_value, + ), + ) + + +def acquire_served_ask_from_handle( + config: Any, + *, + handle: Any, + fallback_surface: str, +) -> AskAcquisitionDecision: + """Resolve a carried handle and pass the candidate through the existing stack. + + The lawful provider boundary in one call: an unresolved handle (gate dark, + structural failure, missing/stale artifact) yields the standing fallback + decision from ``acquire_served_ask_candidate`` with no candidate — the + fallback surface is returned unchanged. A resolved candidate is handed to the + existing acquisition seam, which delegates all Q1-D artifact policy to + ``evaluate_served_ask``. No serving rule is re-implemented here. + """ + resolution = resolve_served_ask_handle(config, handle) + return acquire_served_ask_candidate( + config, + fallback_surface=fallback_surface, + contemplation_result=resolution.candidate, + ) + + +__all__ = [ + "AskArtifactHandle", + "AskHandleResolution", + "ResolvedAskCandidate", + "acquire_served_ask_from_handle", + "resolve_served_ask_handle", +] diff --git a/tests/test_ask_handle.py b/tests/test_ask_handle.py new file mode 100644 index 00000000..346a89ec --- /dev/null +++ b/tests/test_ask_handle.py @@ -0,0 +1,397 @@ +"""Focused tests for the carried-handle ASK acquisition seam. + +These tests intentionally avoid ``chat.runtime`` and never call the off-serving +producer (``pass_manager.contemplate`` / ``deliver_ask`` / ``render_question``). +Artifacts are hand-written in the exact producer shape (the same payload +convention as ``tests/test_ask_serving_integration.py``) with the producer's +content-address recipe applied directly — the seam must resolve a *pre-existing* +artifact, so the tests construct the pre-existing state, not the producer call. +""" + +from __future__ import annotations + +import ast +import dataclasses +import hashlib +import json +from pathlib import Path + +from core.config import RuntimeConfig +from core.epistemic_disclosure.ask_handle import ( + AskArtifactHandle, + AskHandleResolution, + acquire_served_ask_from_handle, + resolve_served_ask_handle, +) +from core.epistemic_disclosure.disposition import ServedDisposition + +_SEAM_PATH = ( + Path(__file__).resolve().parents[1] / "core" / "epistemic_disclosure" / "ask_handle.py" +) + + +def _valid_payload(text: str = "How many crates are left?") -> dict: + return { + "status": "question_only", + "blocking_reason": "missing_total_count", + "owner_organ": "r2_constraint", + "question": { + "text": text, + "reason": "missing_total_count", + "slot_name": "total_count", + "expected_unit_or_type": "count_int", + "binding_target": "collective_unit_total", + }, + "answer_binding": None, + "requires_review": True, + "served": False, + } + + +def _content_hash(payload: dict) -> str: + """The producer's content address: sha256 over blocking_reason:slot_name:text.""" + slot_name = payload["question"].get("slot_name") or "" + digest = f"{payload['blocking_reason']}:{slot_name}:{payload['question']['text']}" + return hashlib.sha256(digest.encode("utf-8")).hexdigest() + + +def _write_addressed_artifact(root: Path, payload: dict) -> tuple[Path, str]: + """Write *payload* at its producer content-addressed path; return (path, hash).""" + digest = _content_hash(payload) + path = root / f"{digest}.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload, indent=2, sort_keys=True), encoding="utf-8") + return path, digest + + +def _handle(path: Path, digest: str, proposal_path: Path | None = None) -> AskArtifactHandle: + return AskArtifactHandle( + question_path=str(path), + content_hash=digest, + proposal_path=str(proposal_path) if proposal_path is not None else None, + ) + + +# --- gate-first behavior ----------------------------------------------------- + + +def test_gate_disabled_rejects_without_filesystem_access(tmp_path: Path, monkeypatch) -> None: + payload = _valid_payload() + path, digest = _write_addressed_artifact(tmp_path / "questions", payload) + + def boom(*args, **kwargs): # pragma: no cover - must not run + raise AssertionError("the seam must not touch the filesystem while the gate is dark") + + monkeypatch.setattr(Path, "is_file", boom) + monkeypatch.setattr(Path, "read_text", boom) + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=False), _handle(path, digest) + ) + + assert isinstance(resolution, AskHandleResolution) + assert resolution.resolved is False + assert resolution.reason == "gate_disabled" + assert resolution.candidate is None + + +def test_gate_disabled_fallback_surface_is_unchanged(tmp_path: Path) -> None: + payload = _valid_payload() + path, digest = _write_addressed_artifact(tmp_path / "questions", payload) + + acquired = acquire_served_ask_from_handle( + RuntimeConfig(ask_serving_enabled=False), + handle=_handle(path, digest), + fallback_surface="I don't know — insufficient grounding for that yet.", + ) + + assert acquired.acquired is False + assert acquired.provider_called is False + assert acquired.decision.served is False + assert acquired.decision.surface == "I don't know — insufficient grounding for that yet." + assert acquired.decision.disposition is ServedDisposition.REFUSE + + +def test_default_config_is_dark(tmp_path: Path) -> None: + payload = _valid_payload() + path, digest = _write_addressed_artifact(tmp_path / "questions", payload) + + resolution = resolve_served_ask_handle(RuntimeConfig(), _handle(path, digest)) + + assert resolution.resolved is False + assert resolution.reason == "gate_disabled" + + +# --- valid resolution through the existing serving stack --------------------- + + +def test_valid_handle_resolves_to_acquisition_compatible_candidate(tmp_path: Path) -> None: + payload = _valid_payload("How many crates are left?") + path, digest = _write_addressed_artifact(tmp_path / "questions", payload) + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=True), _handle(path, digest) + ) + + assert resolution.resolved is True + assert resolution.reason == "resolved" + assert resolution.candidate is not None + assert resolution.candidate.terminal == "QUESTION_NEEDED" + assert resolution.candidate.question_path == str(path) + assert resolution.candidate.proposal_path is None + + +def test_valid_handle_serves_through_existing_stack(tmp_path: Path) -> None: + payload = _valid_payload("How many crates are left?") + path, digest = _write_addressed_artifact(tmp_path / "questions", payload) + + acquired = acquire_served_ask_from_handle( + RuntimeConfig(ask_serving_enabled=True), + handle=_handle(path, digest), + fallback_surface="fallback", + ) + + assert acquired.acquired is True + assert acquired.decision.served is True + assert acquired.decision.terminal == "QUESTION_NEEDED" + assert acquired.decision.surface == "How many crates are left?" + assert acquired.decision.disposition is ServedDisposition.ASK + + +def test_resolved_candidate_carries_no_question_prose(tmp_path: Path) -> None: + payload = _valid_payload("How many crates are left?") + path, digest = _write_addressed_artifact(tmp_path / "questions", payload) + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=True), _handle(path, digest) + ) + + assert resolution.candidate is not None + candidate_fields = dataclasses.asdict(resolution.candidate) + assert "How many crates are left?" not in { + v for v in candidate_fields.values() if isinstance(v, str) + } + + +# --- fail-closed rejections --------------------------------------------------- + + +def test_handle_to_proposal_artifact_fails_closed(tmp_path: Path) -> None: + proposal_payload = { + "status": "proposal_only", + "family": "missing_category_pair", + "requires_review": True, + "mounted": False, + } + path = tmp_path / "proposals" / "p.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(proposal_payload), encoding="utf-8") + digest = hashlib.sha256(b"not-a-question-address").hexdigest() + addressed = path.parent / f"{digest}.json" + addressed.write_text(json.dumps(proposal_payload), encoding="utf-8") + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=True), _handle(addressed, digest) + ) + + # A proposal artifact has no Q1-D content-address fields → malformed body. + assert resolution.resolved is False + assert resolution.reason == "malformed_artifact" + + acquired = acquire_served_ask_from_handle( + RuntimeConfig(ask_serving_enabled=True), + handle=_handle(addressed, digest), + fallback_surface="fallback", + ) + assert acquired.decision.served is False + assert acquired.decision.surface == "fallback" + + +def test_missing_artifact_fails_closed(tmp_path: Path) -> None: + digest = _content_hash(_valid_payload()) + missing = tmp_path / "questions" / f"{digest}.json" + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=True), _handle(missing, digest) + ) + + assert resolution.resolved is False + assert resolution.reason == "missing_artifact" + + +def test_malformed_json_fails_closed(tmp_path: Path) -> None: + digest = _content_hash(_valid_payload()) + path = tmp_path / "questions" / f"{digest}.json" + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("{bad json", encoding="utf-8") + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=True), _handle(path, digest) + ) + + assert resolution.resolved is False + assert resolution.reason == "malformed_artifact" + + +def test_valid_json_with_invalid_q1d_fields_fails_closed_via_adapter(tmp_path: Path) -> None: + """Identity checks pass (the hash covers blocking_reason/slot/text only) but + the adapter's Q1-D policy still rejects — proving delegation, not duplication.""" + for corruption in ("served", "requires_review"): + payload = _valid_payload() + payload[corruption] = not payload[corruption] + path, digest = _write_addressed_artifact(tmp_path / f"questions_{corruption}", payload) + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=True), _handle(path, digest) + ) + assert resolution.resolved is True # identity intact + + acquired = acquire_served_ask_from_handle( + RuntimeConfig(ask_serving_enabled=True), + handle=_handle(path, digest), + fallback_surface="fallback", + ) + assert acquired.decision.served is False + assert acquired.decision.surface == "fallback" + + +def test_question_path_equal_to_proposal_path_fails_closed(tmp_path: Path) -> None: + payload = _valid_payload() + path, digest = _write_addressed_artifact(tmp_path / "questions", payload) + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=True), _handle(path, digest, proposal_path=path) + ) + + assert resolution.resolved is False + assert resolution.reason == "path_collision" + + +def test_stale_or_replaced_artifact_fails_closed(tmp_path: Path) -> None: + """The file at the handle's path no longer re-hashes to the handle's address.""" + original = _valid_payload("How many crates are left?") + path, digest = _write_addressed_artifact(tmp_path / "questions", original) + + replaced = _valid_payload("How many boxes were shipped?") + path.write_text(json.dumps(replaced, indent=2, sort_keys=True), encoding="utf-8") + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=True), _handle(path, digest) + ) + + assert resolution.resolved is False + assert resolution.reason == "content_hash_mismatch" + + +def test_handle_filename_must_be_content_addressed(tmp_path: Path) -> None: + payload = _valid_payload() + digest = _content_hash(payload) + path = tmp_path / "questions" / "q.json" # not the producer's addressed name + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(json.dumps(payload), encoding="utf-8") + + resolution = resolve_served_ask_handle( + RuntimeConfig(ask_serving_enabled=True), _handle(path, digest) + ) + + assert resolution.resolved is False + assert resolution.reason == "handle_address_mismatch" + + +def test_structurally_invalid_handles_fail_closed(tmp_path: Path) -> None: + config = RuntimeConfig(ask_serving_enabled=True) + payload = _valid_payload() + path, digest = _write_addressed_artifact(tmp_path / "questions", payload) + + assert resolve_served_ask_handle(config, None).reason == "missing_handle" + assert ( + resolve_served_ask_handle(config, AskArtifactHandle("", digest)).reason + == "malformed_handle" + ) + assert ( + resolve_served_ask_handle(config, AskArtifactHandle(" ", digest)).reason + == "malformed_handle" + ) + assert ( + resolve_served_ask_handle( + config, AskArtifactHandle(str(path), "not-a-sha256") + ).reason + == "malformed_handle" + ) + assert ( + resolve_served_ask_handle( + config, AskArtifactHandle(str(path), digest.upper()) + ).reason + == "malformed_handle" + ) + + +# --- no-scan / boundary proofs ------------------------------------------------- + + +def test_seam_cannot_discover_artifacts_without_exact_handle(tmp_path: Path) -> None: + """A valid artifact sitting in the sink is unreachable without its exact + content-addressed handle — there is no discovery path.""" + payload = _valid_payload() + _write_addressed_artifact(tmp_path / "questions", payload) + + wrong_digest = hashlib.sha256(b"some-other-question").hexdigest() + wrong = AskArtifactHandle( + question_path=str(tmp_path / "questions" / f"{wrong_digest}.json"), + content_hash=wrong_digest, + ) + + resolution = resolve_served_ask_handle(RuntimeConfig(ask_serving_enabled=True), wrong) + + assert resolution.resolved is False + assert resolution.reason == "missing_artifact" + + +def test_seam_module_contains_no_sink_scanning_primitives() -> None: + tree = ast.parse(_SEAM_PATH.read_text(encoding="utf-8"), filename=str(_SEAM_PATH)) + forbidden_attrs = {"glob", "rglob", "iterdir", "walk", "scandir", "listdir"} + + for node in ast.walk(tree): + if isinstance(node, ast.Attribute): + assert node.attr not in forbidden_attrs, f"sink scan primitive: {node.attr}" + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + assert node.func.id not in forbidden_attrs + + +def test_seam_module_does_not_import_runtime_producer_or_renderer() -> None: + source = _SEAM_PATH.read_text(encoding="utf-8") + tree = ast.parse(source, filename=str(_SEAM_PATH)) + + forbidden_modules = ( + "chat", + "chat.runtime", + "chat.ask_runtime", + "generate.contemplation", + "generate.contemplation.pass_manager", + "generate.contemplation.findings", + "core.epistemic_questions.render", + "core.epistemic_questions.delivery", + ) + forbidden_calls = {"contemplate", "deliver_ask", "render_question", "emit_question"} + + for node in ast.walk(tree): + if isinstance(node, ast.ImportFrom) and node.module: + for forbidden in forbidden_modules: + assert node.module != forbidden + assert not node.module.startswith(forbidden + ".") + if isinstance(node, ast.Import): + for alias in node.names: + for forbidden in forbidden_modules: + assert alias.name != forbidden + assert not alias.name.startswith(forbidden + ".") + if isinstance(node, ast.Call): + if isinstance(node.func, ast.Name): + assert node.func.id not in forbidden_calls + if isinstance(node.func, ast.Attribute): + assert node.func.attr not in forbidden_calls + + +def test_seam_module_constructs_no_question_prose() -> None: + source = _SEAM_PATH.read_text(encoding="utf-8") + for forbidden_template in ("How many", "What ", "Which ", "Please provide"): + assert forbidden_template not in source