feat(ask): add served ASK artifact adapter

Adds the Stage 2 served ASK artifact adapter and adapter-only tests.

- Validates Q1-D DeliveredQuestion artifacts before ASK eligibility
- Uses ask_serving_enabled as a default-dark gate
- Uses the disclosure bus disposition mapping
- Fails closed to fallback surface/disposition on invalid artifacts
- Does not wire chat/runtime, telemetry, TurnEvent, ChatResponse, CLAIMS, or metrics
- Does not retire Q1B_ASK_CARVE_OUT or flip proposal_allowed

core/epistemic_disclosure/__init__.py

@@ -19,6 +19,9 @@ Shipped so far (all off-serving — nothing here imports ``generate.derivation``
     ``EpistemicState × LimitationAssessment × DisclosureClaim → ServedDisposition``.
     Mapping scaffold only — no rendering, no bus, no ``verify.py``; nothing consumes
     it yet.
+  * :mod:`~core.epistemic_disclosure.ask_serving` — a narrow Q1-D served-ASK artifact
+    adapter. It validates already-rendered question artifacts and returns a typed
+    decision; it does not render prose and does not acquire runtime contemplation.
   * :mod:`~core.epistemic_disclosure.verified_contract` (P1-A) — the VERIFIED contract:
     the obligation, the proof shape, the validator, and the single sanctioned route to
     ``EpistemicState.VERIFIED`` / ``DisclosureClaim.VERIFIED``. Contract only — no
@@ -27,6 +30,10 @@ Shipped so far (all off-serving — nothing here imports ``generate.derivation``
 
 from __future__ import annotations
 
+from core.epistemic_disclosure.ask_serving import (
+    ServedAskDecision,
+    evaluate_served_ask,
+)
 from core.epistemic_disclosure.disclosure_claim import (
     DEFAULT_DISCLOSURE_CLAIM,
     DisclosureClaim,
@@ -64,6 +71,7 @@ __all__ = [
     "LimitationKind",
     "MissingSlot",
     "ResolutionAction",
+    "ServedAskDecision",
     "ServedDisposition",
     "VerificationObligation",
     "VerificationProof",
@@ -73,6 +81,7 @@ __all__ = [
     "assess_from_family",
     "choose_served_disposition",
     "disclosure_for_verification",
+    "evaluate_served_ask",
     "evaluate_verification",
     "terminal_for_action",
 ]

core/epistemic_disclosure/ask_serving.py

@@ -0,0 +1,171 @@
+"""Stage 2 ASK served-surface artifact adapter.
+
+This module is intentionally narrow: it validates a pre-rendered Q1-D
+``DeliveredQuestion`` artifact and decides whether that artifact is eligible to
+be exposed as a served ASK/QUESTION_NEEDED surface. It does not acquire
+contemplation results from runtime and does not render question prose.
+
+Validation enforces the Q1-D artifact contract:
+
+- top-level JSON object only;
+- ``status == "question_only"``;
+- ``requires_review is True``;
+- ``served is False``;
+- ``answer_binding`` is absent or ``None``;
+- ``question`` is an object;
+- ``question.text`` is a non-empty string;
+- ``question.slot_name`` is a non-empty string;
+- ``question_path`` exists on disk and differs from ``proposal_path``.
+
+Any validation failure fails closed to the caller's fallback surface and
+standing disposition. The served text is consumed from the artifact exactly; no
+runtime prose construction or mutation happens here.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any
+
+from core.epistemic_disclosure.disposition import ServedDisposition, choose_served_disposition
+from core.epistemic_disclosure.limitation import LimitationAssessment
+from core.epistemic_questions.serving_gate import ask_serving_enabled
+from core.epistemic_state import EpistemicState
+
+_MISSING = object()
+
+
+@dataclass(frozen=True, slots=True)
+class ServedAskDecision:
+    """The adapter's served-ASK decision."""
+
+    served: bool
+    terminal: str
+    surface: str
+    disposition: ServedDisposition
+
+
+def _terminal_value(contemplation_result: Any) -> str:
+    terminal = getattr(contemplation_result, "terminal", None)
+    if terminal is None:
+        return "NO_PROGRESS"
+    return str(getattr(terminal, "value", terminal))
+
+
+def _fallback_disposition(terminal: str) -> ServedDisposition:
+    if terminal == "PROPOSAL_EMITTED":
+        return ServedDisposition.PROPOSE
+    if terminal == "SOLVED_VERIFIED":
+        return ServedDisposition.COMMIT
+    return ServedDisposition.REFUSE
+
+
+def _fallback_decision(contemplation_result: Any, fallback_surface: str) -> ServedAskDecision:
+    terminal = _terminal_value(contemplation_result)
+    return ServedAskDecision(
+        served=False,
+        terminal=terminal,
+        surface=fallback_surface,
+        disposition=_fallback_disposition(terminal),
+    )
+
+
+def _validate_question_artifact(data: Any, *, question_path: Path, proposal_path: Any) -> str | None:
+    """Return the valid question text, or ``None`` for any contract violation."""
+
+    if not isinstance(data, dict):
+        return None
+    if data.get("status") != "question_only":
+        return None
+    if data.get("requires_review") is not True:
+        return None
+    served = data.get("served", _MISSING)
+    if served is _MISSING or served is not False:
+        return None
+    answer_binding = data.get("answer_binding", _MISSING)
+    if answer_binding is not _MISSING and answer_binding is not None:
+        return None
+
+    question = data.get("question")
+    if not isinstance(question, dict):
+        return None
+
+    text = question.get("text")
+    if not isinstance(text, str) or not text.strip():
+        return None
+
+    slot_name = question.get("slot_name")
+    if not isinstance(slot_name, str) or not slot_name.strip():
+        return None
+
+    if proposal_path is not None and str(question_path) == str(proposal_path):
+        return None
+
+    return text.strip()
+
+
+def evaluate_served_ask(
+    config: Any,
+    contemplation_result: Any,
+    fallback_surface: str,
+) -> ServedAskDecision:
+    """Evaluate whether a Q1-D question artifact may be surfaced as ASK.
+
+    This is a bus/disposition adapter, not a renderer and not the runtime
+    acquisition path. The caller supplies a contemplation result that already
+    points to a delivered question artifact. When the gate is disabled or any
+    artifact invariant fails, the adapter returns the fallback surface and the
+    standing fallback disposition.
+    """
+
+    if not ask_serving_enabled(config):
+        return _fallback_decision(contemplation_result, fallback_surface)
+
+    if _terminal_value(contemplation_result) != "QUESTION_NEEDED":
+        return _fallback_decision(contemplation_result, fallback_surface)
+
+    question_path_value = getattr(contemplation_result, "question_path", None)
+    proposal_path_value = getattr(contemplation_result, "proposal_path", None)
+    if not question_path_value or question_path_value == proposal_path_value:
+        return _fallback_decision(contemplation_result, fallback_surface)
+
+    question_path = Path(question_path_value)
+    if not question_path.is_file():
+        return _fallback_decision(contemplation_result, fallback_surface)
+
+    try:
+        payload = json.loads(question_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError):
+        return _fallback_decision(contemplation_result, fallback_surface)
+
+    question_text = _validate_question_artifact(
+        payload,
+        question_path=question_path,
+        proposal_path=proposal_path_value,
+    )
+    if question_text is None:
+        return _fallback_decision(contemplation_result, fallback_surface)
+
+    limitation = LimitationAssessment(
+        limitation_kind="missing_information",
+        resolution_action="ask_question",
+        epistemic_state=EpistemicState.UNDETERMINED,
+        owner_organ=payload.get("owner_organ"),
+        blocking_reason=str(payload.get("blocking_reason", "")),
+    )
+    disposition = choose_served_disposition(
+        epistemic_state=EpistemicState.UNDETERMINED,
+        limitation=limitation,
+    )
+
+    return ServedAskDecision(
+        served=True,
+        terminal="QUESTION_NEEDED",
+        surface=question_text,
+        disposition=disposition,
+    )
+
+
+__all__ = ["ServedAskDecision", "evaluate_served_ask"]

docs/analysis/stage2-disclosure-bus-implementation-map-2026-06-09.md

@@ -0,0 +1,317 @@
+# Stage 2 Disclosure Bus Implementation Map — session pivot to current slices
+
+**Date:** 2026-06-09  
+**Status:** implementation map / controlling checklist — docs only  
+
+This document reconciles the June 8 session-design pivot with the implementation slices that have now landed. It is not a new architecture. It is the map that prevents the work from drifting into isolated feature patches.
+
+## 0. Why this map exists
+
+The recent implementation work touched ASK and VERIFIED serving foundations in small slices. That was correct, but the controlling plan is larger than any one PR:
+
+- `docs/sessions/2026-06-08-practice-attempts-and-servability-blade.md`
+- `docs/sessions/2026-06-08-epistemic-question-articulation-first-skill-of-contemplation.md`
+- `docs/analysis/stage2-epistemic-disclosure-bus-verified-v1-scoping-2026-06-08.md`
+- `docs/analysis/q1-epistemic-question-articulation-v1-scoping-2026-06-08.md`
+- `docs/analysis/ask-serving-integration-scoping-2026-06-09.md`
+- `docs/analysis/verified-serving-wiring-scoping-2026-06-09.md`
+
+The governing pivot is:
+
+```text
+wrong=0 is not a binary answer/refuse command.
+wrong=0 is no false presentation of epistemic status.
+practice/contemplation may explore typed, isolated candidates.
+serving must disclose only what is truthfully labelable.
+```
+
+Therefore the served surface should not grow by one-off feature branches. It should grow by activating tenants on the already-scoped Epistemic Disclosure Bus.
+
+## 1. Controlling doctrine from the session docs
+
+### 1.1 Practice versus serving
+
+The practice/servability session separates two lanes:
+
+| Lane | Purpose | License | Primary danger |
+| --- | --- | --- | --- |
+| practice / contemplation | explore, attempt, eliminate, learn | typed and isolated candidates may exist | getting stuck / never learning |
+| serving | emit to a user/downstream surface | no false presentation of epistemic status | misrepresenting uncertainty as truth |
+
+The practical consequence is that internal attempts, ASK artifacts, proposals, and verification proofs may exist off-serving without being user-visible. Serving requires an explicit governance decision.
+
+### 1.2 The servability blade is not a new parallel system
+
+The session document explicitly amends the initial sketch: the servability blade must reuse the already-shipped ADR-0206 response-governance seam (`ReachPolicy`, `govern_response`, `shape_surface`) rather than invent a new parallel object.
+
+For Stage 2 work, every served-surface slice must therefore ask:
+
+```text
+Does this activate the existing governance seam, or is it bypassing it?
+```
+
+A bypass is architectural drift.
+
+### 1.3 ASK is typed intake, not chat clarification
+
+The question-articulation session defines a question as a typed request for missing state. The first contemplative move is not “ask a question”; it is:
+
+```text
+What is preventing resolution — and what KIND of limitation is it?
+```
+
+Only limitations of kind `missing_information` or `ambiguous_structure` may become ASK. Capability gaps propose. Hard boundaries refuse. Contradictions report. Input-shape cases step aside.
+
+### 1.4 QUESTION_NEEDED and PROPOSAL_EMITTED are siblings
+
+The session doc draws a hard line:
+
+| Terminal | Meaning |
+| --- | --- |
+| `QUESTION_NEEDED` | the input is under-specified but the problem is potentially knowable after one missing datum is supplied |
+| `PROPOSAL_EMITTED` | the input is sufficiently specified but CORE lacks the transform/capability |
+
+This distinction must be preserved in artifacts, paths, tests, and served surfaces.
+
+## 2. Stage 2 bus frame
+
+The Stage 2 scoping doc reframes the frontier:
+
+```text
+What may CORE disclose through the served surface — and under what governed disposition?
+```
+
+The bus is the consolidating view:
+
+```text
+EpistemicState + LimitationAssessment + proof/license evidence
+  -> ServedDisposition / disclosure decision
+  -> shaped surface through the governance seam
+```
+
+VERIFIED is not the whole system. It is one tenant. ASK is another. Scope-boundary explanation, contradiction reporting, proposal-only, partial progress, multiple candidates, and provisional working answers are later tenants or modes on the same governance surface.
+
+## 3. Current implementation state
+
+### 3.1 ASK tenant — landed foundation
+
+The following ASK foundation is now present:
+
+```text
+Q1-B typed ASK residue / MissingSlot / LimitationAssessment ask_question
+Q1-C grounded-only renderer
+Q1-D off-serving DeliveredQuestion / deliver_ask / teaching/questions sink
+ask_serving_enabled helper, default-dark
+RuntimeConfig.ask_serving_enabled = False
+pass_manager off-serving ASK integration behind exercise_ask
+ContemplationResult.question_path separated from proposal_path
+```
+
+Important preserved boundaries:
+
+```text
+no chat/runtime.py serving path
+no served ASK surface
+no Q1B_ASK_CARVE_OUT retirement
+no proposal_allowed registry flip
+no CLAIMS or benchmark movement
+boundary-first remains before ASK
+question_only sink remains distinct from proposal_only sink
+```
+
+### 3.2 ASK tenant — not yet done
+
+The following are still open:
+
+```text
+served ASK / QUESTION_NEEDED through the governance bus
+Q2 AnswerBinding and re-run through the owner organ
+no-question/no-proposal dead-zone proof
+Q1B_ASK_CARVE_OUT retirement proof
+registry flip for missing_total_count / missing_weighted_total
+broader ASK families beyond the current typed-residue subset
+```
+
+### 3.3 VERIFIED tenant — landed foundation
+
+The following VERIFIED foundation is now present:
+
+```text
+P1-A VERIFIED contract
+P1-B off-serving gold-setup-backed R2 producer
+P1-C bound_slots_digest proof hardening
+verified_serving_enabled helper, default-dark
+RuntimeConfig.verified_serving_enabled = False
+verified serving scoping / gold-free independence doc
+```
+
+Important preserved boundaries:
+
+```text
+no served VERIFIED surface
+no verify.py consumption
+no eval-gold-backed proof in serving
+no CLAIMS or benchmark movement
+no gold-free independent reader yet
+```
+
+### 3.4 VERIFIED tenant — not yet done
+
+The following are still open:
+
+```text
+gold-free independent reader/proof source
+poison fixture harness
+holdout-gated verification harness
+deterministic replay digest proof for served verification traces
+verify.py consumption of contract verdict
+served [verified] surface behind verified_serving_enabled
+```
+
+## 4. Recent PRs in plan terms
+
+| PR | Plan role | Status |
+| --- | --- | --- |
+| #664 | Q1-B typed ASK residue + carve-out | merged |
+| #666 | Q1-C grounded-only renderer | merged |
+| #667 | Q1-D decision record | merged |
+| #668 | Q1-D off-serving delivery | merged |
+| #670 | ASK default-dark gate helper | merged |
+| #671 | ASK serving / carve-out retirement scoping | merged |
+| #672 | VERIFIED serving / gold-free independence scoping | merged |
+| #673 | VERIFIED default-dark gate helper | merged |
+| #674 | explicit ASK + VERIFIED config fields | merged |
+| #675 | off-serving ASK pass_manager integration | merged |
+| #676 | split question_path from proposal_path | merged |
+
+This sequence was mostly faithful to the plan, but the controlling label should be:
+
+```text
+ASK tenant implementation on the Epistemic Disclosure Bus foundation
+```
+
+not merely:
+
+```text
+question feature / pass_manager feature
+```
+
+## 5. Correct next-slice ordering
+
+### 5.1 ASK next code slice — bus activation, not runtime bypass
+
+The next ASK code slice should not be described as “wire chat/runtime.” It should be:
+
+```text
+Activate ASK/clarify as a served tenant through the disclosure/governance bus.
+```
+
+Minimum requirements:
+
+- use `ask_serving_enabled(config)` as a necessary gate;
+- consume `ContemplationResult.question_path` / delivered question artifact;
+- do not construct question prose in serving;
+- do not serve unrenderable ASK;
+- preserve `Q1B_ASK_CARVE_OUT`;
+- preserve proposal signal when gate is disabled;
+- do not flip `proposal_allowed`;
+- do not bypass `govern_response` / `shape_surface` if that seam is applicable;
+- if the current governance seam cannot carry ASK yet, stop and scope the exact missing adapter first.
+
+Expected test names or equivalents:
+
+```text
+test_ask_serving_disabled_preserves_existing_proposal_signal
+test_ask_serving_enabled_surfaces_question_needed_from_artifact
+test_unrenderable_ask_never_serves_question_needed
+test_question_only_not_proposal_only
+test_served_ask_does_not_construct_question_prose
+test_served_ask_uses_governance_bus_not_parallel_runtime_path
+```
+
+### 5.2 ASK following slice — no-dead-zone proof
+
+After served ASK exists behind the gate, prove that the carve-out families cannot fall into a no-question/no-proposal dead zone.
+
+Required proof shape:
+
+```text
+for missing_total_count / missing_weighted_total:
+  gate disabled -> proposal signal preserved
+  gate enabled + renderable -> QUESTION_NEEDED served safely
+  gate enabled + unrenderable -> standing fallback, never contentless QUESTION_NEEDED
+  no case yields neither proposal nor valid question
+```
+
+### 5.3 ASK later slice — Q1B_ASK_CARVE_OUT retirement
+
+Only after the no-dead-zone proof passes may a later PR consider:
+
+```text
+proposal_allowed=False for missing_total_count / missing_weighted_total
+remove/retire Q1B_ASK_CARVE_OUT
+```
+
+That PR must be explicit, separate, and guarded by tests.
+
+### 5.4 VERIFIED next code slice — gold-free independent reader
+
+The next VERIFIED code slice is not served VERIFIED. It is:
+
+```text
+Design/prototype a gold-free independent reader/proof source.
+```
+
+Requirements:
+
+- no eval-gold setup in serving;
+- no gold answer;
+- no benchmark fixture;
+- distinct primary/independent reader lineages;
+- convergent canonical read digests;
+- strict rejection of same-reader-twice;
+- strict rejection of second-solver-over-one-read;
+- off-serving only.
+
+### 5.5 VERIFIED later slices
+
+Only after gold-free independence exists:
+
+```text
+poison fixture harness
+holdout-gated verification harness
+verify.py consumption scoping
+served [verified] behind verified_serving_enabled
+```
+
+## 6. Explicit anti-drift rules
+
+The following are forbidden unless a future ADR/decision record explicitly reopens them:
+
+```text
+no standalone chat/runtime ASK patch that bypasses the bus
+no served ASK without ask_serving_enabled
+no served ASK that constructs prose instead of consuming DeliveredQuestion
+no contentless QUESTION_NEEDED
+no question artifact under proposal_path
+no proposal artifact under question_path
+no Q1B carve-out retirement before no-dead-zone proof
+no served VERIFIED from eval-gold producer
+no verify.py VERIFIED consumption before gold-free independent reader + poison/holdout harness
+no direct EpistemicState.VERIFIED construction outside the contract route
+no CLAIMS or benchmark movement from off-serving artifacts
+```
+
+## 7. Current status checkpoint
+
+As of the `question_path` cleanup landing, the project is here:
+
+```text
+Stage 2 bus doctrine: scoped, not fully active
+ASK tenant: off-serving foundation complete enough for served-bus scoping
+VERIFIED tenant: contract/gate foundation complete, serving blocked on gold-free independence
+Next ASK move: bus-governed served ASK, no carve-out retirement
+Next VERIFIED move: gold-free independent reader, no served VERIFIED
+```
+
+Use this document as the checklist before issuing the next implementation brief.

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(
     {
@@ -74,16 +78,76 @@ class ContemplationResult:
     answer: int | None = None
     family: str | None = None
     proposal_path: str | None = None
+    question_path: str | None = None
     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=None,
+            question_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 +175,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 +187,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 +196,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 +253,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 +310,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 +322,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 +370,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 +379,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 +393,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,336 @@
+"""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, and chat/chat.runtime is not imported
+        if isinstance(node, ast.ImportFrom):
+            assert node.module != "core.epistemic_questions.render"
+            assert node.module != "chat.runtime"
+            assert node.module != "chat"
+            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"
+                assert alias.name != "chat.runtime"
+                assert alias.name != "chat"
+
+        # 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
+    assert res_normal.proposal_path is not None
+    assert res_normal.question_path is None
+    assert proposal_root in Path(res_normal.proposal_path).parents
+
+    # 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 None
+    assert res_ask.question_path is not None
+    artifact_path = Path(res_ask.question_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()
+    assert res.question_path is None
+
+
+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
+    assert res.question_path is None
+    assert res.proposal_path is None
+    # Ensure no question is written under question_root
+    assert not question_root.exists() or len(list(question_root.glob("**/*"))) == 0
+
+
+def test_unrenderable_ask_falls_back_to_no_progress_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.NO_PROGRESS, None, "some_fallback_reason")
+    )
+
+    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
+    assert res.proposal_path is None
+    assert res.question_path is None

tests/test_ask_serving_integration.py

@@ -0,0 +1,287 @@
+"""Focused tests for the Stage 2 served ASK artifact adapter.
+
+These tests intentionally avoid ``chat.runtime``. This slice is adapter-only:
+it validates Q1-D question artifacts and returns a typed decision, but does not
+wire runtime acquisition of ``ContemplationResult``.
+"""
+
+from __future__ import annotations
+
+import ast
+import json
+from pathlib import Path
+
+from core.config import RuntimeConfig
+from core.epistemic_disclosure import ServedAskDecision, evaluate_served_ask
+from core.epistemic_disclosure.disposition import ServedDisposition, choose_served_disposition
+
+
+class DummyTerminal:
+    def __init__(self, value: str):
+        self.value = value
+
+
+class DummyContemplationResult:
+    def __init__(
+        self,
+        terminal: str,
+        *,
+        question_path: str | None = None,
+        proposal_path: str | None = None,
+        family: str | None = None,
+    ) -> None:
+        self.terminal = DummyTerminal(terminal)
+        self.question_path = question_path
+        self.proposal_path = proposal_path
+        self.family = family
+
+
+def _write_artifact(path: Path, data: dict) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(data), encoding="utf-8")
+
+
+def _valid_payload(text: str = "How many crates are left?") -> dict:
+    return {
+        "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,
+    }
+
+
+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_ask_serving_disabled_preserves_existing_fallback_surface(tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "q.json"
+    _write_artifact(q_path, _valid_payload())
+
+    decision = evaluate_served_ask(
+        RuntimeConfig(ask_serving_enabled=False),
+        _question_result(q_path),
+        "fallback",
+    )
+
+    assert isinstance(decision, ServedAskDecision)
+    assert decision.served is False
+    assert decision.terminal == "QUESTION_NEEDED"
+    assert decision.surface == "fallback"
+    assert decision.disposition is ServedDisposition.REFUSE
+
+
+def test_ask_serving_enabled_surfaces_question_needed_from_artifact(tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "q.json"
+    p_path = tmp_path / "proposals" / "p.json"
+    _write_artifact(q_path, _valid_payload("How many crates are left?"))
+
+    decision = evaluate_served_ask(
+        RuntimeConfig(ask_serving_enabled=True),
+        _question_result(q_path, p_path),
+        "fallback",
+    )
+
+    assert decision.served is True
+    assert decision.terminal == "QUESTION_NEEDED"
+    assert decision.surface == "How many crates are left?"
+    assert decision.disposition is ServedDisposition.ASK
+
+
+def test_served_ask_uses_governance_disposition_bus(monkeypatch, tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "q.json"
+    _write_artifact(q_path, _valid_payload())
+    calls = []
+
+    def traced_choose_served_disposition(*args, **kwargs):
+        calls.append((args, kwargs))
+        return choose_served_disposition(*args, **kwargs)
+
+    monkeypatch.setattr(
+        "core.epistemic_disclosure.ask_serving.choose_served_disposition",
+        traced_choose_served_disposition,
+    )
+
+    decision = evaluate_served_ask(
+        RuntimeConfig(ask_serving_enabled=True),
+        _question_result(q_path),
+        "fallback",
+    )
+
+    assert decision.served is True
+    assert decision.disposition is ServedDisposition.ASK
+    assert len(calls) == 1
+    assert calls[0][1]["limitation"].resolution_action == "ask_question"
+
+
+def test_question_path_must_not_equal_proposal_path(tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "same.json"
+    _write_artifact(q_path, _valid_payload())
+
+    decision = evaluate_served_ask(
+        RuntimeConfig(ask_serving_enabled=True),
+        _question_result(q_path, q_path),
+        "fallback",
+    )
+
+    assert decision.served is False
+    assert decision.surface == "fallback"
+
+
+def test_non_question_needed_terminal_preserves_proposal_signal(tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "q.json"
+    _write_artifact(q_path, _valid_payload())
+    result = DummyContemplationResult(
+        "PROPOSAL_EMITTED",
+        question_path=str(q_path),
+        proposal_path=str(tmp_path / "proposals" / "p.json"),
+    )
+
+    decision = evaluate_served_ask(
+        RuntimeConfig(ask_serving_enabled=True),
+        result,
+        "fallback",
+    )
+
+    assert decision.served is False
+    assert decision.terminal == "PROPOSAL_EMITTED"
+    assert decision.surface == "fallback"
+    assert decision.disposition is ServedDisposition.PROPOSE
+
+
+def test_missing_or_unreadable_question_artifact_fails_closed(tmp_path: Path) -> None:
+    missing = tmp_path / "questions" / "missing.json"
+
+    decision = evaluate_served_ask(
+        RuntimeConfig(ask_serving_enabled=True),
+        _question_result(missing),
+        "fallback",
+    )
+
+    assert decision.served is False
+    assert decision.surface == "fallback"
+
+
+def test_malformed_question_artifact_fails_closed(tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "q.json"
+    q_path.parent.mkdir(parents=True, exist_ok=True)
+    q_path.write_text("{bad json", encoding="utf-8")
+
+    decision = evaluate_served_ask(
+        RuntimeConfig(ask_serving_enabled=True),
+        _question_result(q_path),
+        "fallback",
+    )
+
+    assert decision.served is False
+    assert decision.surface == "fallback"
+
+
+def test_rejects_artifact_status_proposal_only(tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "q.json"
+    data = _valid_payload()
+    data["status"] = "proposal_only"
+    _write_artifact(q_path, data)
+
+    decision = evaluate_served_ask(RuntimeConfig(ask_serving_enabled=True), _question_result(q_path), "fallback")
+
+    assert decision.served is False
+    assert decision.surface == "fallback"
+
+
+def test_rejects_artifact_served_true(tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "q.json"
+    data = _valid_payload()
+    data["served"] = True
+    _write_artifact(q_path, data)
+
+    decision = evaluate_served_ask(RuntimeConfig(ask_serving_enabled=True), _question_result(q_path), "fallback")
+
+    assert decision.served is False
+
+
+def test_rejects_artifact_requires_review_false(tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "q.json"
+    data = _valid_payload()
+    data["requires_review"] = False
+    _write_artifact(q_path, data)
+
+    decision = evaluate_served_ask(RuntimeConfig(ask_serving_enabled=True), _question_result(q_path), "fallback")
+
+    assert decision.served is False
+
+
+def test_rejects_artifact_non_null_answer_binding(tmp_path: Path) -> None:
+    q_path = tmp_path / "questions" / "q.json"
+    data = _valid_payload()
+    data["answer_binding"] = {"slot": "total_count", "value": 12}
+    _write_artifact(q_path, data)
+
+    decision = evaluate_served_ask(RuntimeConfig(ask_serving_enabled=True), _question_result(q_path), "fallback")
+
+    assert decision.served is False
+
+
+def test_rejects_missing_or_blank_question_text(tmp_path: Path) -> None:
+    for value in (None, ""):
+        q_path = tmp_path / f"questions_{value!r}" / "q.json"
+        data = _valid_payload()
+        if value is None:
+            del data["question"]["text"]
+        else:
+            data["question"]["text"] = "   "
+        _write_artifact(q_path, data)
+
+        decision = evaluate_served_ask(RuntimeConfig(ask_serving_enabled=True), _question_result(q_path), "fallback")
+
+        assert decision.served is False
+        assert decision.surface == "fallback"
+
+
+def test_rejects_missing_or_blank_slot_name(tmp_path: Path) -> None:
+    for value in (None, ""):
+        q_path = tmp_path / f"questions_slot_{value!r}" / "q.json"
+        data = _valid_payload()
+        if value is None:
+            del data["question"]["slot_name"]
+        else:
+            data["question"]["slot_name"] = "   "
+        _write_artifact(q_path, data)
+
+        decision = evaluate_served_ask(RuntimeConfig(ask_serving_enabled=True), _question_result(q_path), "fallback")
+
+        assert decision.served is False
+        assert decision.surface == "fallback"
+
+
+def test_adapter_does_not_import_question_renderer_or_construct_prose() -> None:
+    path = Path(__file__).resolve().parents[1] / "core" / "epistemic_disclosure" / "ask_serving.py"
+    source = path.read_text(encoding="utf-8")
+    tree = ast.parse(source, filename=str(path))
+
+    for node in ast.walk(tree):
+        if isinstance(node, ast.ImportFrom) and node.module:
+            assert node.module != "core.epistemic_questions.render"
+            assert not node.module.startswith("core.epistemic_questions.render.")
+        if isinstance(node, ast.Import):
+            for alias in node.names:
+                assert alias.name != "core.epistemic_questions.render"
+                assert not alias.name.startswith("core.epistemic_questions.render.")
+        if isinstance(node, ast.Call) and isinstance(node.func, ast.Name):
+            assert node.func.id != "render_question"
+
+    for forbidden_template in ("How many", "What ", "Which ", "Please provide"):
+        assert forbidden_template not in source