Compare commits

..

6 commits

Author SHA1 Message Date
Shay
4308519ca9
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
2026-06-09 12:49:52 -07:00
Shay
426adbd197 docs(stage2): map disclosure-bus implementation state and next slices 2026-06-09 11:32:00 -07:00
Shay
9cff9755cd
feat(ask): split question artifact path from proposal path
Adds dedicated ContemplationResult.question_path for off-serving ASK artifacts and stops storing QUESTION_NEEDED question paths in proposal_path. Proposal artifacts remain under proposal_path. No chat/runtime wiring, served ASK surface, Q1B carve-out retirement, registry flip, or CLAIMS/metric movement.
2026-06-09 11:28:52 -07:00
Shay
b687f62cf0
feat(ask): integrate off-serving ASK delivery in pass manager
Adds off-serving pass_manager integration for ASK delivery through deliver_ask behind explicit exercise_ask. Preserves boundary-first ordering, writes question artifacts to the question sink via question_root, avoids double delivery, and keeps chat/runtime, served ASK, Q1B carve-out, registry, CLAIMS, and metrics untouched.
2026-06-09 10:50:36 -07:00
Shay
3cfaaf9153
feat(config): add explicit default-dark serving gate fields
Adds explicit default-false RuntimeConfig ask_serving_enabled and verified_serving_enabled fields while preserving default-dark helper behavior. Includes focused ASK/VERIFIED gate tests and cross-gate independence coverage. No pass_manager wiring, runtime wiring, verify.py wiring, served behavior, or CLAIMS/metric movement.
2026-06-09 10:24:28 -07:00
Shay
45835987b9
feat(verified): add default-dark VERIFIED serving gate helper
Adds a default-dark verified_serving_enabled helper with focused tests. The helper is unwired: no runtime integration, no verify.py wiring, no eval producer imports, no served VERIFIED behavior, and no CLAIMS/metric movement.
2026-06-09 10:06:36 -07:00
10 changed files with 1294 additions and 6 deletions

View file

@ -335,6 +335,15 @@ class RuntimeConfig:
# default); the engine never raises its own ceiling.
estimation_enabled: bool = False
# ASK serving gate enable flag. When True, ASK serving is allowed.
# Default False (dark).
ask_serving_enabled: bool = False
# VERIFIED serving gate enable flag. When True, VERIFIED serving is allowed.
# Default False (dark).
verified_serving_enabled: bool = False
DEFAULT_IDENTITY_PACK: str = "default_general_v1"
DEFAULT_ETHICS_PACK: str = "default_general_ethics_v1"

View file

@ -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",
]

View file

@ -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"]

View file

@ -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.

View file

@ -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:

View file

@ -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

View file

@ -26,10 +26,14 @@ class _OptInConfig:
def test_default_runtime_config_keeps_ask_serving_dark() -> None:
assert hasattr(RuntimeConfig(), "ask_serving_enabled")
assert RuntimeConfig().ask_serving_enabled is False
assert DEFAULT_CONFIG.ask_serving_enabled is False
assert ask_serving_enabled(DEFAULT_CONFIG) is False
assert ask_serving_enabled(RuntimeConfig()) is False
def test_missing_flag_is_dark_for_legacy_config_shape() -> None:
assert ask_serving_enabled(_LegacyConfig()) is False

View file

@ -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

View file

@ -0,0 +1,31 @@
"""Cross-gate configuration field tests.
Verifies that RuntimeConfig has both ask_serving_enabled and
verified_serving_enabled, that both are default false, and that setting
one does not implicitly enable the other.
"""
from __future__ import annotations
import dataclasses
from core.config import RuntimeConfig
def test_cross_serving_gates_independence() -> None:
# 1. Verify both are present on a fresh config and default to False
config = RuntimeConfig()
assert hasattr(config, "ask_serving_enabled")
assert hasattr(config, "verified_serving_enabled")
assert config.ask_serving_enabled is False
assert config.verified_serving_enabled is False
# 2. Verify setting ask_serving_enabled to True does not enable verified_serving_enabled
config_ask = dataclasses.replace(config, ask_serving_enabled=True)
assert config_ask.ask_serving_enabled is True
assert config_ask.verified_serving_enabled is False
# 3. Verify setting verified_serving_enabled to True does not enable ask_serving_enabled
config_verified = dataclasses.replace(config, verified_serving_enabled=True)
assert config_verified.ask_serving_enabled is False
assert config_verified.verified_serving_enabled is True

View file

@ -30,10 +30,14 @@ class _OptInConfig:
def test_default_runtime_config_keeps_verified_serving_dark() -> None:
assert hasattr(RuntimeConfig(), "verified_serving_enabled")
assert RuntimeConfig().verified_serving_enabled is False
assert DEFAULT_CONFIG.verified_serving_enabled is False
assert verified_serving_enabled(DEFAULT_CONFIG) is False
assert verified_serving_enabled(RuntimeConfig()) is False
def test_missing_flag_is_dark_for_legacy_config_shape() -> None:
assert verified_serving_enabled(_LegacyConfig()) is False