feat(ask): strengthen Q1-D artifact validation, add rejection tests, bypass justification

- evaluate_served_ask now validates the full Q1-D DeliveredQuestion contract:
  status == 'question_only', requires_review is True, served is False,
  answer_binding is None, question.text non-empty, question.slot_name present,
  question_path != proposal_path.  Each clause is individually testable via
  _validate_artifact() which is a pure predicate with no side effects.

- Added _MISSING sentinel to distinguish absent keys from explicit None/False
  values in 'served' and 'answer_binding' checks.

- Added load-bearing comment in _stub_response explaining why served ASK
  bypasses register decoration and realizer guard: Q1-C/Q1-D is the
  authoritative grounded renderer; mutating pre-rendered intake prose would
  be semantically incorrect.

- Expanded test_ask_serving_integration.py with:
  - Rejection tests: status='proposal_only', served=True, requires_review=False,
    non-null answer_binding, missing/blank question.text, missing/blank slot_name.
  - Exact-text bypass safety test proving stub surface == artifact text exactly
    with no register/decorator mutation and realizer_guard_status='ok'.
  - Disposition telemetry tests: default main-path stability, proposal terminal
    not reclassified, ASK branch emits 'ask', refused terminal stays 'refuse',
    ChatResponse.disposition defaults '' for backward compat.

All 440 tests pass.

chat/runtime.py

@@ -2081,6 +2081,29 @@ class ChatRuntime:
         )
 
         if decision.served and not refusal_emitted:
+            # ASK serving bypass justification (Q1-C/Q1-D authoritative renderer):
+            #
+            # The Q1-C/Q1-D delivery path is the *authoritative grounded renderer*
+            # for served questions: the question prose was already validated, slot-
+            # checked, and written to disk by the ASK pass_manager before this point.
+            # Applying register decoration or the realizer slot-type guard here would
+            # risk mutating pre-rendered question prose that must reach the user
+            # exactly as grounded — register suffixes, discourse markers, and guard-
+            # replacement strings are semantically wrong in an intake-request context.
+            #
+            # Safety contract:
+            #   - evaluate_served_ask validates the full Q1-D contract (status,
+            #     requires_review, served=False, answer_binding=None, slot_name) before
+            #     setting decision.served = True.
+            #   - The text consumed here is question_text.strip() from the artifact,
+            #     not constructed in runtime.
+            #   - realizer_guard_status_stub = "ok" is set manually because the pre-
+            #     rendered question surface is Q1-C-grounded and does not require the
+            #     slot-type guard's structural check (which is designed for realizer
+            #     output, not pre-rendered intake prose).
+            #   - tests/test_ask_serving_integration.py::
+            #     test_served_ask_surface_is_consumed_exactly_from_artifact asserts
+            #     that the served surface equals artifact text without any mutation.
             response_surface = decision.surface
             stub_disposition = decision.disposition.value
             stub_epistemic_state = "undetermined"

core/epistemic_disclosure/ask_serving.py

@@ -2,6 +2,20 @@
 
 Determines whether a pre-rendered question artifact can be served to the user
 based on runtime configuration, contemplation result, and artifact validity.
