From 16979f48d07aa58675f0d30e86fb4bbeeb5f5220 Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 9 Jun 2026 12:58:31 -0700 Subject: [PATCH 1/3] feat(ask): add served ASK acquisition seam --- core/epistemic_disclosure/ask_acquisition.py | 121 +++++++++++++++++++ 1 file changed, 121 insertions(+) create mode 100644 core/epistemic_disclosure/ask_acquisition.py diff --git a/core/epistemic_disclosure/ask_acquisition.py b/core/epistemic_disclosure/ask_acquisition.py new file mode 100644 index 00000000..ebc9f7a6 --- /dev/null +++ b/core/epistemic_disclosure/ask_acquisition.py @@ -0,0 +1,121 @@ +"""Serving-safe acquisition seam for Stage 2 ASK candidates. + +This module owns only the narrow boundary between a serving caller and the +already-shipped ASK artifact adapter: + +1. honor the default-dark ``ask_serving_enabled`` gate before any provider call; +2. acquire, or accept, a candidate contemplation result; +3. delegate all artifact validation and served-surface decisioning to + :func:`core.epistemic_disclosure.ask_serving.evaluate_served_ask`. + +It does not render question prose, does not import the Q1-C renderer, does not +call ``generate.contemplation.pass_manager`` directly, and does not mutate +runtime/telemetry schemas. A future runtime slice can supply the provider once +the legal turn-boundary is explicit. +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass +from typing import Any + +from core.epistemic_disclosure.ask_serving import ServedAskDecision, evaluate_served_ask +from core.epistemic_disclosure.disposition import ServedDisposition +from core.epistemic_questions.serving_gate import ask_serving_enabled + +ContemplationProvider = Callable[[], Any | None] + + +@dataclass(frozen=True, slots=True) +class AskAcquisitionDecision: + """Result of the serving-safe ASK acquisition seam.""" + + acquired: bool + provider_called: bool + decision: ServedAskDecision + + +_NO_PROGRESS_FALLBACK = ServedAskDecision( + served=False, + terminal="NO_PROGRESS", + surface="", + disposition=ServedDisposition.REFUSE, +) + + +def _fallback(surface: str) -> ServedAskDecision: + return ServedAskDecision( + served=False, + terminal=_NO_PROGRESS_FALLBACK.terminal, + surface=surface, + disposition=_NO_PROGRESS_FALLBACK.disposition, + ) + + +def acquire_served_ask_candidate( + config: Any, + *, + fallback_surface: str, + contemplation_result: Any | None = None, + provider: ContemplationProvider | None = None, +) -> AskAcquisitionDecision: + """Acquire and evaluate a candidate served-ASK artifact. + + The provider is never called while ``ask_serving_enabled`` is false. That + gate-first rule keeps this seam side-effect free under default runtime + configuration and prevents accidental contemplation work from changing + normal serving behavior. + + If a caller already has a ``ContemplationResult``, pass it via + ``contemplation_result``. Otherwise, pass a provider that can return one. + The result is delegated to ``evaluate_served_ask``; this seam duplicates no + Q1-D artifact validation and constructs no user-facing question text. + """ + + if not ask_serving_enabled(config): + # Preserve any already-known terminal/disposition without causing a + # provider side effect. ``evaluate_served_ask`` itself is gate-aware and + # will fail closed without reading artifacts when the gate is disabled. + if contemplation_result is not None: + decision = evaluate_served_ask(config, contemplation_result, fallback_surface) + else: + decision = _fallback(fallback_surface) + return AskAcquisitionDecision( + acquired=contemplation_result is not None, + provider_called=False, + decision=decision, + ) + + provider_called = False + candidate = contemplation_result + if candidate is None and provider is not None: + provider_called = True + try: + candidate = provider() + except Exception: + return AskAcquisitionDecision( + acquired=False, + provider_called=True, + decision=_fallback(fallback_surface), + ) + + if candidate is None: + return AskAcquisitionDecision( + acquired=False, + provider_called=provider_called, + decision=_fallback(fallback_surface), + ) + + return AskAcquisitionDecision( + acquired=True, + provider_called=provider_called, + decision=evaluate_served_ask(config, candidate, fallback_surface), + ) + + +__all__ = [ + "AskAcquisitionDecision", + "ContemplationProvider", + "acquire_served_ask_candidate", +] From 4403a2104467592f0df5bb931752ba31acb9f527 Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 9 Jun 2026 12:59:15 -0700 Subject: [PATCH 2/3] feat(ask): export served ASK acquisition seam --- core/epistemic_disclosure/__init__.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/core/epistemic_disclosure/__init__.py b/core/epistemic_disclosure/__init__.py index b612c998..142332c5 100644 --- a/core/epistemic_disclosure/__init__.py +++ b/core/epistemic_disclosure/__init__.py @@ -22,6 +22,9 @@ Shipped so far (all off-serving — nothing here imports ``generate.derivation`` * :mod:`~core.epistemic_disclosure.ask_serving` — a narrow Q1-D served-ASK artifact adapter. It validates already-rendered question artifacts and returns a typed decision; it does not render prose and does not acquire runtime contemplation. + * :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.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 @@ -30,6 +33,11 @@ Shipped so far (all off-serving — nothing here imports ``generate.derivation`` from __future__ import annotations +from core.epistemic_disclosure.ask_acquisition import ( + AskAcquisitionDecision, + ContemplationProvider, + acquire_served_ask_candidate, +) from core.epistemic_disclosure.ask_serving import ( ServedAskDecision, evaluate_served_ask, @@ -66,6 +74,8 @@ __all__ = [ "DEFAULT_DISCLOSURE_CLAIM", "Q1B_ASK_CARVE_OUT", "VERIFICATION_OBLIGATION", + "AskAcquisitionDecision", + "ContemplationProvider", "DisclosureClaim", "LimitationAssessment", "LimitationKind", @@ -77,6 +87,7 @@ __all__ = [ "VerificationProof", "VerificationResult", "VerificationVerdict", + "acquire_served_ask_candidate", "assess_from_attempt", "assess_from_family", "choose_served_disposition", From 14285b5f21283f6979ff66e075eb1973cfcd01a2 Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 9 Jun 2026 13:00:11 -0700 Subject: [PATCH 3/3] test(ask): cover served ASK acquisition seam --- tests/test_ask_acquisition.py | 283 ++++++++++++++++++++++++++++++++++ 1 file changed, 283 insertions(+) create mode 100644 tests/test_ask_acquisition.py diff --git a/tests/test_ask_acquisition.py b/tests/test_ask_acquisition.py new file mode 100644 index 00000000..0f252550 --- /dev/null +++ b/tests/test_ask_acquisition.py @@ -0,0 +1,283 @@ +"""Tests for the Stage 2 served ASK acquisition seam. + +The seam is provider-gated and delegates artifact validation to +``evaluate_served_ask``. It intentionally avoids ``chat.runtime`` and does not +call ``generate.contemplation.pass_manager`` directly. +""" + +from __future__ import annotations + +import ast +import json +from pathlib import Path + +from core.config import RuntimeConfig +from core.epistemic_disclosure.ask_acquisition import acquire_served_ask_candidate +from core.epistemic_disclosure.disposition import ServedDisposition + + +class DummyTerminal: + def __init__(self, value: str) -> None: + self.value = value + + +class DummyContemplationResult: + def __init__( + self, + terminal: str, + *, + question_path: str | None = None, + proposal_path: str | None = None, + ) -> None: + self.terminal = DummyTerminal(terminal) + self.question_path = question_path + self.proposal_path = proposal_path + + +def _write_valid_question(path: Path, text: str = "How many crates are left?") -> None: + path.parent.mkdir(parents=True, exist_ok=True) + payload = { + "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, + } + path.write_text(json.dumps(payload), encoding="utf-8") + + +def _question_result(q_path: Path, p_path: Path | None = None) -> DummyContemplationResult: + return DummyContemplationResult( + "QUESTION_NEEDED", + question_path=str(q_path), + proposal_path=str(p_path) if p_path is not None else None, + ) + + +def test_gate_disabled_does_not_call_provider() -> None: + calls = [] + + def provider(): + calls.append("called") + raise AssertionError("provider must not run while ASK serving gate is disabled") + + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=False), + fallback_surface="fallback", + provider=provider, + ) + + assert calls == [] + assert acquired.provider_called is False + assert acquired.acquired is False + assert acquired.decision.served is False + assert acquired.decision.surface == "fallback" + assert acquired.decision.disposition is ServedDisposition.REFUSE + + +def test_gate_disabled_with_existing_result_fails_closed_without_provider(tmp_path: Path) -> None: + q_path = tmp_path / "questions" / "q.json" + _write_valid_question(q_path) + + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=False), + fallback_surface="fallback", + contemplation_result=_question_result(q_path), + provider=lambda: (_ for _ in ()).throw(AssertionError("provider must not run")), + ) + + assert acquired.provider_called is False + assert acquired.acquired is True + assert acquired.decision.served is False + assert acquired.decision.terminal == "QUESTION_NEEDED" + assert acquired.decision.surface == "fallback" + + +def test_gate_enabled_provider_valid_question_serves_via_adapter(tmp_path: Path) -> None: + q_path = tmp_path / "questions" / "q.json" + p_path = tmp_path / "proposals" / "p.json" + _write_valid_question(q_path, "How many crates are left?") + calls = [] + + def provider(): + calls.append("called") + return _question_result(q_path, p_path) + + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=True), + fallback_surface="fallback", + provider=provider, + ) + + assert calls == ["called"] + assert acquired.provider_called is True + 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_gate_enabled_existing_result_skips_provider_and_serves(tmp_path: Path) -> None: + q_path = tmp_path / "questions" / "q.json" + _write_valid_question(q_path) + + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=True), + fallback_surface="fallback", + contemplation_result=_question_result(q_path), + provider=lambda: (_ for _ in ()).throw(AssertionError("provider should not run")), + ) + + assert acquired.provider_called is False + assert acquired.acquired is True + assert acquired.decision.served is True + + +def test_provider_none_fails_closed() -> None: + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=True), + fallback_surface="fallback", + provider=lambda: None, + ) + + assert acquired.provider_called is True + assert acquired.acquired is False + assert acquired.decision.served is False + assert acquired.decision.surface == "fallback" + + +def test_provider_exception_fails_closed() -> None: + def provider(): + raise RuntimeError("candidate unavailable") + + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=True), + fallback_surface="fallback", + provider=provider, + ) + + assert acquired.provider_called is True + assert acquired.acquired is False + assert acquired.decision.served is False + assert acquired.decision.surface == "fallback" + + +def test_proposal_only_without_question_path_does_not_serve(tmp_path: Path) -> None: + result = DummyContemplationResult( + "PROPOSAL_EMITTED", + question_path=None, + proposal_path=str(tmp_path / "proposals" / "p.json"), + ) + + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=True), + fallback_surface="fallback", + contemplation_result=result, + ) + + assert acquired.acquired is True + assert acquired.decision.served is False + assert acquired.decision.terminal == "PROPOSAL_EMITTED" + assert acquired.decision.disposition is ServedDisposition.PROPOSE + assert acquired.decision.surface == "fallback" + + +def test_question_path_equal_to_proposal_path_fails_closed(tmp_path: Path) -> None: + q_path = tmp_path / "shared" / "artifact.json" + _write_valid_question(q_path) + + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=True), + fallback_surface="fallback", + contemplation_result=_question_result(q_path, q_path), + ) + + assert acquired.decision.served is False + assert acquired.decision.surface == "fallback" + + +def test_malformed_artifact_fails_closed(tmp_path: Path) -> None: + q_path = tmp_path / "questions" / "bad.json" + q_path.parent.mkdir(parents=True, exist_ok=True) + q_path.write_text("{bad json", encoding="utf-8") + + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=True), + fallback_surface="fallback", + contemplation_result=_question_result(q_path), + ) + + assert acquired.decision.served is False + assert acquired.decision.surface == "fallback" + + +def test_invalid_question_artifact_fails_closed(tmp_path: Path) -> None: + q_path = tmp_path / "questions" / "invalid.json" + q_path.parent.mkdir(parents=True, exist_ok=True) + q_path.write_text( + json.dumps( + { + "status": "proposal_only", + "question": {"text": "How many crates are left?", "slot_name": "total_count"}, + "requires_review": True, + "served": False, + } + ), + encoding="utf-8", + ) + + acquired = acquire_served_ask_candidate( + RuntimeConfig(ask_serving_enabled=True), + fallback_surface="fallback", + contemplation_result=_question_result(q_path), + ) + + assert acquired.decision.served is False + assert acquired.decision.surface == "fallback" + + +def test_acquisition_module_does_not_import_renderer_or_pass_manager() -> None: + path = Path(__file__).resolve().parents[1] / "core" / "epistemic_disclosure" / "ask_acquisition.py" + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + + forbidden_modules = ( + "core.epistemic_questions.render", + "generate.contemplation.pass_manager", + "chat", + "chat.runtime", + ) + + 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) and isinstance(node.func, ast.Name): + assert node.func.id != "render_question" + + +def test_runtime_and_telemetry_schemas_are_not_changed_by_this_slice() -> None: + runtime_path = Path(__file__).resolve().parents[1] / "chat" / "runtime.py" + telemetry_path = Path(__file__).resolve().parents[1] / "chat" / "telemetry.py" + identity_path = Path(__file__).resolve().parents[1] / "core" / "physics" / "identity.py" + + assert "contemplation_result" not in runtime_path.read_text(encoding="utf-8") + assert "disposition" not in telemetry_path.read_text(encoding="utf-8") + + identity_source = identity_path.read_text(encoding="utf-8") + turn_event_section = identity_source[identity_source.index("class TurnEvent") :] + assert "disposition" not in turn_event_section