core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(ask): add carried-handle acquisition seam for served ASK

Browse source 
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).
This commit is contained in:
Shay 2026-06-10 10:53:54 -07:00
parent 7d6b7918aa
commit 41664ce60d
3 changed files with 665 additions and 0 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

17
core/epistemic_disclosure/__init__.py
View file

@ -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",
]

251
core/epistemic_disclosure/ask_handle.py Normal file
View file

@ -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",
]

397
tests/test_ask_handle.py Normal file
View file

@ -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