Merge pull request #678 from AssetOverflow/add-ask-acquisition-seam

feat(ask): add serving-safe ASK acquisition seam

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

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

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