From 14285b5f21283f6979ff66e075eb1973cfcd01a2 Mon Sep 17 00:00:00 2001 From: Shay Date: Tue, 9 Jun 2026 13:00:11 -0700 Subject: [PATCH] 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