fix(ask): preserve boundary-first order for ASK delivery

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:

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