feat(adr-0024): Phase 2 — honest refusal with typed evidence
Replace plain ValueError at both inner-loop exhaustion sites in
generate/stream.py with InnerLoopExhaustion, a typed ValueError
subclass carrying machine-readable refusal evidence:
reason : RefusalReason (INNER_LOOP_EXHAUSTION)
region_label : which AdmissibilityRegion blocked
step_index : -1 = pre-walk empty intersection;
>=0 = in-walk per-step exhaustion
rejected_attempts : ordered (idx, word, score) triples
Backward-compat by construction: subclassing ValueError preserves
every pre-Phase-2 `except ValueError` handler in chat/runtime.py,
eval lanes, and tests. No edits to chat/runtime.py, field/propagate.py,
algebra/versor.py, or vault/store.py.
Trace path wired:
- CognitiveTurnResult.refusal_reason (str, default "")
- compute_trace_hash folds refusal_reason only when non-empty
-> byte-identical hashes preserved for non-refused turns
- CognitiveTurnPipeline reads via getattr from ChatResponse and
forwards into both trace_hash and result construction
Contract documented in docs/runtime_contracts.md §"Refusal contract".
Tests (tests/test_refusal_contract.py — 10 passing):
- InnerLoopExhaustion isinstance(ValueError) at both raise sites
- In-walk site carries reason/region_label/step_index>=0/
rejected_attempts with (int,str,float) triples
- Pre-walk site uses step_index=-1 sentinel + empty
rejected_attempts
- Pre-walk fires even when inner_loop_admissibility=False
- Trace hash: empty refusal_reason preserves legacy bytes;
non-empty differs; same inputs are stable
Suite results:
smoke: 67 passed
cognition: 121 passed
runtime: 19 passed
full: 1024 passed, 2 skipped
core eval cognition: 13/13, 100% intent accuracy, 100% versor closure
Residual silent path (documented as out-of-scope for Phase 2):
chat/runtime.respond()/arespond() still convert any ValueError to
"" for their public str return contract. So a refused turn today
produces surface == "" with refusal_reason == "" — the typed
evidence is unread between the raise site and the result. The
plumbing on result + trace + pipeline is in place so a future ADR
can wire materialisation (propagate exception to
ChatResponse.refusal_reason, or catch at the pipeline seam) without
re-deriving the contract.
Phase 1 (commit 3940290) and Phase 2 (this commit) were developed
in parallel with disjoint file scope to avoid conflicts.
This commit is contained in:
parent
394029008e
commit
310793a4ea
7 changed files with 529 additions and 8 deletions
|
|
@ -254,6 +254,12 @@ class CognitiveTurnPipeline:
|
|||
)
|
||||
admissibility_trace_hash = hash_admissibility_trace(admissibility_trace)
|
||||
ratification_outcome = ratified.outcome.value
|
||||
# ADR-0024 Phase 2 — refusal_reason flows from a future
|
||||
# materialisation site on ChatResponse. For Phase 2 it is
|
||||
# absent on every non-refused turn; reading via getattr keeps
|
||||
# the trace_hash byte-identical to pre-Phase-2 when no refusal
|
||||
# was materialised (the empty string skips the payload fold).
|
||||
refusal_reason = getattr(response, "refusal_reason", "") or ""
|
||||
trace_hash = compute_trace_hash(
|
||||
input_text=text,
|
||||
filtered_tokens=filtered_tokens,
|
||||
|
|
@ -271,6 +277,7 @@ class CognitiveTurnPipeline:
|
|||
admissibility_trace_hash=admissibility_trace_hash,
|
||||
ratification_outcome=ratification_outcome,
|
||||
region_was_unconstrained=region_was_unconstrained,
|
||||
refusal_reason=refusal_reason,
|
||||
)
|
||||
|
||||
return CognitiveTurnResult(
|
||||
|
|
@ -298,6 +305,7 @@ class CognitiveTurnPipeline:
|
|||
admissibility_trace_hash=admissibility_trace_hash,
|
||||
ratification_outcome=ratification_outcome,
|
||||
region_was_unconstrained=region_was_unconstrained,
|
||||
refusal_reason=refusal_reason,
|
||||
versor_condition=response.versor_condition,
|
||||
trace_hash=trace_hash,
|
||||
)
|
||||
|
|
|
|||
|
|
@ -87,6 +87,19 @@ class CognitiveTurnResult:
|
|||
ratification_outcome: str = ""
|
||||
region_was_unconstrained: bool = True
|
||||
|
||||
# --- inner-loop refusal evidence (ADR-0024 Phase 2) ---
|
||||
# ``refusal_reason`` is the stable string value of a
|
||||
# ``generate.exhaustion.RefusalReason`` when the walk refused this
|
||||
# turn, or the empty string otherwise. Empty-string default is
|
||||
# the contract for "no refusal materialised"; folding into
|
||||
# trace_hash is gated on non-emptiness so non-refused turns keep
|
||||
# byte-identical hashes relative to pre-Phase-2 (CLAUDE.md
|
||||
# determinism invariant). Phase 2 leaves the materialisation site
|
||||
# in chat/runtime.py untouched per the ADR-0024 Phase 2 scope
|
||||
# decision — this field exists so the trace contract is already
|
||||
# in place when a future ADR wires the materialisation path.
|
||||
refusal_reason: str = ""
|
||||
|
||||
# --- invariant bookkeeping ---
|
||||
versor_condition: float = 0.0 # must be < 1e-6
|
||||
trace_hash: str = "" # SHA-256 over deterministic key fields
|
||||
|
|
|
|||
|
|
@ -41,6 +41,7 @@ def compute_trace_hash(
|
|||
admissibility_trace_hash: str = "",
|
||||
ratification_outcome: str = "",
|
||||
region_was_unconstrained: bool = True,
|
||||
refusal_reason: str = "",
|
||||
) -> str:
|
||||
"""Return a deterministic SHA-256 hex digest over the turn's key outputs.
|
||||
|
||||
|
|
@ -85,6 +86,13 @@ def compute_trace_hash(
|
|||
payload["ratification_outcome"] = ratification_outcome
|
||||
if not region_was_unconstrained:
|
||||
payload["region_was_unconstrained"] = False
|
||||
# ADR-0024 Phase 2 — fold refusal_reason only when non-empty so a
|
||||
# turn that did not refuse keeps byte-identical payload bytes (and
|
||||
# therefore byte-identical trace_hash) relative to pre-Phase-2.
|
||||
# Once a turn does materialise a refusal, the reason becomes
|
||||
# load-bearing in replay equality.
|
||||
if refusal_reason:
|
||||
payload["refusal_reason"] = refusal_reason
|
||||
serialized = json.dumps(payload, sort_keys=True, ensure_ascii=False)
|
||||
return hashlib.sha256(serialized.encode("utf-8")).hexdigest()
|
||||
|
||||
|
|
@ -141,4 +149,5 @@ def trace_hash_from_result(result: "CognitiveTurnResult") -> str:
|
|||
admissibility_trace_hash=getattr(result, "admissibility_trace_hash", ""),
|
||||
ratification_outcome=getattr(result, "ratification_outcome", ""),
|
||||
region_was_unconstrained=getattr(result, "region_was_unconstrained", True),
|
||||
refusal_reason=getattr(result, "refusal_reason", ""),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -53,6 +53,47 @@ gated. This closes `evals/calibration/gaps.md` Finding 2.
|
|||
Future realizer work may change the selection policy, but must update this
|
||||
document and the contract tests in the same PR.
|
||||
|
||||
### Refusal contract (ADR-0024 Phase 2)
|
||||
|
||||
When the inner-loop admissibility check leaves no admissible destination
|
||||
for the next step, the generation walk in `generate/stream.py` raises
|
||||
`generate.exhaustion.InnerLoopExhaustion`, a typed subclass of
|
||||
`ValueError` carrying:
|
||||
|
||||
```text
|
||||
reason : RefusalReason (machine-readable taxonomy)
|
||||
region_label : str (which AdmissibilityRegion blocked)
|
||||
step_index : int (-1 = pre-walk empty intersection;
|
||||
>=0 = in-walk per-step exhaustion)
|
||||
rejected_attempts : tuple[(int, str, float), ...] (per-step evidence)
|
||||
```
|
||||
|
||||
Reason codes are minimal in Phase 2: a single `INNER_LOOP_EXHAUSTION`
|
||||
covers both raise sites. Phase 4 (rotor-frame admissibility, ADR-0025)
|
||||
is expected to add a second reason for rotor exhaustion.
|
||||
|
||||
`CognitiveTurnResult.refusal_reason` carries the stable string value of
|
||||
the `RefusalReason` when a turn refuses, and the empty string otherwise.
|
||||
`compute_trace_hash` folds `refusal_reason` into the payload only when
|
||||
non-empty, preserving byte-identical hashes for non-refused turns
|
||||
relative to pre-Phase-2 (determinism invariant). When the field is
|
||||
non-empty, it becomes load-bearing in replay equality.
|
||||
|
||||
Backward compatibility: `InnerLoopExhaustion` is a `ValueError`, so
|
||||
every pre-Phase-2 `except ValueError` handler in `chat/runtime.py`,
|
||||
eval lanes, and tests continues to catch it without modification.
|
||||
|
||||
Residual silent path (out of scope for Phase 2, future ADR):
|
||||
`ChatRuntime.respond()` and `arespond()` still convert any `ValueError`
|
||||
to the empty string for their public `str` return contract, so a real
|
||||
turn that refuses today produces `surface == ""` with
|
||||
`refusal_reason == ""` — the typed evidence is unread between the
|
||||
raise site and the result. The plumbing on `CognitiveTurnResult`,
|
||||
`compute_trace_hash`, and `CognitiveTurnPipeline` is in place so a
|
||||
future ADR can wire materialisation (e.g. propagate the typed
|
||||
exception to `ChatResponse.refusal_reason` or catch at the pipeline
|
||||
seam) without re-deriving the contract.
|
||||
|
||||
## TurnEvent contract
|
||||
|
||||
`TurnEvent.surface`
|
||||
|
|
|
|||
112
generate/exhaustion.py
Normal file
112
generate/exhaustion.py
Normal file
|
|
@ -0,0 +1,112 @@
|
|||
"""
|
||||
InnerLoopExhaustion — evidence-carrying honest refusal at the walk site.
|
||||
|
||||
ADR-0024 Phase 2. When the inner-loop admissibility check leaves no
|
||||
admissible destination for the next step, the walk must refuse rather
|
||||
than silently relax the constraint. Phase 1 raised a plain
|
||||
``ValueError`` with a message string; Phase 2 promotes that to a
|
||||
typed exception that carries the refusal evidence (region label,
|
||||
step index, rejected attempts, machine-readable reason) so downstream
|
||||
layers can materialise the refusal into trace evidence without
|
||||
re-parsing the message.
|
||||
|
||||
This is *not* normalization or repair (CLAUDE.md §Normalization Rules).
|
||||
The walk still refuses at the same site, with the same algebra, with
|
||||
the same effect on caller control flow. The exception subclasses
|
||||
``ValueError`` so every existing ``except ValueError`` handler in the
|
||||
runtime/eval code paths continues to work byte-for-byte — the carrying
|
||||
of structured evidence is additive.
|
||||
|
||||
Reason codes are minimal in Phase 2. A single
|
||||
``INNER_LOOP_EXHAUSTION`` reason covers both raise sites in
|
||||
``generate/stream.py``:
|
||||
|
||||
1. Pre-walk: the region's allowed-index intersection with the
|
||||
candidate set is empty before any step ran. ``step_index = -1``
|
||||
and ``rejected_attempts = ()`` distinguish this site — no
|
||||
inner-loop rejections were issued; the region was already empty.
|
||||
|
||||
2. In-walk: at some step, every candidate in the admissible set was
|
||||
rejected by ``check_transition`` at the configured threshold.
|
||||
``step_index >= 0`` and ``rejected_attempts`` records the tried
|
||||
(index, word, score) triples in attempt order.
|
||||
|
||||
Splitting these into separate reasons can wait for Phase 4, when
|
||||
rotor-frame refusal introduces a third structurally distinct mode
|
||||
(ADR-0025).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from enum import Enum, unique
|
||||
|
||||
|
||||
@unique
|
||||
class RefusalReason(Enum):
|
||||
"""Machine-readable refusal taxonomy.
|
||||
|
||||
The string value is what flows into ``trace_hash`` payloads when
|
||||
refusal materialisation is wired through a future ADR. Stable
|
||||
string values are part of the replay contract.
|
||||
"""
|
||||
|
||||
INNER_LOOP_EXHAUSTION = "inner_loop_exhaustion"
|
||||
|
||||
|
||||
class InnerLoopExhaustion(ValueError):
|
||||
"""Honest refusal raised by the generation walk.
|
||||
|
||||
Subclasses ``ValueError`` so pre-Phase-2 ``except ValueError``
|
||||
handlers in ``chat/runtime.py``, ``evals/...``, and tests still
|
||||
catch it without modification.
|
||||
|
||||
Attributes
|
||||
----------
|
||||
reason : RefusalReason
|
||||
Machine-readable taxonomy code. Phase 2 uses a single value;
|
||||
Phase 4 (rotor frame admissibility) is expected to add more.
|
||||
region_label : str
|
||||
The ``AdmissibilityRegion.label`` that produced the refusal —
|
||||
the operator-visible identifier of which constraint blocked
|
||||
propagation.
|
||||
step_index : int
|
||||
Index of the step at which the inner loop exhausted. ``-1``
|
||||
marks the pre-walk site (region intersection empty before any
|
||||
step ran); non-negative values identify the in-walk site.
|
||||
rejected_attempts : tuple[tuple[int, str, float], ...]
|
||||
Ordered record of ``(candidate_index, word, score)`` triples
|
||||
that the inner-loop check rejected at the step that failed.
|
||||
Empty tuple for the pre-walk site.
|
||||
|
||||
``str(exc)`` returns the human-readable message — preserving the
|
||||
Phase 1 exception message contract for callers that only inspected
|
||||
``str(e)``. Instances are logically immutable: attributes are set
|
||||
once in ``__init__`` and should not be reassigned.
|
||||
"""
|
||||
|
||||
__slots__ = ("reason", "region_label", "step_index", "rejected_attempts")
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
*,
|
||||
reason: RefusalReason,
|
||||
region_label: str,
|
||||
step_index: int,
|
||||
rejected_attempts: tuple[tuple[int, str, float], ...] = (),
|
||||
message: str | None = None,
|
||||
) -> None:
|
||||
self.reason = reason
|
||||
self.region_label = region_label
|
||||
self.step_index = step_index
|
||||
self.rejected_attempts = tuple(rejected_attempts)
|
||||
super().__init__(message if message is not None else self._default_message())
|
||||
|
||||
def _default_message(self) -> str:
|
||||
if self.step_index < 0:
|
||||
return (
|
||||
f"AdmissibilityRegion[{self.region_label}] left no walk candidates."
|
||||
)
|
||||
return (
|
||||
f"AdmissibilityRegion[{self.region_label}] inner-loop "
|
||||
f"rejected all candidates at step {self.step_index}."
|
||||
)
|
||||
|
|
@ -27,6 +27,7 @@ from generate.admissibility import (
|
|||
filter_candidates,
|
||||
)
|
||||
from generate.attention import AttentionOperator
|
||||
from generate.exhaustion import InnerLoopExhaustion, RefusalReason
|
||||
from generate.result import GenerationResult
|
||||
from generate.salience import SalienceOperator
|
||||
|
||||
|
|
@ -339,8 +340,18 @@ def generate(
|
|||
if region is not None and not region.is_unconstrained():
|
||||
candidate_indices = filter_candidates(region, candidate_indices)
|
||||
if candidate_indices is not None and len(candidate_indices) == 0:
|
||||
raise ValueError(
|
||||
f"AdmissibilityRegion[{region.label}] left no walk candidates."
|
||||
# ADR-0024 Phase 2 — pre-walk exhaustion site. The region's
|
||||
# allowed-index intersection with the candidate set is empty
|
||||
# before any step ran. ``step_index = -1`` and
|
||||
# ``rejected_attempts = ()`` distinguish this site from the
|
||||
# in-walk exhaustion site below; no inner-loop rejections were
|
||||
# issued because the region was already empty. Subclasses
|
||||
# ValueError so existing handlers continue to catch it.
|
||||
raise InnerLoopExhaustion(
|
||||
reason=RefusalReason.INNER_LOOP_EXHAUSTION,
|
||||
region_label=region.label,
|
||||
step_index=-1,
|
||||
rejected_attempts=(),
|
||||
)
|
||||
candidates_used = None if candidate_indices is None else len(candidate_indices)
|
||||
admissibility_trace: list[AdmissibilityTraceStep] = []
|
||||
|
|
@ -417,17 +428,29 @@ def generate(
|
|||
if int(word_idx) in step_exclude:
|
||||
# Selector returned the same exhausted candidate — no
|
||||
# further admissible destinations. Honest refusal.
|
||||
raise ValueError(
|
||||
f"AdmissibilityRegion[{effective_region_label}] inner-loop "
|
||||
f"rejected all candidates at step {step_index}."
|
||||
# ADR-0024 Phase 2 — in-walk exhaustion site; carries the
|
||||
# ordered ``rejected_attempts`` accumulated this step so
|
||||
# downstream layers can read refusal evidence without
|
||||
# re-parsing the exception message.
|
||||
raise InnerLoopExhaustion(
|
||||
reason=RefusalReason.INNER_LOOP_EXHAUSTION,
|
||||
region_label=effective_region_label,
|
||||
step_index=step_index,
|
||||
rejected_attempts=tuple(rejected_attempts),
|
||||
)
|
||||
step_exclude.add(int(word_idx))
|
||||
else:
|
||||
# max_attempts exhausted without break — every admissible
|
||||
# candidate was rejected by the inner-loop threshold.
|
||||
raise ValueError(
|
||||
f"AdmissibilityRegion[{effective_region_label}] inner-loop "
|
||||
f"rejected all candidates at step {step_index}."
|
||||
# Same refusal shape as the same-candidate-loop site above;
|
||||
# both are structurally "inner-loop produced no admissible
|
||||
# candidate at this step". Splitting into separate reasons
|
||||
# can wait for Phase 4 (rotor frame, ADR-0025).
|
||||
raise InnerLoopExhaustion(
|
||||
reason=RefusalReason.INNER_LOOP_EXHAUSTION,
|
||||
region_label=effective_region_label,
|
||||
step_index=step_index,
|
||||
rejected_attempts=tuple(rejected_attempts),
|
||||
)
|
||||
|
||||
tokens.append(_articulate(vocab, word))
|
||||
|
|
|
|||
315
tests/test_refusal_contract.py
Normal file
315
tests/test_refusal_contract.py
Normal file
|
|
@ -0,0 +1,315 @@
|
|||
"""ADR-0024 Phase 2 — refusal contract.
|
||||
|
||||
Pins the typed-refusal contract introduced in Phase 2:
|
||||
|
||||
1. ``generate.stream`` raises ``InnerLoopExhaustion`` (not bare
|
||||
``ValueError``) at both inner-loop exhaustion sites.
|
||||
2. ``InnerLoopExhaustion`` is a ``ValueError`` subclass, so every
|
||||
pre-Phase-2 ``except ValueError`` handler still catches it.
|
||||
3. Each raise carries evidence: ``reason``, ``region_label``,
|
||||
``step_index``, and (for in-walk site) ``rejected_attempts``.
|
||||
4. Pre-walk site uses ``step_index == -1`` and empty
|
||||
``rejected_attempts``; in-walk site uses ``step_index >= 0`` and
|
||||
non-empty ``rejected_attempts``.
|
||||
5. ``compute_trace_hash`` accepts ``refusal_reason`` and:
|
||||
a. byte-identical hash when ``refusal_reason == ""`` (default),
|
||||
relative to the same call without the kwarg → preserves
|
||||
pre-Phase-2 hashes for non-refused turns;
|
||||
b. different hash when ``refusal_reason`` is set to a non-empty
|
||||
value, so refusal is load-bearing in replay equality once
|
||||
materialised.
|
||||
6. ``CognitiveTurnResult`` exposes ``refusal_reason`` defaulting to
|
||||
"" so call sites can be ignorant of refusal materialisation
|
||||
during the Phase 2 → Phase ? transition window.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from core.cognition.result import CognitiveTurnResult
|
||||
from core.cognition.trace import compute_trace_hash
|
||||
from field.state import FieldState
|
||||
from generate.admissibility import AdmissibilityRegion, RegionSource
|
||||
from generate.exhaustion import InnerLoopExhaustion, RefusalReason
|
||||
from generate.stream import generate
|
||||
|
||||
|
||||
# --------------------------------------------------------------------
|
||||
# Stub vocab / persona — mirrors test_inner_loop_admissibility.py
|
||||
# --------------------------------------------------------------------
|
||||
|
||||
|
||||
class _ControllableVocab:
|
||||
def __init__(self, *, words, preference, versor_signs):
|
||||
self._words = words
|
||||
self._preference = preference
|
||||
self._versors = []
|
||||
for sign in versor_signs:
|
||||
v = np.zeros(32, dtype=np.float32)
|
||||
v[1] = float(sign)
|
||||
self._versors.append(v)
|
||||
|
||||
def __len__(self):
|
||||
return len(self._words)
|
||||
|
||||
def nearest(self, F, exclude_idx=-1, exclude_indices=None, candidate_indices=None):
|
||||
blocked = set(exclude_indices or ())
|
||||
if candidate_indices is not None:
|
||||
allowed = {int(i) for i in candidate_indices}
|
||||
else:
|
||||
allowed = set(range(len(self._words)))
|
||||
for idx in self._preference:
|
||||
if idx == exclude_idx or idx in blocked or idx not in allowed:
|
||||
continue
|
||||
return self._words[idx], idx
|
||||
raise ValueError("No candidate word available after exclusions.")
|
||||
|
||||
def get_versor_at(self, idx):
|
||||
return self._versors[idx]
|
||||
|
||||
def index_of(self, word):
|
||||
try:
|
||||
return self._words.index(word)
|
||||
except ValueError:
|
||||
raise KeyError(word)
|
||||
|
||||
|
||||
class _IdentityPersona:
|
||||
def apply(self, F):
|
||||
return F
|
||||
|
||||
|
||||
def _initial_state():
|
||||
F = np.zeros(32, dtype=np.float32)
|
||||
F[1] = 1.0
|
||||
return FieldState(F=F, node=0, step=0)
|
||||
|
||||
|
||||
def _positive_blade_region(allowed, label="adr0024-phase2-test"):
|
||||
blade = np.zeros(32, dtype=np.float32)
|
||||
blade[1] = 1.0
|
||||
return AdmissibilityRegion(
|
||||
allowed_indices=np.asarray(allowed, dtype=np.int64),
|
||||
relation_blade=blade,
|
||||
source=RegionSource.RELATION,
|
||||
label=label,
|
||||
)
|
||||
|
||||
|
||||
def _empty_region(label="adr0024-phase2-empty"):
|
||||
"""Region whose allowed-index intersection with the candidate set
|
||||
is empty before any step runs — exercises the pre-walk raise site."""
|
||||
blade = np.zeros(32, dtype=np.float32)
|
||||
blade[1] = 1.0
|
||||
return AdmissibilityRegion(
|
||||
allowed_indices=np.asarray([], dtype=np.int64),
|
||||
relation_blade=blade,
|
||||
source=RegionSource.RELATION,
|
||||
label=label,
|
||||
)
|
||||
|
||||
|
||||
# --------------------------------------------------------------------
|
||||
# Property 1 + 2 + 3 + 4: exception type, subclassing, evidence
|
||||
# --------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestExceptionIsValueErrorSubclass:
|
||||
"""Pre-Phase-2 handlers caught ValueError; Phase 2 must remain
|
||||
backwards-compatible."""
|
||||
|
||||
def test_inner_loop_exhaustion_is_value_error(self) -> None:
|
||||
exc = InnerLoopExhaustion(
|
||||
reason=RefusalReason.INNER_LOOP_EXHAUSTION,
|
||||
region_label="r",
|
||||
step_index=-1,
|
||||
)
|
||||
assert isinstance(exc, ValueError)
|
||||
|
||||
def test_in_walk_exhaustion_caught_as_value_error(self) -> None:
|
||||
# Both admissible candidates rejected — exercises the in-walk
|
||||
# raise site at the loop's ``else`` branch.
|
||||
vocab = _ControllableVocab(
|
||||
words=["seed", "alpha", "beta"],
|
||||
preference=[1, 2],
|
||||
versor_signs=[+1.0, -1.0, -1.0],
|
||||
)
|
||||
with pytest.raises(ValueError):
|
||||
generate(
|
||||
_initial_state(),
|
||||
vocab,
|
||||
_IdentityPersona(),
|
||||
max_tokens=1,
|
||||
region=_positive_blade_region((1, 2)),
|
||||
inner_loop_admissibility=True,
|
||||
)
|
||||
|
||||
def test_pre_walk_exhaustion_caught_as_value_error(self) -> None:
|
||||
vocab = _ControllableVocab(
|
||||
words=["seed", "alpha", "beta"],
|
||||
preference=[1, 2],
|
||||
versor_signs=[+1.0, +1.0, +1.0],
|
||||
)
|
||||
with pytest.raises(ValueError):
|
||||
generate(
|
||||
_initial_state(),
|
||||
vocab,
|
||||
_IdentityPersona(),
|
||||
max_tokens=1,
|
||||
region=_empty_region(),
|
||||
inner_loop_admissibility=True,
|
||||
)
|
||||
|
||||
|
||||
class TestInWalkExhaustionCarriesEvidence:
|
||||
def test_typed_exception_with_full_evidence(self) -> None:
|
||||
vocab = _ControllableVocab(
|
||||
words=["seed", "alpha", "beta"],
|
||||
preference=[1, 2],
|
||||
versor_signs=[+1.0, -1.0, -1.0],
|
||||
)
|
||||
with pytest.raises(InnerLoopExhaustion) as excinfo:
|
||||
generate(
|
||||
_initial_state(),
|
||||
vocab,
|
||||
_IdentityPersona(),
|
||||
max_tokens=1,
|
||||
region=_positive_blade_region((1, 2), label="in-walk-label"),
|
||||
inner_loop_admissibility=True,
|
||||
)
|
||||
exc = excinfo.value
|
||||
assert exc.reason is RefusalReason.INNER_LOOP_EXHAUSTION
|
||||
assert exc.region_label == "in-walk-label"
|
||||
# In-walk site fires at step_index >= 0.
|
||||
assert exc.step_index >= 0
|
||||
# Both admissible candidates rejected ⇒ rejected_attempts
|
||||
# records each in attempt order.
|
||||
assert len(exc.rejected_attempts) >= 1
|
||||
for idx, word, score in exc.rejected_attempts:
|
||||
assert isinstance(idx, int)
|
||||
assert isinstance(word, str)
|
||||
assert isinstance(score, float)
|
||||
# Message preserves the Phase 1 contract — region label is
|
||||
# embedded so existing ``match=region.label`` callers still pass.
|
||||
assert "in-walk-label" in str(exc)
|
||||
|
||||
|
||||
class TestPreWalkExhaustionCarriesEvidence:
|
||||
def test_typed_exception_with_pre_walk_sentinel(self) -> None:
|
||||
vocab = _ControllableVocab(
|
||||
words=["seed", "alpha", "beta"],
|
||||
preference=[1, 2],
|
||||
versor_signs=[+1.0, +1.0, +1.0],
|
||||
)
|
||||
with pytest.raises(InnerLoopExhaustion) as excinfo:
|
||||
generate(
|
||||
_initial_state(),
|
||||
vocab,
|
||||
_IdentityPersona(),
|
||||
max_tokens=1,
|
||||
region=_empty_region(label="pre-walk-label"),
|
||||
inner_loop_admissibility=True,
|
||||
)
|
||||
exc = excinfo.value
|
||||
assert exc.reason is RefusalReason.INNER_LOOP_EXHAUSTION
|
||||
assert exc.region_label == "pre-walk-label"
|
||||
# Pre-walk site is marked with the -1 sentinel.
|
||||
assert exc.step_index == -1
|
||||
# No inner-loop rejections were issued — the intersection was
|
||||
# empty before any step ran.
|
||||
assert exc.rejected_attempts == ()
|
||||
assert "pre-walk-label" in str(exc)
|
||||
|
||||
|
||||
class TestPreWalkSiteFiresEvenWithoutInnerLoop:
|
||||
"""The pre-walk site predates inner-loop wiring (ADR-0023) — it
|
||||
still must raise the typed exception so refusal evidence is
|
||||
uniform whether or not the inner loop is on."""
|
||||
|
||||
def test_pre_walk_raise_typed_when_inner_loop_off(self) -> None:
|
||||
vocab = _ControllableVocab(
|
||||
words=["seed", "alpha", "beta"],
|
||||
preference=[1, 2],
|
||||
versor_signs=[+1.0, +1.0, +1.0],
|
||||
)
|
||||
with pytest.raises(InnerLoopExhaustion) as excinfo:
|
||||
generate(
|
||||
_initial_state(),
|
||||
vocab,
|
||||
_IdentityPersona(),
|
||||
max_tokens=1,
|
||||
region=_empty_region(label="pre-walk-no-inner"),
|
||||
inner_loop_admissibility=False,
|
||||
)
|
||||
assert excinfo.value.step_index == -1
|
||||
assert excinfo.value.rejected_attempts == ()
|
||||
|
||||
|
||||
# --------------------------------------------------------------------
|
||||
# Property 5: compute_trace_hash determinism contract
|
||||
# --------------------------------------------------------------------
|
||||
|
||||
|
||||
_BASE_HASH_KWARGS = dict(
|
||||
input_text="hello",
|
||||
filtered_tokens=("hello",),
|
||||
surface="hi",
|
||||
walk_surface="hi",
|
||||
articulation_surface="hi",
|
||||
dialogue_role="assert",
|
||||
versor_condition=0.0,
|
||||
vault_hits=0,
|
||||
intent_tag="declare",
|
||||
)
|
||||
|
||||
|
||||
class TestTraceHashRefusalReasonFold:
|
||||
def test_default_refusal_reason_preserves_legacy_hash(self) -> None:
|
||||
"""A non-refused turn (refusal_reason="") must produce the
|
||||
same trace_hash as a call that omits the kwarg entirely. This
|
||||
is what protects every pre-Phase-2 turn hash from drift."""
|
||||
legacy = compute_trace_hash(**_BASE_HASH_KWARGS)
|
||||
with_kwarg = compute_trace_hash(refusal_reason="", **_BASE_HASH_KWARGS)
|
||||
assert legacy == with_kwarg
|
||||
|
||||
def test_non_empty_refusal_reason_changes_hash(self) -> None:
|
||||
"""Once a refusal is materialised, the reason becomes
|
||||
load-bearing — different reasons must yield different hashes."""
|
||||
base = compute_trace_hash(**_BASE_HASH_KWARGS)
|
||||
with_reason = compute_trace_hash(
|
||||
refusal_reason=RefusalReason.INNER_LOOP_EXHAUSTION.value,
|
||||
**_BASE_HASH_KWARGS,
|
||||
)
|
||||
assert base != with_reason
|
||||
|
||||
def test_same_refusal_reason_is_stable(self) -> None:
|
||||
"""Determinism: identical inputs ⇒ identical hash."""
|
||||
h1 = compute_trace_hash(
|
||||
refusal_reason=RefusalReason.INNER_LOOP_EXHAUSTION.value,
|
||||
**_BASE_HASH_KWARGS,
|
||||
)
|
||||
h2 = compute_trace_hash(
|
||||
refusal_reason=RefusalReason.INNER_LOOP_EXHAUSTION.value,
|
||||
**_BASE_HASH_KWARGS,
|
||||
)
|
||||
assert h1 == h2
|
||||
|
||||
|
||||
# --------------------------------------------------------------------
|
||||
# Property 6: CognitiveTurnResult exposes refusal_reason with safe default
|
||||
# --------------------------------------------------------------------
|
||||
|
||||
|
||||
class TestCognitiveTurnResultRefusalField:
|
||||
def test_refusal_reason_defaults_to_empty(self) -> None:
|
||||
# We do not need to construct a full CognitiveTurnResult here;
|
||||
# we only need to confirm the dataclass declares the field with
|
||||
# the empty-string default. Use dataclasses.fields() so we
|
||||
# don't have to assemble all the other required fields.
|
||||
import dataclasses
|
||||
|
||||
names = {f.name: f for f in dataclasses.fields(CognitiveTurnResult)}
|
||||
assert "refusal_reason" in names
|
||||
assert names["refusal_reason"].default == ""
|
||||
Loading…
Reference in a new issue