diff --git a/generate/contemplation/pass_manager.py b/generate/contemplation/pass_manager.py index 9afd0d6b..7f5c9a9b 100644 --- a/generate/contemplation/pass_manager.py +++ b/generate/contemplation/pass_manager.py @@ -28,7 +28,7 @@ from __future__ import annotations from dataclasses import dataclass from pathlib import Path -from typing import Any +from typing import Any, TYPE_CHECKING from core.comprehension_attempt import ( ComprehensionAttempt, @@ -49,6 +49,10 @@ from generate.rate_comprehension.solver import solve_rate from generate.contemplation.findings import Finding, Terminal from generate.meaning_graph.reader import Refusal +if TYPE_CHECKING: + from core.epistemic_disclosure.limitation import LimitationAssessment + from core.epistemic_questions.delivery import DeliveryOutcome + #: Substantive boundaries that are *recognized-but-unsupported* capabilities (not hard errors). _UNSUPPORTED_FAMILIES = frozenset( { @@ -77,13 +81,71 @@ class ContemplationResult: message: str | None = None +def _delivery_outcome_for_limitation(assessment: LimitationAssessment) -> DeliveryOutcome: + """Helper to delegate to deliver_ask, pure and testable.""" + from core.epistemic_questions.delivery import deliver_ask + return deliver_ask(assessment) + + +def _handle_ask_delivery( + assessment: LimitationAssessment, + family_name: str, + findings: list[Finding], + attempts: tuple[ComprehensionAttempt, ...], + text: str, + proposal_root: Path | None, + question_root: Path | None, + exercise_ask: bool, + selected_organ: str | None = None, +) -> ContemplationResult: + outcome = _delivery_outcome_for_limitation(assessment) + if outcome.terminal == Terminal.QUESTION_NEEDED: + assert outcome.question is not None + import json + from core.epistemic_questions.delivery import question_path + + path = question_path(outcome.question, question_root) + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text( + json.dumps(outcome.question.to_json_dict(), indent=2, sort_keys=True), + encoding="utf-8", + ) + + findings.append(Finding("ask", f"emitted question-only {assessment.blocking_reason}")) + findings.append(Finding("terminal", Terminal.QUESTION_NEEDED.value)) + return ContemplationResult( + Terminal.QUESTION_NEEDED, tuple(findings), attempts, + selected_organ=selected_organ, family=family_name, + proposal_path=str(path), + ) + else: + findings.append(Finding("ask", f"unrenderable ask: {outcome.fallback_reason}")) + findings.append(Finding("terminal", outcome.terminal.value)) + if outcome.terminal == Terminal.PROPOSAL_EMITTED: + from core.comprehension_attempt.failure_family import family_by_name + family_obj = family_by_name(family_name) + if family_obj is not None: + path = emit_proposal(text, family_obj, attempts, root=proposal_root) + return ContemplationResult( + Terminal.PROPOSAL_EMITTED, tuple(findings), attempts, + selected_organ=selected_organ, family=family_name, + proposal_path=str(path) if path else None, + ) + return ContemplationResult( + outcome.terminal, tuple(findings), attempts, + selected_organ=selected_organ, family=family_name, + ) + + def contemplate( text: str, *, options: dict[str, Any] | None = None, answer_key: str | None = None, proposal_root: Path | None = None, + question_root: Path | None = None, case_id: str | None = None, + exercise_ask: bool = False, ) -> ContemplationResult: """Run one bounded contemplation pass over *text*.""" findings: list[Finding] = [] @@ -111,11 +173,11 @@ def contemplate( if route.status == "routed": assert route.selected is not None if route.selected.organ == "r2_constraints": - return _solve_and_verify_r2(text, options, answer_key, findings, attempts) + return _solve_and_verify_r2(text, options, answer_key, findings, attempts, proposal_root, question_root, exercise_ask) if route.selected.organ == "r3_rate": - return _solve_and_verify_r3(text, options, answer_key, findings, attempts) + return _solve_and_verify_r3(text, options, answer_key, findings, attempts, proposal_root, question_root, exercise_ask) if route.selected.organ == "r4_combined_rate": - return _solve_and_verify_cmb(text, options, answer_key, findings, attempts) + return _solve_and_verify_cmb(text, options, answer_key, findings, attempts, proposal_root, question_root, exercise_ask) findings.append(Finding("solve", "r1 admissible setup (numeric answer is the eval lane in v0)")) findings.append(Finding("terminal", Terminal.SOLVED_VERIFIED.value)) return ContemplationResult( @@ -123,7 +185,7 @@ def contemplate( ) # route.status == "all_refused" - return _classify_all_refused(text, attempts, findings, proposal_root) + return _classify_all_refused(text, attempts, findings, proposal_root, question_root, exercise_ask) def _solve_and_verify_r2( @@ -132,12 +194,26 @@ def _solve_and_verify_r2( answer_key: str | None, findings: list[Finding], attempts: tuple[ComprehensionAttempt, ...], + proposal_root: Path | None, + question_root: Path | None, + exercise_ask: bool, ) -> ContemplationResult: problem = read_constraint_problem(text) assert not isinstance(problem, Refusal) # routed => the reader admitted a setup value = answer_constraint_problem(problem) if isinstance(value, Refusal): findings.append(Finding("solve", f"solver refused: {value.reason}")) + from core.comprehension_attempt.failure_family import family_by_name + from core.epistemic_disclosure.limitation import assess_from_family + family_obj = family_by_name(value.reason) + if family_obj is not None: + assessment = assess_from_family(family_obj) + if assessment.resolution_action == "ask_question": + if exercise_ask: + return _handle_ask_delivery( + assessment, family_obj.name, findings, attempts, text, proposal_root, question_root, exercise_ask, + selected_organ="r2_constraints" + ) findings.append(Finding("terminal", Terminal.REFUSED_KNOWN_BOUNDARY.value)) return ContemplationResult( Terminal.REFUSED_KNOWN_BOUNDARY, tuple(findings), attempts, @@ -175,12 +251,26 @@ def _solve_and_verify_r3( answer_key: str | None, findings: list[Finding], attempts: tuple[ComprehensionAttempt, ...], + proposal_root: Path | None, + question_root: Path | None, + exercise_ask: bool, ) -> ContemplationResult: problem = read_rate_problem(text) assert not isinstance(problem, Refusal) # routed => the reader admitted a setup value = solve_rate(problem) if isinstance(value, Refusal): findings.append(Finding("solve", f"solver refused: {value.reason}")) + from core.comprehension_attempt.failure_family import family_by_name + from core.epistemic_disclosure.limitation import assess_from_family + family_obj = family_by_name(value.reason) + if family_obj is not None: + assessment = assess_from_family(family_obj) + if assessment.resolution_action == "ask_question": + if exercise_ask: + return _handle_ask_delivery( + assessment, family_obj.name, findings, attempts, text, proposal_root, question_root, exercise_ask, + selected_organ="r3_rate" + ) findings.append(Finding("terminal", Terminal.REFUSED_KNOWN_BOUNDARY.value)) return ContemplationResult( Terminal.REFUSED_KNOWN_BOUNDARY, tuple(findings), attempts, @@ -218,6 +308,9 @@ def _solve_and_verify_cmb( answer_key: str | None, findings: list[Finding], attempts: tuple[ComprehensionAttempt, ...], + proposal_root: Path | None, + question_root: Path | None, + exercise_ask: bool, ) -> ContemplationResult: problem = read_combined_rate_problem(text) assert not isinstance(problem, Refusal) # routed => the reader admitted a setup @@ -227,6 +320,18 @@ def _solve_and_verify_cmb( # answerable boundary. A solver refusal is a terminal boundary, never a proposal — and the # reason is namespaced cmb_* so it resolves to the CMB solver family, not R2/R3's. findings.append(Finding("solve", f"solver refused: {value.reason}")) + from core.comprehension_attempt.failure_family import family_by_name + from core.epistemic_disclosure.limitation import assess_from_family + reason = cmb_reason(value.reason) + family_obj = family_by_name(reason) + if family_obj is not None: + assessment = assess_from_family(family_obj) + if assessment.resolution_action == "ask_question": + if exercise_ask: + return _handle_ask_delivery( + assessment, family_obj.name, findings, attempts, text, proposal_root, question_root, exercise_ask, + selected_organ="r4_combined_rate" + ) findings.append(Finding("terminal", Terminal.REFUSED_KNOWN_BOUNDARY.value)) return ContemplationResult( Terminal.REFUSED_KNOWN_BOUNDARY, tuple(findings), attempts, @@ -263,6 +368,8 @@ def _classify_all_refused( attempts: tuple[ComprehensionAttempt, ...], findings: list[Finding], proposal_root: Path | None, + question_root: Path | None, + exercise_ask: bool, ) -> ContemplationResult: # CMB-over-R3 precedence (family side): when CMB substantively recognized the text, R3's broader # partial classification is suppressed, so CMB's sharper diagnosis owns the terminal/proposal @@ -270,9 +377,10 @@ def _classify_all_refused( considered = attempts if cmb_is_authoritative(attempts): considered = tuple(a for a in attempts if a.organ != "r3_rate") + families = [(a, family_for_reason(a.refusal_reason)) for a in considered] - # Boundary-first: a substantive recognized boundary blocks any proposal. + # Boundary-first: a substantive recognized boundary blocks any proposal or ASK. for _attempt, family in families: if family is not None and family.must_remain_refused and family.name != _NOT_MY_DOMAIN: terminal = ( @@ -283,6 +391,16 @@ def _classify_all_refused( findings.append(Finding("terminal", f"{terminal.value} via {family.name}")) return ContemplationResult(terminal, tuple(findings), attempts, family=family.name) + # Check for ASK delivery only after substantive boundaries are ruled out. + for attempt in considered: + from core.epistemic_disclosure.limitation import assess_from_attempt + assessment = assess_from_attempt(attempt) + if assessment is not None and assessment.resolution_action == "ask_question": + if exercise_ask: + return _handle_ask_delivery( + assessment, assessment.blocking_reason, findings, attempts, text, proposal_root, question_root, exercise_ask + ) + # No substantive boundary: a genuine growth surface may emit a proposal-only artifact. for _attempt, family in families: if family is not None and family.proposal_allowed: diff --git a/tests/test_ask_pass_manager_delivery.py b/tests/test_ask_pass_manager_delivery.py new file mode 100644 index 00000000..5bae4312 --- /dev/null +++ b/tests/test_ask_pass_manager_delivery.py @@ -0,0 +1,293 @@ +"""Focused tests for off-serving ASK delivery integration in the contemplation pass manager (N6). + +Ensures that the pass manager seam delegates to deliver_ask, correctly handles fallback dispositions, +avoids direct rendering imports/text construction, and preserves the Q1-B carve-out. +""" + +from __future__ import annotations + +import ast +from pathlib import Path + +import pytest + +from core.epistemic_disclosure.limitation import ( + Q1B_ASK_CARVE_OUT, + LimitationAssessment, + MissingSlot, +) +from core.epistemic_state import EpistemicState +from generate.contemplation import Terminal, contemplate +from generate.contemplation.pass_manager import ( + _delivery_outcome_for_limitation, +) +from core.comprehension_attempt.failure_family import family_by_name + + +def _make_assessment( + *, + blocking_reason: str, + slots: tuple[MissingSlot, ...], + resolution_action: str = "ask_question", +) -> LimitationAssessment: + return LimitationAssessment( + limitation_kind="missing_information", + resolution_action=resolution_action, # type: ignore[arg-type] + epistemic_state=EpistemicState.UNDETERMINED, + owner_organ="r2_constraint", + blocking_reason=blocking_reason, + missing_slots=slots, + ) + + +_TOTAL_COUNT_SLOT = MissingSlot( + slot_name="total_count", + expected_unit_or_type="count_int", + binding_target="collective_unit_total", +) +_WEIGHTED_SLOT = MissingSlot( + slot_name="weighted_total", + expected_unit_or_type="measured_unit_int", + binding_target="weighted_total_value", +) + + +def test_pass_manager_uses_deliver_ask_for_renderable_ask() -> None: + assessment = _make_assessment( + blocking_reason="missing_total_count", slots=(_TOTAL_COUNT_SLOT,) + ) + outcome = _delivery_outcome_for_limitation(assessment) + assert outcome.terminal == Terminal.QUESTION_NEEDED + assert outcome.question is not None + assert outcome.fallback_reason is None + + +def test_pass_manager_unrenderable_ask_falls_back_without_question_needed() -> None: + # Multi-slot => unrenderable, falls back to standing disposition (PROPOSAL_EMITTED since missing_total_count is carve-out) + assessment = _make_assessment( + blocking_reason="missing_total_count", + slots=(_TOTAL_COUNT_SLOT, _WEIGHTED_SLOT), + ) + outcome = _delivery_outcome_for_limitation(assessment) + assert outcome.terminal == Terminal.PROPOSAL_EMITTED + assert outcome.terminal is not Terminal.QUESTION_NEEDED + assert outcome.question is None + assert outcome.fallback_reason == "multi_slot_not_supported" + + +def test_pass_manager_does_not_import_or_call_render_question_directly() -> None: + path = ( + Path(__file__).resolve().parents[1] + / "generate" + / "contemplation" + / "pass_manager.py" + ) + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + for node in ast.walk(tree): + # Ensure render_question is not imported + if isinstance(node, ast.ImportFrom): + assert node.module != "core.epistemic_questions.render" + if node.names: + for alias in node.names: + assert alias.name != "render_question" + elif isinstance(node, ast.Import): + for alias in node.names: + assert alias.name != "core.epistemic_questions.render" + + # Ensure render_question is not called directly + if isinstance(node, ast.Call) and isinstance(node.func, ast.Name): + assert node.func.id != "render_question" + + +def test_pass_manager_does_not_construct_question_text() -> None: + path = ( + Path(__file__).resolve().parents[1] + / "generate" + / "contemplation" + / "pass_manager.py" + ) + content = path.read_text(encoding="utf-8") + forbidden_templates = ["What ", "Which ", "How many", "Please provide"] + for template in forbidden_templates: + assert template not in content, ( + f"pass_manager.py must not construct prose templates like {template!r}" + ) + + +def test_q1b_carveout_preserved_during_pass_manager_ask_integration() -> None: + assert "missing_total_count" in Q1B_ASK_CARVE_OUT + assert "missing_weighted_total" in Q1B_ASK_CARVE_OUT + + for name in Q1B_ASK_CARVE_OUT: + family = family_by_name(name) + assert family is not None + assert family.proposal_allowed is True + + +def test_renderable_ask_path_returns_question_needed_under_exercise_ask(monkeypatch, tmp_path) -> None: + from core.comprehension_attempt import ComprehensionAttempt, RouteResult + import generate.contemplation.pass_manager as pm + from generate.binding_graph.model import SourceSpanLink + + span = SourceSpanLink(source_id="src", start=0, end=8, text="chickens") + attempt = ComprehensionAttempt( + organ="r2_constraints", + outcome="setup_refused", + refusal_reason="missing_total_count", + evidence=(span,), + ) + + monkeypatch.setattr(pm, "route_setup", lambda text, case_id=None: RouteResult((attempt,), None, "all_refused")) + + question_root = tmp_path / "teaching" / "questions" + proposal_root = tmp_path / "teaching" / "proposals" + + repo_questions_dir = Path(__file__).resolve().parents[1] / "teaching" / "questions" + before_files = set(repo_questions_dir.glob("**/*")) if repo_questions_dir.exists() else set() + + # 1. Assert that without exercise_ask, it falls back to PROPOSAL_EMITTED (due to carve-out) + res_normal = contemplate( + "chickens", + proposal_root=proposal_root, + question_root=question_root, + ) + assert res_normal.terminal == Terminal.PROPOSAL_EMITTED + + # 2. Assert that with exercise_ask=True, it returns QUESTION_NEEDED + calls = [] + orig = pm._delivery_outcome_for_limitation + def wrapped(assessment): + calls.append(assessment) + return orig(assessment) + monkeypatch.setattr(pm, "_delivery_outcome_for_limitation", wrapped) + + res_ask = contemplate( + "chickens", + proposal_root=proposal_root, + question_root=question_root, + exercise_ask=True, + ) + assert res_ask.terminal == Terminal.QUESTION_NEEDED + assert len(calls) == 1 # Verify it is called exactly once! No double delivery / double render! + + # Verify the question artifact path + assert res_ask.proposal_path is not None + artifact_path = Path(res_ask.proposal_path) + assert artifact_path.exists() + + # Assert question artifact is under question_root + assert question_root in artifact_path.parents + # Assert question artifact is not under proposal_root + assert proposal_root not in artifact_path.parents + + # Assert no repo-local teaching/questions artifact is created during tests + after_files = set(repo_questions_dir.glob("**/*")) if repo_questions_dir.exists() else set() + assert before_files == after_files + + +def test_unrenderable_ask_falls_back_in_pass_manager(monkeypatch, tmp_path) -> None: + from core.comprehension_attempt import ComprehensionAttempt, RouteResult + import generate.contemplation.pass_manager as pm + from core.epistemic_questions.delivery import DeliveryOutcome + + attempt = ComprehensionAttempt( + organ="r2_constraints", + outcome="setup_refused", + refusal_reason="missing_total_count", + ) + + monkeypatch.setattr(pm, "route_setup", lambda text, case_id=None: RouteResult((attempt,), None, "all_refused")) + monkeypatch.setattr( + pm, + "_delivery_outcome_for_limitation", + lambda assessment: DeliveryOutcome(Terminal.PROPOSAL_EMITTED, None, "multi_slot_not_supported") + ) + + question_root = tmp_path / "teaching" / "questions" + proposal_root = tmp_path / "teaching" / "proposals" + + res = contemplate( + "chickens", + proposal_root=proposal_root, + question_root=question_root, + exercise_ask=True, + ) + assert res.terminal == Terminal.PROPOSAL_EMITTED + assert res.terminal is not Terminal.QUESTION_NEEDED + assert res.proposal_path is not None + assert Path(res.proposal_path).exists() + + +def test_family_none_does_not_crash_ask_branch(monkeypatch, tmp_path) -> None: + from core.comprehension_attempt import ComprehensionAttempt, RouteResult + import generate.contemplation.pass_manager as pm + from core.epistemic_disclosure.limitation import LimitationAssessment + import core.epistemic_disclosure.limitation as lim_mod + + fake_assessment = LimitationAssessment( + limitation_kind="missing_information", + resolution_action="ask_question", + epistemic_state=EpistemicState.UNDETERMINED, + owner_organ="r2_constraint", + blocking_reason="nonexistent_family_name", + ) + + attempt = ComprehensionAttempt( + organ="r2_constraints", + outcome="setup_refused", + refusal_reason="nonexistent_family_name", + ) + + monkeypatch.setattr(pm, "route_setup", lambda text, case_id=None: RouteResult((attempt,), None, "all_refused")) + monkeypatch.setattr(lim_mod, "assess_from_attempt", lambda att: fake_assessment) + + question_root = tmp_path / "teaching" / "questions" + proposal_root = tmp_path / "teaching" / "proposals" + + res = contemplate( + "chickens", + proposal_root=proposal_root, + question_root=question_root, + exercise_ask=True, + ) + assert res.terminal == Terminal.NO_PROGRESS + + +def test_boundary_wins_over_ask_in_pass_manager(monkeypatch, tmp_path) -> None: + from core.comprehension_attempt import ComprehensionAttempt, RouteResult + import generate.contemplation.pass_manager as pm + from generate.binding_graph.model import SourceSpanLink + + span = SourceSpanLink(source_id="src", start=0, end=8, text="chickens") + attempt_boundary = ComprehensionAttempt( + organ="r2_constraints", + outcome="setup_refused", + refusal_reason="too_many_categories", # maps to unsupported_system_size (must_remain_refused = True) + evidence=(span,), + ) + attempt_ask = ComprehensionAttempt( + organ="r2_constraints", + outcome="setup_refused", + refusal_reason="missing_total_count", # ask carve-out + evidence=(span,), + ) + + monkeypatch.setattr( + pm, + "route_setup", + lambda text, case_id=None: RouteResult((attempt_boundary, attempt_ask), None, "all_refused"), + ) + + question_root = tmp_path / "teaching" / "questions" + proposal_root = tmp_path / "teaching" / "proposals" + + res = contemplate( + "chickens", + proposal_root=proposal_root, + question_root=question_root, + exercise_ask=True, + ) + # Assert terminal remains REFUSED_UNSUPPORTED_FAMILY or REFUSED_KNOWN_BOUNDARY, not QUESTION_NEEDED + assert res.terminal == Terminal.REFUSED_UNSUPPORTED_FAMILY + # Ensure no question is written under question_root + assert not question_root.exists() or len(list(question_root.glob("**/*"))) == 0