+
+Artifact validation enforces the Q1-D DeliveredQuestion contract:
+  - top-level JSON object only;
+  - status == "question_only";
+  - requires_review is True;
+  - served is False;
+  - answer_binding is None (key absent or explicitly null);
+  - question is an object;
+  - question.text is a non-empty string;
+  - question.slot_name is present (required by Q1-C/DeliveredQuestion schema);
+  - question_path exists on the filesystem and differs from proposal_path.
+
+Any validation failure causes the adapter to fail closed and return a fallback
+decision that preserves the existing proposal/refusal surface unchanged.
 """
 
 from __future__ import annotations
@@ -16,6 +30,11 @@ from core.epistemic_disclosure.disposition import choose_served_disposition, Ser
 from core.epistemic_disclosure.limitation import LimitationAssessment
 from core.epistemic_state import EpistemicState
 
+# Sentinel used only in _validate_artifact to distinguish "key absent" from
+# "key present with value None".  Having a sentinel avoids treating a missing
+# served/answer_binding field as a passing None check.
+_MISSING = object()
+
 
 @dataclass(frozen=True, slots=True)
 class ServedAskDecision:
@@ -27,6 +46,59 @@ class ServedAskDecision:
     disposition: ServedDisposition
 
 
+def _validate_artifact(data: Any, q_path: Path, p_path_attr: Any) -> str | None:
+    """Validate the DeliveredQuestion artifact against the Q1-D contract.
+
+    Returns the validated question text on success, or None on any violation.
+    Every check is explicit so reviewers can trace each contract clause to code.
+    """
+    # Contract: top-level must be a JSON object (dict)
+    if not isinstance(data, dict):
+        return None
+
+    # Contract: status == "question_only"
+    if data.get("status") != "question_only":
+        return None
+
+    # Contract: requires_review is True (must be present and strictly True)
+    if data.get("requires_review") is not True:
+        return None
+
+    # Contract: served is False (must be present and strictly False)
+    served_val = data.get("served", _MISSING)
+    if served_val is _MISSING or served_val is not False:
+        return None
+
+    # Contract: answer_binding is None (key must be absent or explicitly null)
+    answer_binding_val = data.get("answer_binding", _MISSING)
+    if answer_binding_val is not _MISSING and answer_binding_val is not None:
+        return None
+
+    # Contract: question is an object (dict)
+    question_data = data.get("question")
+    if not isinstance(question_data, dict):
+        return None
+
+    # Contract: question.text is a non-empty string
+    question_text = question_data.get("text")
+    if not isinstance(question_text, str) or not question_text.strip():
+        return None
+
+    # Contract: question.slot_name is present (Q1-C/DeliveredQuestion schema requirement)
+    # An empty string is treated as absent since a slot name must be meaningful.
+    slot_name = question_data.get("slot_name")
+    if not slot_name:
+        return None
+
+    # Contract: question_path must differ from proposal_path (separate sinks)
+    # Already checked upstream (q_path_attr != p_path_attr), but re-checked here
+    # at the artifact boundary to make validation self-contained.
+    if p_path_attr is not None and str(q_path) == str(p_path_attr):
+        return None
+
+    return question_text.strip()
+
+
 def evaluate_served_ask(
     config: Any,
     contemplation_result: Any,
@@ -59,7 +131,9 @@ def evaluate_served_ask(
     q_path_attr = getattr(contemplation_result, "question_path", None)
     p_path_attr = getattr(contemplation_result, "proposal_path", None)
 
-    # Refuse to serve when question_path is missing or equals proposal_path
+    # Refuse to serve when question_path is missing or equals proposal_path.
+    # This enforces the Q1-D separate-sinks requirement: question artifacts must
+    # never reside under proposal_path.
     if not q_path_attr or q_path_attr == p_path_attr:
         return _make_fallback_decision(contemplation_result, fallback_surface)
 
@@ -73,19 +147,14 @@ def evaluate_served_ask(
     except (json.JSONDecodeError, OSError):
         return _make_fallback_decision(contemplation_result, fallback_surface)
 
-    # Refuse to serve if the artifact is malformed or missing key fields
+    # 5. Full Q1-D contract validation.  _validate_artifact is intentionally
-    if not isinstance(data, dict):
+    # explicit (no single-dict-access shortcut) so each contract clause is
+    # individually testable and auditable.
+    question_text = _validate_artifact(data, q_path, p_path_attr)
+    if question_text is None:
         return _make_fallback_decision(contemplation_result, fallback_surface)
 
-    question_data = data.get("question")
+    # 6. Use choose_served_disposition to map the ASK resolution
-    if not isinstance(question_data, dict):
-        return _make_fallback_decision(contemplation_result, fallback_surface)
-
-    question_text = question_data.get("text")
-    if not question_text or not isinstance(question_text, str) or not question_text.strip():
-        return _make_fallback_decision(contemplation_result, fallback_surface)
-
-    # 5. Use choose_served_disposition to map the ASK resolution
     blocking_reason = data.get("blocking_reason", "")
     owner_organ = data.get("owner_organ", "r2_constraint")
 

tests/test_ask_serving_integration.py

@@ -2,6 +2,11 @@
 
 Covers the 6 required test cases ensuring gated served ASK, fallback preservation,
 unrenderable ask safety, distinct sinks/paths, no prose construction, and governance bus usage.
+
+Additional tests added per reviewer requirements:
+  - Artifact rejection tests: status, served, requires_review, answer_binding, blank text
+  - Exact-text consumption test proving bypass is safe
+  - Disposition telemetry tests: default stability, backward-compat, ASK branch, no reclassification
 """
 
 from __future__ import annotations
