fix(ask): avoid double delivery in pass manager ASK path

generate/contemplation/pass_manager.py

@@ -89,7 +89,7 @@ def _delivery_outcome_for_limitation(assessment: LimitationAssessment) -> Delive
 
 def _handle_ask_delivery(
     assessment: LimitationAssessment,
-    family_obj: Any,
+    family_name: str,
     findings: list[Finding],
     attempts: tuple[ComprehensionAttempt, ...],
     text: str,
@@ -99,28 +99,40 @@ def _handle_ask_delivery(
 ) -> ContemplationResult:
     outcome = _delivery_outcome_for_limitation(assessment)
     if outcome.terminal == Terminal.QUESTION_NEEDED:
-        from core.epistemic_questions.delivery import emit_question
+        assert outcome.question is not None
-        path = emit_question(assessment, root=proposal_root)
+        import json
+        from core.epistemic_questions.delivery import question_path
+
+        path = question_path(outcome.question, None)
+        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_obj.name,
+            selected_organ=selected_organ, family=family_name,
-            proposal_path=str(path) if path else None,
+            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_obj.name,
+                    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_obj.name,
+            selected_organ=selected_organ, family=family_name,
         )
 
 
@@ -189,14 +201,14 @@ def _solve_and_verify_r2(
     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, Q1B_ASK_CARVE_OUT
+        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, findings, attempts, text, proposal_root, exercise_ask,
+                        assessment, family_obj.name, findings, attempts, text, proposal_root, exercise_ask,
                         selected_organ="r2_constraints"
                     )
         findings.append(Finding("terminal", Terminal.REFUSED_KNOWN_BOUNDARY.value))
@@ -245,14 +257,14 @@ def _solve_and_verify_r3(
     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, Q1B_ASK_CARVE_OUT
+        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, findings, attempts, text, proposal_root, exercise_ask,
+                        assessment, family_obj.name, findings, attempts, text, proposal_root, exercise_ask,
                         selected_organ="r3_rate"
                     )
         findings.append(Finding("terminal", Terminal.REFUSED_KNOWN_BOUNDARY.value))
@@ -304,7 +316,7 @@ def _solve_and_verify_cmb(
         # 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, Q1B_ASK_CARVE_OUT
+        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:
@@ -312,7 +324,7 @@ def _solve_and_verify_cmb(
             if assessment.resolution_action == "ask_question":
                 if exercise_ask:
                     return _handle_ask_delivery(
-                        assessment, family_obj, findings, attempts, text, proposal_root, exercise_ask,
+                        assessment, family_obj.name, findings, attempts, text, proposal_root, exercise_ask,
                         selected_organ="r4_combined_rate"
                     )
         findings.append(Finding("terminal", Terminal.REFUSED_KNOWN_BOUNDARY.value))
@@ -362,14 +374,12 @@ def _classify_all_refused(
 
     # Check for ASK delivery.
     for attempt in considered:
-        from core.epistemic_disclosure.limitation import assess_from_attempt, Q1B_ASK_CARVE_OUT
+        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:
-                from core.comprehension_attempt.failure_family import family_by_name
-                family_obj = family_by_name(assessment.blocking_reason)
                 return _handle_ask_delivery(
-                    assessment, family_obj, findings, attempts, text, proposal_root, exercise_ask
+                    assessment, assessment.blocking_reason, findings, attempts, text, proposal_root, exercise_ask
                 )
 
     families = [(a, family_for_reason(a.refusal_reason)) for a in considered]

tests/test_ask_pass_manager_delivery.py

@@ -122,3 +122,95 @@ def test_q1b_carveout_preserved_during_pass_manager_ask_integration() -> None:
         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"))
+
+    # 1. Assert that without exercise_ask, it falls back to PROPOSAL_EMITTED (due to carve-out)
+    res_normal = contemplate("chickens", proposal_root=tmp_path)
+    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=tmp_path, 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()
+    # Confirm it is NOT written under proposal_root / teaching/proposals
+    assert "teaching/questions" in res_ask.proposal_path
+    assert str(tmp_path) not in res_ask.proposal_path
+
+
+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")
+    )
+
+    res = contemplate("chickens", proposal_root=tmp_path, 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)
+
+    res = contemplate("chickens", proposal_root=tmp_path, exercise_ask=True)
+    assert res.terminal == Terminal.NO_PROGRESS