f3cc408f82
Closes three audit gaps left by the ADR-0035→ADR-0038 pack-layer surface: 1. TurnVerdicts bundle (chat/verdicts.py) — frozen dataclass aggregating identity_score + safety_verdict + ethics_verdict + refusal_emitted + hedge_injected. Attached to both ChatResponse.verdicts and TurnEvent.verdicts. Fields typed as object for the same module-coupling reason as TurnEvent.safety_verdict. 2. Stub-path TurnEvent emission — _stub_response accepts optional tokens kwarg and appends a TurnEvent to turn_log when invoked from a real turn. Audit consumers can now iterate turn_log end-to-end without missing stub paths. Defensive call sites (correct() fallback) bypass the append by omitting tokens. 3. refusal_emitted / hedge_injected flags — runtime tracks whether it actually mutated the surface this turn. hedge_injected uses idempotent-on-prefix semantics (True iff the runtime ADDED a hedge, not iff a hedge happens to be present). Test-pattern note: previous "gate on rt.turn_log to detect main vs stub" pattern is now broken; updated to gate on walk_surface == _UNKNOWN_DOMAIN_SURFACE. One existing hedge-injection test gate updated accordingly. Back-compat: ADR-0035→0038 per-field accessors (response.safety_verdict, etc.) still work. New consumers should read response.verdicts. Files: - chat/verdicts.py (new) — TurnVerdicts dataclass - chat/runtime.py — _stub_response tokens kwarg + stub TurnEvent append + hedge_injected tracking + bundle construction - core/physics/identity.py — TurnEvent.verdicts: object = None - tests/test_turn_verdicts_bundle.py (new) — 16 tests - tests/test_hedge_injection.py — gate fix for stub detection - docs/decisions/ADR-0039-audit-completeness.md (new) Verification: - Combined pack-layer suite: 170 green (was 154 after ADR-0038) - CLI suites unchanged: smoke 67, runtime 19, cognition 121 - core eval cognition: intent 100%, versor_closure 100% (baseline)
242 lines
8.7 KiB
Python
242 lines
8.7 KiB
Python
"""ADR-0039 — audit completeness: TurnVerdicts bundle, stub-path
|
|
TurnEvent emission, hedge_injected signal.
|
|
|
|
The bundle replaces the per-field correlation pattern downstream
|
|
audit consumers had to do (read identity_score, safety_verdict,
|
|
ethics_verdict, then infer remediation from surface text). The
|
|
remediation flags ``refusal_emitted`` and ``hedge_injected`` make
|
|
the runtime's decisions auditable without text inspection.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from dataclasses import replace
|
|
|
|
from chat.runtime import ChatRuntime, _UNKNOWN_DOMAIN_SURFACE
|
|
from chat.verdicts import TurnVerdicts
|
|
from core.config import RuntimeConfig
|
|
from packs.ethics.check import EthicsCheckResult, EthicsVerdict
|
|
from packs.safety.check import SafetyCheckResult, SafetyVerdict
|
|
|
|
|
|
# ---------- TurnVerdicts shape ----------
|
|
|
|
|
|
class TestTurnVerdictsShape:
|
|
def test_construct_with_required_fields(self) -> None:
|
|
v = TurnVerdicts(
|
|
identity_score=None,
|
|
safety_verdict=None,
|
|
ethics_verdict=None,
|
|
refusal_emitted=False,
|
|
hedge_injected=False,
|
|
)
|
|
assert v.refusal_emitted is False
|
|
assert v.hedge_injected is False
|
|
|
|
def test_is_frozen(self) -> None:
|
|
v = TurnVerdicts(
|
|
identity_score=None,
|
|
safety_verdict=None,
|
|
ethics_verdict=None,
|
|
refusal_emitted=False,
|
|
hedge_injected=False,
|
|
)
|
|
try:
|
|
v.refusal_emitted = True # type: ignore[misc]
|
|
except (AttributeError, Exception):
|
|
return
|
|
else:
|
|
raise AssertionError("TurnVerdicts should be frozen")
|
|
|
|
|
|
# ---------- ChatResponse and TurnEvent carry the bundle ----------
|
|
|
|
|
|
class TestBundleAttachment:
|
|
def test_chat_response_carries_verdicts_bundle(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
resp = rt.chat("light is")
|
|
assert isinstance(resp.verdicts, TurnVerdicts)
|
|
# Bundle's individual verdicts match the per-field versions.
|
|
assert resp.verdicts.safety_verdict is resp.safety_verdict
|
|
assert resp.verdicts.ethics_verdict is resp.ethics_verdict
|
|
|
|
def test_turn_event_carries_verdicts_bundle(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
rt.chat("light is")
|
|
assert rt.turn_log, "stub path should now append TurnEvent (ADR-0039)"
|
|
event = rt.turn_log[-1]
|
|
assert isinstance(event.verdicts, TurnVerdicts)
|
|
|
|
|
|
# ---------- stub-path TurnEvent emission (ADR-0039) ----------
|
|
|
|
|
|
class TestStubPathTurnEvent:
|
|
def test_stub_path_appends_turn_event(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
before = len(rt.turn_log)
|
|
resp = rt.chat("light is")
|
|
# Cold start with no vault hits → stub path fires.
|
|
assert resp.walk_surface == _UNKNOWN_DOMAIN_SURFACE
|
|
assert len(rt.turn_log) == before + 1
|
|
|
|
def test_stub_event_records_unknown_walk_surface(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
rt.chat("light is")
|
|
event = rt.turn_log[-1]
|
|
assert event.walk_surface == _UNKNOWN_DOMAIN_SURFACE
|
|
assert event.articulation_surface == _UNKNOWN_DOMAIN_SURFACE
|
|
|
|
def test_stub_event_carries_input_tokens(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
rt.chat("light is")
|
|
event = rt.turn_log[-1]
|
|
assert event.input_tokens, "stub TurnEvent must record input tokens"
|
|
|
|
def test_stub_event_identity_score_is_none(self) -> None:
|
|
"""No reasoning trajectory ran on stub path, so no
|
|
IdentityScore is computed."""
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
rt.chat("light is")
|
|
event = rt.turn_log[-1]
|
|
assert event.identity_score is None
|
|
|
|
def test_stub_event_versor_condition_recorded(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
rt.chat("light is")
|
|
event = rt.turn_log[-1]
|
|
assert isinstance(event.versor_condition, float)
|
|
|
|
|
|
# ---------- refusal_emitted flag ----------
|
|
|
|
|
|
class TestRefusalEmittedFlag:
|
|
def test_no_violation_flag_is_false(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
resp = rt.chat("light is")
|
|
assert resp.verdicts.refusal_emitted is False
|
|
|
|
def test_safety_violation_sets_flag(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
|
|
def _failing(ctx): # noqa: ANN001
|
|
return SafetyCheckResult(
|
|
boundary_id="preserve_versor_closure",
|
|
upheld=False,
|
|
reason="forced",
|
|
runtime_checkable=True,
|
|
)
|
|
|
|
rt.safety_check.register("preserve_versor_closure", _failing)
|
|
resp = rt.chat("light is")
|
|
assert resp.verdicts.refusal_emitted is True
|
|
# And the flag matches the surface state.
|
|
assert resp.verdicts.refusal_emitted == (resp.surface != resp.walk_surface)
|
|
|
|
def test_flag_appears_on_turn_event_too(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
|
|
def _failing(ctx): # noqa: ANN001
|
|
return SafetyCheckResult(
|
|
boundary_id="preserve_versor_closure",
|
|
upheld=False,
|
|
reason="forced",
|
|
runtime_checkable=True,
|
|
)
|
|
|
|
rt.safety_check.register("preserve_versor_closure", _failing)
|
|
rt.chat("light is")
|
|
event = rt.turn_log[-1]
|
|
assert event.verdicts.refusal_emitted is True
|
|
|
|
|
|
# ---------- hedge_injected flag ----------
|
|
|
|
|
|
class TestHedgeInjectedFlag:
|
|
def test_no_opt_in_flag_is_false(self) -> None:
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
resp = rt.chat("light is")
|
|
assert resp.verdicts.hedge_injected is False
|
|
|
|
def test_stub_path_never_sets_hedge_injected(self) -> None:
|
|
"""Even with opted-in hedge_commitments and a forced
|
|
violation, the stub path must report hedge_injected=False
|
|
(stub never hedges — the unknown-domain marker is already
|
|
a disclosure)."""
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
rt.ethics_pack = replace(
|
|
rt.ethics_pack,
|
|
hedge_commitments=frozenset({"acknowledge_uncertainty"}),
|
|
)
|
|
|
|
def _failing_ethics(ctx): # noqa: ANN001
|
|
return EthicsCheckResult(
|
|
commitment_id="acknowledge_uncertainty",
|
|
upheld=False,
|
|
reason="forced",
|
|
runtime_checkable=True,
|
|
)
|
|
|
|
rt.ethics_check.register("acknowledge_uncertainty", _failing_ethics)
|
|
resp = rt.chat("light is")
|
|
if resp.walk_surface == _UNKNOWN_DOMAIN_SURFACE:
|
|
assert resp.verdicts.hedge_injected is False
|
|
|
|
|
|
# ---------- mutual exclusion refusal vs hedge in the bundle ----------
|
|
|
|
|
|
class TestRemediationMutualExclusion:
|
|
def test_refusal_and_hedge_never_both_true(self) -> None:
|
|
"""The runtime contract: refusal supersedes hedge. When both
|
|
a runtime-checkable safety violation AND an opted-in ethics
|
|
hedge would fire, refusal wins and hedge_injected stays
|
|
False."""
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
rt.ethics_pack = replace(
|
|
rt.ethics_pack,
|
|
hedge_commitments=frozenset({"acknowledge_uncertainty"}),
|
|
)
|
|
|
|
def _failing_safety(ctx): # noqa: ANN001
|
|
return SafetyCheckResult(
|
|
boundary_id="preserve_versor_closure",
|
|
upheld=False,
|
|
reason="forced",
|
|
runtime_checkable=True,
|
|
)
|
|
|
|
def _failing_ethics(ctx): # noqa: ANN001
|
|
return EthicsCheckResult(
|
|
commitment_id="acknowledge_uncertainty",
|
|
upheld=False,
|
|
reason="forced",
|
|
runtime_checkable=True,
|
|
)
|
|
|
|
rt.safety_check.register("preserve_versor_closure", _failing_safety)
|
|
rt.ethics_check.register("acknowledge_uncertainty", _failing_ethics)
|
|
resp = rt.chat("light is")
|
|
assert resp.verdicts.refusal_emitted is True
|
|
assert resp.verdicts.hedge_injected is False
|
|
|
|
|
|
# ---------- response and event bundles agree ----------
|
|
|
|
|
|
class TestBundleConsistency:
|
|
def test_response_and_event_bundle_are_consistent(self) -> None:
|
|
"""The TurnVerdicts on ChatResponse and TurnEvent for the same
|
|
turn carry the same remediation flags and reference the same
|
|
underlying verdicts."""
|
|
rt = ChatRuntime(config=RuntimeConfig())
|
|
resp = rt.chat("light is")
|
|
event = rt.turn_log[-1]
|
|
assert resp.verdicts.refusal_emitted == event.verdicts.refusal_emitted
|
|
assert resp.verdicts.hedge_injected == event.verdicts.hedge_injected
|
|
assert resp.verdicts.safety_verdict is event.verdicts.safety_verdict
|
|
assert resp.verdicts.ethics_verdict is event.verdicts.ethics_verdict
|