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

generate/contemplation/pass_manager.py

@@ -94,6 +94,7 @@ def _handle_ask_delivery(
     attempts: tuple[ComprehensionAttempt, ...],
     text: str,
     proposal_root: Path | None,
+    question_root: Path | None,
     exercise_ask: bool,
     selected_organ: str | None = None,
 ) -> ContemplationResult:
@@ -103,7 +104,7 @@ def _handle_ask_delivery(
         import json
         from core.epistemic_questions.delivery import question_path
 
-        path = question_path(outcome.question, None)
+        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),
@@ -142,6 +143,7 @@ def contemplate(
     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:
@@ -171,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, proposal_root, exercise_ask)
+            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, proposal_root, exercise_ask)
+            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, proposal_root, exercise_ask)
+            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(
@@ -183,7 +185,7 @@ def contemplate(
         )
 
     # route.status == "all_refused"
-    return _classify_all_refused(text, attempts, findings, proposal_root, exercise_ask)
+    return _classify_all_refused(text, attempts, findings, proposal_root, question_root, exercise_ask)
 
 
 def _solve_and_verify_r2(
@@ -193,6 +195,7 @@ def _solve_and_verify_r2(
     findings: list[Finding],
     attempts: tuple[ComprehensionAttempt, ...],
     proposal_root: Path | None,
+    question_root: Path | None,
     exercise_ask: bool,
 ) -> ContemplationResult:
     problem = read_constraint_problem(text)
@@ -208,7 +211,7 @@ def _solve_and_verify_r2(
             if assessment.resolution_action == "ask_question":
                 if exercise_ask:
                     return _handle_ask_delivery(
-                        assessment, family_obj.name, findings, attempts, text, proposal_root, exercise_ask,
+                        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))
@@ -249,6 +252,7 @@ def _solve_and_verify_r3(
     findings: list[Finding],
     attempts: tuple[ComprehensionAttempt, ...],
     proposal_root: Path | None,
+    question_root: Path | None,
     exercise_ask: bool,
 ) -> ContemplationResult:
     problem = read_rate_problem(text)
@@ -264,7 +268,7 @@ def _solve_and_verify_r3(
             if assessment.resolution_action == "ask_question":
                 if exercise_ask:
                     return _handle_ask_delivery(
-                        assessment, family_obj.name, findings, attempts, text, proposal_root, exercise_ask,
+                        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))
@@ -305,6 +309,7 @@ def _solve_and_verify_cmb(
     findings: list[Finding],
     attempts: tuple[ComprehensionAttempt, ...],
     proposal_root: Path | None,
+    question_root: Path | None,
     exercise_ask: bool,
 ) -> ContemplationResult:
     problem = read_combined_rate_problem(text)
@@ -324,7 +329,7 @@ def _solve_and_verify_cmb(
             if assessment.resolution_action == "ask_question":
                 if exercise_ask:
                     return _handle_ask_delivery(
-                        assessment, family_obj.name, findings, attempts, text, proposal_root, exercise_ask,
+                        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))
@@ -363,6 +368,7 @@ 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
@@ -372,19 +378,9 @@ def _classify_all_refused(
     if cmb_is_authoritative(attempts):
         considered = tuple(a for a in attempts if a.organ != "r3_rate")
 
-    # Check for ASK delivery.
-    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, exercise_ask
-                )
-
     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 = (
@@ -395,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

@@ -139,8 +139,18 @@ def test_renderable_ask_path_returns_question_needed_under_exercise_ask(monkeypa
 
     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=tmp_path)
+    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
@@ -151,7 +161,12 @@ def test_renderable_ask_path_returns_question_needed_under_exercise_ask(monkeypa
         return orig(assessment)
     monkeypatch.setattr(pm, "_delivery_outcome_for_limitation", wrapped)
 
-    res_ask = contemplate("chickens", proposal_root=tmp_path, exercise_ask=True)
+    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!
 
@@ -159,9 +174,15 @@ def test_renderable_ask_path_returns_question_needed_under_exercise_ask(monkeypa
     assert res_ask.proposal_path is not None
     artifact_path = Path(res_ask.proposal_path)
     assert artifact_path.exists()
-    # Confirm it is NOT written under proposal_root / teaching/proposals
+
-    assert "teaching/questions" in res_ask.proposal_path
+    # Assert question artifact is under question_root
-    assert str(tmp_path) not in res_ask.proposal_path
+    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:
@@ -182,7 +203,15 @@ def test_unrenderable_ask_falls_back_in_pass_manager(monkeypatch, tmp_path) -> N
         lambda assessment: DeliveryOutcome(Terminal.PROPOSAL_EMITTED, None, "multi_slot_not_supported")
     )
 
-    res = contemplate("chickens", proposal_root=tmp_path, exercise_ask=True)
+    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
@@ -212,5 +241,53 @@ def test_family_none_does_not_crash_ask_branch(monkeypatch, tmp_path) -> None:
     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)
 
-    res = contemplate("chickens", proposal_root=tmp_path, exercise_ask=True)
+    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