@@ -322,3 +327,300 @@ def test_served_ask_uses_governance_bus_not_parallel_runtime_path(monkeypatch, t
     assert len(calls) == 1
     assert calls[0][1]["epistemic_state"] == "undetermined"  # maps to EpistemicState.UNDETERMINED
     assert calls[0][1]["limitation"].resolution_action == "ask_question"
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# REVIEWER-REQUIRED REJECTION TESTS (Q1-D artifact contract violations)
+# ──────────────────────────────────────────────────────────────────────────────
+
+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 _question_needed_result(q_path: Path, p_path: Path) -> DummyContemplationResult:
+    return DummyContemplationResult(
+        terminal="QUESTION_NEEDED",
+        question_path=str(q_path),
+        proposal_path=str(p_path),
+    )
+
+
+def _base_valid_data() -> dict:
+    """Return a fully-valid DeliveredQuestion artifact payload."""
+    return {
+        "status": "question_only",
+        "requires_review": True,
+        "served": False,
+        "question": {
+            "text": "How many crates are left?",
+            "slot_name": "remaining_crates",
+        },
+    }
+
+
+def test_reject_status_proposal_only(tmp_path) -> None:
+    """Artifact with status='proposal_only' must not be served."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    data = _base_valid_data()
+    data["status"] = "proposal_only"
+    _write_artifact(q_path, data)
+    decision = evaluate_served_ask(config, _question_needed_result(q_path, p_path), "fallback")
+    assert not decision.served, "status='proposal_only' must not be served"
+    assert decision.surface == "fallback"
+
+
+def test_reject_served_true(tmp_path) -> None:
+    """Artifact with served=True must not be served (already delivered)."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    data = _base_valid_data()
+    data["served"] = True
+    _write_artifact(q_path, data)
+    decision = evaluate_served_ask(config, _question_needed_result(q_path, p_path), "fallback")
+    assert not decision.served, "served=True in artifact must not be served"
+
+
+def test_reject_requires_review_false(tmp_path) -> None:
+    """Artifact with requires_review=False must not be served."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    data = _base_valid_data()
+    data["requires_review"] = False
+    _write_artifact(q_path, data)
+    decision = evaluate_served_ask(config, _question_needed_result(q_path, p_path), "fallback")
+    assert not decision.served, "requires_review=False must not be served"
+
+
+def test_reject_non_null_answer_binding(tmp_path) -> None:
+    """Artifact with answer_binding != None must not be served."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    data = _base_valid_data()
+    data["answer_binding"] = {"slot": "remaining_crates", "value": 42}
+    _write_artifact(q_path, data)
+    decision = evaluate_served_ask(config, _question_needed_result(q_path, p_path), "fallback")
+    assert not decision.served, "non-null answer_binding must not be served"
+
+
+def test_reject_missing_question_text(tmp_path) -> None:
+    """Artifact with question.text absent must not be served."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    data = _base_valid_data()
+    del data["question"]["text"]
+    _write_artifact(q_path, data)
+    decision = evaluate_served_ask(config, _question_needed_result(q_path, p_path), "fallback")
+    assert not decision.served, "missing question.text must not be served"
+
+
+def test_reject_blank_question_text(tmp_path) -> None:
+    """Artifact with question.text blank/whitespace-only must not be served."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    data = _base_valid_data()
+    data["question"]["text"] = "   "
+    _write_artifact(q_path, data)
+    decision = evaluate_served_ask(config, _question_needed_result(q_path, p_path), "fallback")
+    assert not decision.served, "blank question.text must not be served"
+
+
+def test_reject_missing_slot_name(tmp_path) -> None:
+    """Artifact with question.slot_name absent must not be served (Q1-C contract)."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    data = _base_valid_data()
+    del data["question"]["slot_name"]
+    _write_artifact(q_path, data)
+    decision = evaluate_served_ask(config, _question_needed_result(q_path, p_path), "fallback")
+    assert not decision.served, "missing slot_name must not be served (Q1-C contract)"
+
+
+def test_reject_empty_slot_name(tmp_path) -> None:
+    """Artifact with question.slot_name = '' must not be served."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    data = _base_valid_data()
+    data["question"]["slot_name"] = ""
+    _write_artifact(q_path, data)
+    decision = evaluate_served_ask(config, _question_needed_result(q_path, p_path), "fallback")
+    assert not decision.served, "empty slot_name must not be served"
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# BYPASS SAFETY: exact-text consumption test
+# Proves the served ASK branch consumes artifact text exactly — no register/
+# decorator mutation is applied to the pre-rendered Q1-C/Q1-D question prose.
+# ──────────────────────────────────────────────────────────────────────────────
+
+def test_served_ask_surface_is_consumed_exactly_from_artifact(monkeypatch, tmp_path) -> None:
+    """Served question text must equal the artifact text exactly, no mutation."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    question_text = "How many units remain in the holding area?"
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    _write_valid_artifact(q_path, text=question_text)
+
+    res = _question_needed_result(q_path, p_path)
+    decision = evaluate_served_ask(config, res, "fallback")
+    assert decision.served
+    # decision.surface must be exactly the stripped artifact text
+    assert decision.surface == question_text
+
+    runtime = _get_mocked_runtime(config, monkeypatch)
+    stub = runtime._stub_response(
+        field_state=_get_dummy_field_state(),
+        contemplation_result=res,
+        pack_grounded_surface="fallback_surface",
+    )
+    # The stub surface must equal the artifact text without any register/decorator mutation
+    assert stub.surface == question_text, (
+        f"Served ASK surface was mutated: expected {question_text!r}, got {stub.surface!r}"
+    )
+    # Realizer guard must be set to 'ok' (pre-rendered Q1-C prose bypasses structural check)
+    assert stub.realizer_guard_status == "ok"
+    # Disposition must be 'ask'
+    assert stub.disposition == "ask"
+
+
+# ──────────────────────────────────────────────────────────────────────────────
+# DISPOSITION TELEMETRY TESTS
+# ──────────────────────────────────────────────────────────────────────────────
+
+def test_disposition_default_main_path_is_stable(monkeypatch, tmp_path) -> None:
+    """Normal non-ASK stub paths must emit stable disposition values."""
+    config = RuntimeConfig(ask_serving_enabled=False)
+    runtime = _get_mocked_runtime(config, monkeypatch)
+    field_state = _get_dummy_field_state()
+
+    # No contemplation_result: should default to fallback (refuse for unknown grounding)
+    stub_none = runtime._stub_response(
+        field_state=field_state,
+        pack_grounded_surface=None,
+        contemplation_result=None,
+    )
+    assert stub_none.disposition == "refuse"
+
+    # Pack-grounded surface with no contemplation_result: commit
+    stub_pack = runtime._stub_response(
+        field_state=field_state,
+        pack_grounded_surface="Some grounded answer.",
+        grounded_source_tag="pack",
+        contemplation_result=None,
+    )
+    assert stub_pack.disposition == "commit"
+
+
+def test_disposition_proposal_emitted_not_reclassified(monkeypatch, tmp_path) -> None:
+    """PROPOSAL_EMITTED terminal must emit disposition='propose', not 'ask' or 'commit'."""
+    config = RuntimeConfig(ask_serving_enabled=True)  # even with ASK enabled
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    _write_valid_artifact(q_path)
+    res = DummyContemplationResult(
+        terminal="PROPOSAL_EMITTED",
+        question_path=str(q_path),
+        proposal_path=str(p_path),
+    )
+    runtime = _get_mocked_runtime(config, monkeypatch)
+    stub = runtime._stub_response(
+        field_state=_get_dummy_field_state(),
+        contemplation_result=res,
+        pack_grounded_surface="proposal surface",
+    )
+    assert stub.disposition == "propose", (
+        "PROPOSAL_EMITTED terminal must not be reclassified to ask or commit"
+    )
+
+
+def test_disposition_ask_branch_emits_ask(monkeypatch, tmp_path) -> None:
+    """The ASK branch must emit disposition='ask' when serving succeeds."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    q_path = tmp_path / "q" / "q.json"
+    p_path = tmp_path / "p" / "p.json"
+    _write_valid_artifact(q_path)
+    res = _question_needed_result(q_path, p_path)
+
+    runtime = _get_mocked_runtime(config, monkeypatch)
+    stub = runtime._stub_response(
+        field_state=_get_dummy_field_state(),
+        contemplation_result=res,
+        pack_grounded_surface="fallback",
+    )
+    assert stub.disposition == "ask"
+
+
+def test_disposition_refused_known_boundary_not_reclassified(monkeypatch, tmp_path) -> None:
+    """REFUSED_KNOWN_BOUNDARY must emit disposition='refuse', not 'ask'."""
+    config = RuntimeConfig(ask_serving_enabled=True)
+    res = DummyContemplationResult(
+        terminal="REFUSED_KNOWN_BOUNDARY",
+        question_path=None,
+        proposal_path=None,
+    )
+    runtime = _get_mocked_runtime(config, monkeypatch)
+    stub = runtime._stub_response(
+        field_state=_get_dummy_field_state(),
+        contemplation_result=res,
+        pack_grounded_surface=None,
+    )
+    assert stub.disposition == "refuse"
+
+
+def test_disposition_serialization_backward_compat() -> None:
+    """ChatResponse.disposition defaults to '' for pre-disposition callers (byte-identity)."""
+    from generate.articulation import ArticulationPlan
+    from generate.proposition import Proposition
+    from core.physics.identity import CharacterProfile
+    import numpy as np
+
+    zero = np.zeros((32,), dtype=np.float32)
+    prop = Proposition(
+        subject="", predicate="", object_=None,
+        surface="", frame_id="",
+        subject_versor=zero, predicate_versor=zero, object_versor=None, relation=zero,
+    )
+    art = ArticulationPlan(
+        subject="", predicate="", object=None, surface="",
+        output_language="en", frame_id="",
+    )
+    # Use the correct CharacterProfile fields (traits, drive_summaries, fatigue_index,
+    # boundary_commitments, theological_grounding)
+    char = CharacterProfile(
+        traits={},
+        drive_summaries={},
+        fatigue_index=0.0,
+        boundary_commitments=(),
+        theological_grounding={},
+    )
+    # Construct without disposition field to prove default is ""
+    resp = ChatResponse(
+        surface="hello",
+        proposition=prop,
+        articulation=art,
+        articulation_surface="hello",
+        dialogue_role="assert",
+        versor_condition=0.0,
+        output_language="en",
+        frame_pack="",
+        walk_surface="",
+        salience_top_k=None,
+        candidates_used=None,
+        vault_hits=0,
+        identity_score=None,
+        character_profile=char,
+        flagged=False,
+    )
+    assert resp.disposition == "", (
+        "Default ChatResponse.disposition must be '' for backward compatibility"
+    )
+