diff --git a/chat/runtime.py b/chat/runtime.py index 361c9c63..45bdd9a9 100644 --- a/chat/runtime.py +++ b/chat/runtime.py @@ -19,6 +19,7 @@ from core.physics.identity import ( TurnEvent, ) from packs.identity.loader import load_identity_manifold +from packs.safety.check import SafetyCheck from packs.safety.loader import load_safety_pack from field.state import FieldState from generate.articulation import ArticulationPlan, realize @@ -245,6 +246,12 @@ class ChatRuntime: fatigue_index=0.0, ) self._identity_check = IdentityCheck() + # ADR-0032 — structural safety surface. Observational at v1: + # ChatRuntime exposes ``safety_check`` for callers (audit / + # logging / future enforcement), but does not auto-invoke it in + # the turn loop. Wiring violations into refusal paths is a + # future ADR. + self.safety_check = SafetyCheck() self.turn_log: List[TurnEvent] = [] self._correction_pass = CorrectionPass() self._last_valence: float = 0.0 diff --git a/docs/decisions/ADR-0032-safety-check-surface.md b/docs/decisions/ADR-0032-safety-check-surface.md new file mode 100644 index 00000000..cebf8bdd --- /dev/null +++ b/docs/decisions/ADR-0032-safety-check-surface.md @@ -0,0 +1,128 @@ +# ADR-0032: SafetyCheck — Structural Surface for Safety-Pack Boundaries + +**Status:** Accepted (2026-05-17) +**Author:** Joshua Shay + planner pass +**Companion docs:** [`safety_packs.md`](../safety_packs.md), [`ADR-0029-safety-packs.md`](ADR-0029-safety-packs.md) + +## Context + +[ADR-0029](ADR-0029-safety-packs.md) established the safety pack as an always-loaded, never-replaceable, fail-closed sibling to identity packs. The pack contributes `boundary_ids` to the runtime manifold; identity packs may add to that set but never remove from it. + +What ADR-0029 did *not* establish was a centralized surface for *checking* the boundaries at runtime. Today, the boundaries are enforced (where they are runtime-enforceable) by scattered call sites: source allowlists in the forge, typed exceptions in `generate/exhaustion.py`, the versor-condition halt in `formation/runner.py` and elsewhere. The boundary ids exist as labels; their enforcement is implicit, distributed, and hard to audit. + +`IdentityCheck` (ADR-0010) provides a clean precedent: a structural surface that takes a trajectory and a manifold, produces an `IdentityScore` with `deviation_axes`, and lets downstream callers (assembler, refusal paths, logging) decide what to do with the verdict. The natural follow-up is a parallel surface for safety boundaries: `SafetyCheck`. + +But the parallel is shape-only, not mechanism. Identity check is geometric — projection onto value axes. Safety boundaries are propositional — each is a different kind of constraint, and several are not even runtime-checkable. + +## Decision + +SafetyCheck is a registry of named predicates, one per boundary id, with sensible defaults for the five v1 boundaries. It is **observational**: it produces a verdict; it does not refuse. Wiring verdicts into refusal paths is a future ADR. + +### What's runtime-checkable + +| Boundary | Predicate evaluates | Runtime-checkable? | +|---|---|---| +| `preserve_versor_closure` | `field_state.versor_condition < 1.0e-6` | Yes — direct attribute read | +| `no_fabricated_source` | `cited_source_shas ⊆ allowed_source_shas` | Yes — set membership | +| `no_silent_correction` | `last_refusal_was_typed` flag | Yes — bookkeeping by the runtime | +| `no_identity_override` | `identity_manifold_hash_before == identity_manifold_hash_after` | Yes — hash comparison | +| `no_hot_path_repair` | code-path constraint; no runtime evidence available | **No** — static-analysis + code-review boundary | + +The last row is the architecturally interesting one: `no_hot_path_repair` is a *code-path* boundary. It forbids normalization / drift-repair operators in `field/propagate.py`, `generate/stream.py`, and `vault/store.py`. There is no runtime evidence that could prove or disprove it. The honest answer is `runtime_checkable=False`, with `upheld=True` and a clear `reason` explaining that enforcement lives in static analysis and code review. + +A predicate that *silently* reported `upheld=True` for `no_hot_path_repair` would be a small lie, exactly the kind of thing CLAUDE.md forbids ("no silent correction"). The structural surface acknowledges what it cannot judge. + +### API shape + +```python +@dataclass(frozen=True, slots=True) +class SafetyContext: + field_state: object | None = None # for versor closure + versor_halt_threshold: float = 1.0e-6 + cited_source_shas: frozenset[str] = frozenset() + allowed_source_shas: frozenset[str] = frozenset() + last_refusal_was_typed: bool = True + identity_manifold_hash_before: str = "" + identity_manifold_hash_after: str = "" + +@dataclass(frozen=True, slots=True) +class SafetyCheckResult: + boundary_id: str + upheld: bool + reason: str + runtime_checkable: bool + evidence: tuple[tuple[str, str], ...] = () + +@dataclass(frozen=True, slots=True) +class SafetyVerdict: + pack_id: str + results: tuple[SafetyCheckResult, ...] # lex order on boundary_id + upheld: bool # all results upheld + violated_boundaries: frozenset[str] + runtime_checkable_count: int + +class SafetyCheck: + def __init__(self, predicates: Mapping[str, SafetyPredicate] | None = None) -> None: ... + def register(self, boundary_id: str, predicate: SafetyPredicate) -> None: ... + def check(self, ctx: SafetyContext, safety_pack: SafetyPack) -> SafetyVerdict: ... +``` + +Every field on `SafetyContext` is optional. Predicates over fields the caller didn't populate default to `upheld=True, runtime_checkable=False`. The interpretation is deliberate: SafetyCheck is observational, so absence of evidence is not evidence of violation. This keeps the surface composable — callers populate whatever they have access to, without crashing on what they don't. + +### Unknown-boundary behavior + +When a pack declares a boundary id for which no predicate is registered, the verdict records `upheld=True, runtime_checkable=False, reason="no predicate registered for boundary"`. This lets downstream deployments add custom boundaries without crashing the runtime, while still surfacing in audit that the runtime had no opinion on them. + +A future production deployment can choose to treat unknown-but-declared boundaries more strictly (e.g., `require_runtime_checkable=True` flag that turns unknowns into errors). That's a deployment policy decision, not a surface-shape decision. + +### Custom predicate registration + +```python +check = SafetyCheck() +check.register("my_robotics_safety_boundary", my_predicate) +``` + +A robotics deployment ships a custom safety pack with deployment-specific boundary ids and a `SafetyCheck` constructed with predicates for each. The five default predicates remain registered unless explicitly replaced. + +### Defensive: predicate-result rebinding + +If a registered predicate returns a `SafetyCheckResult` whose `boundary_id` field doesn't match the boundary it was registered under, `SafetyCheck.check` rebinds the result to the correct boundary id. This is defensive — a buggy predicate should not silently associate a verdict with the wrong boundary in audit logs. + +### ChatRuntime integration + +`ChatRuntime` instantiates `self.safety_check = SafetyCheck()` alongside `self._identity_check`. The turn loop **does not** auto-invoke it at v1. Callers (audit / logging / future enforcement) can call `runtime.safety_check.check(ctx, runtime.safety_pack)` whenever they want a verdict. Auto-invocation is a future ADR with its own scope: + +- Where in the turn loop does the check fire (before / after articulation, or both)? +- What does the runtime do with a violation (log, refuse, escalate)? +- How does refusal interact with ADR-0028 / ADR-0030 / ADR-0031 surface preferences? + +None of those questions are settled by ADR-0032 and shouldn't be settled in the same pass that establishes the surface. + +## Consequences + +### Positive + +- **Boundary checks are now centralized**, queryable, and uniformly shaped. An auditor reviewing a turn no longer has to traverse five scattered call sites to confirm boundaries held; they read one `SafetyVerdict`. +- **Honest about what's runtime-checkable.** `runtime_checkable=False` for code-path boundaries is the truth, not a silent pass. +- **Extensible.** Custom predicates for deployment-specific safety boundaries register without touching CORE code. +- **Forward-compatible with enforcement.** When a future ADR wires SafetyCheck into refusal paths, the surface won't need to change — only the call site will. +- **No regression.** Existing scattered enforcement continues to do its job; SafetyCheck is additive. + +### Negative / risks + +- **Observation isn't enforcement.** A violation reported by SafetyCheck at v1 has no automatic consequence — it lives in audit. This is deliberate (the surface lands first; refusal wiring comes later) but worth naming. +- **Predicate authoring is per-deployment work** for any boundary beyond the five v1 defaults. Documentation in `docs/safety_packs.md` will need to grow as deployment patterns emerge. +- **Defensive boundary-id rebinding masks predicate bugs.** A predicate that returns the wrong boundary id gets its result rebinding-corrected, with no warning by default. We accept this trade for safety — better to have the audit verdict reach the right boundary than to crash on a misbehaving predicate. A future debug-mode flag could surface the bug visibly. + +### Scope limits (explicit non-goals for this ADR) + +- No auto-invocation in the turn loop. +- No refusal wiring. +- No refactoring of the existing scattered enforcement sites to delegate to SafetyCheck. +- No structural difference between "violated" and "would-have-been-violated-if-checkable". + +## Verification + +- `tests/test_safety_check.py` — 20 tests covering each of the five default predicates (positive + negative paths where applicable), the unknown-boundary fallback, custom predicate registration, defensive boundary-id rebinding, verdict aggregation, and `ChatRuntime` integration. +- Cognition (121), teaching (17), runtime (19), formation (182), smoke (67) suites green at the same revision. +- The full identity/safety surface suite (`test_identity_packs`, `test_safety_pack`, `test_identity_surface_divergence`, `test_identity_surface_divergence_depth`, `test_identity_score_decomposition`, `test_safety_check`) is 108 tests, all green. diff --git a/docs/safety_packs.md b/docs/safety_packs.md index fee7049f..6d9c77bc 100644 --- a/docs/safety_packs.md +++ b/docs/safety_packs.md @@ -137,9 +137,53 @@ A safety pack is unique to a deployment. The shipping default is `core_safety_ax A new major version means a new `pack_id`. The old pack remains in the repo for replay and audit; the runtime loads whichever pack id is shipped (currently hardcoded in `packs.safety.loader.DEFAULT_SAFETY_PACK`). +## SafetyCheck — structural surface (ADR-0032) + +A centralized, observational surface for evaluating safety boundaries at runtime, parallel in shape to `IdentityCheck`. Produces a `SafetyVerdict`; does not refuse. Wiring violations into refusal paths is a future ADR. + +```python +from packs.safety import SafetyCheck, SafetyContext + +check = SafetyCheck() # ships with default predicates for the five v1 boundaries +ctx = SafetyContext( + field_state=current_field_state, + cited_source_shas=frozenset({...}), + allowed_source_shas=frozenset({...}), + last_refusal_was_typed=True, + identity_manifold_hash_before=before_hash, + identity_manifold_hash_after=after_hash, +) +verdict = check.check(ctx, safety_pack) +# verdict.upheld: bool, verdict.violated_boundaries: frozenset[str] +# verdict.results: tuple of per-boundary SafetyCheckResult +``` + +Every field on `SafetyContext` is optional. Predicates over fields the caller didn't populate default to `upheld=True, runtime_checkable=False` — absence of evidence is not evidence of violation. `ChatRuntime` exposes a pre-constructed instance as `runtime.safety_check`; the turn loop does not auto-invoke it at v1. + +### Default predicates per v1 boundary + +| Boundary | Runtime-checkable? | What it checks | +|---|---|---| +| `preserve_versor_closure` | Yes | `field_state.versor_condition < 1.0e-6` | +| `no_fabricated_source` | Yes (when allowlist supplied) | `cited_source_shas ⊆ allowed_source_shas` | +| `no_silent_correction` | Yes | `last_refusal_was_typed` flag | +| `no_identity_override` | Yes (when both hashes supplied) | identity-manifold hash before == after | +| `no_hot_path_repair` | **No** | code-path boundary; enforced by static analysis + code review | + +The `no_hot_path_repair` predicate reports `runtime_checkable=False` and `upheld=True` honestly. A predicate that silently reported `upheld=True` would be a small lie — the surface acknowledges what it cannot judge. + +### Custom predicates + +```python +check = SafetyCheck() +check.register("my_boundary_id", my_predicate) +``` + +Unknown boundaries (declared in the pack but no predicate registered) default to `upheld=True, runtime_checkable=False, reason="no predicate registered for boundary"`. Doesn't crash; surfaces in audit. + ## Known limits -1. **No `SafetyCheck` parallel to `IdentityCheck`.** Boundaries are enforced elsewhere in the pipeline (refusal paths, allowlist enforcement). A future structural safety-score surface would be valuable but isn't in scope here. +1. ~~**No `SafetyCheck` parallel to `IdentityCheck`.**~~ **Closed by [ADR-0032](decisions/ADR-0032-safety-check-surface.md) (2026-05-17).** See §SafetyCheck above. v1 is observational; turn-loop auto-invocation and refusal wiring are future ADRs. 2. **No per-tenant safety packs.** Multi-tenant CORE deployments share one safety pack. 3. **No human-in-the-loop ratification step.** Operational discipline lives in PR review, not the code. 4. **English-only boundary descriptions** at v1. diff --git a/packs/safety/__init__.py b/packs/safety/__init__.py index 5d752b28..6aeb1ffa 100644 --- a/packs/safety/__init__.py +++ b/packs/safety/__init__.py @@ -12,6 +12,13 @@ cause ``SafetyPackError`` and prevent ``ChatRuntime`` startup. See ``docs/decisions/ADR-0029-safety-packs.md``. """ +from packs.safety.check import ( + SafetyCheck, + SafetyCheckResult, + SafetyContext, + SafetyPredicate, + SafetyVerdict, +) from packs.safety.loader import ( DEFAULT_SAFETY_PACK, SafetyPack, @@ -21,7 +28,12 @@ from packs.safety.loader import ( __all__ = [ "DEFAULT_SAFETY_PACK", + "SafetyCheck", + "SafetyCheckResult", + "SafetyContext", "SafetyPack", "SafetyPackError", + "SafetyPredicate", + "SafetyVerdict", "load_safety_pack", ] diff --git a/packs/safety/check.py b/packs/safety/check.py new file mode 100644 index 00000000..8d503b27 --- /dev/null +++ b/packs/safety/check.py @@ -0,0 +1,332 @@ +"""SafetyCheck — structural surface for safety-pack boundary checks. + +Parallel in *shape* to :class:`core.physics.identity.IdentityCheck`, but +different in kind: identity boundaries are geometric (projection onto +value axes), while safety boundaries are propositional (each is a +predicate over runtime evidence). Mechanically therefore SafetyCheck +is a registry of named predicates, one per boundary id, rather than a +manifold projector. + +Per ADR-0032: + +* SafetyCheck is **observational**. It produces a :class:`SafetyVerdict` + per turn; it does NOT refuse. Wiring violations into refusal paths + is a future ADR. +* Predicates report ``runtime_checkable=False`` honestly when a boundary + is a code-path constraint that cannot be evaluated from runtime + evidence (the canonical example is ``no_hot_path_repair``, whose + enforcement lives in static analysis and code review). +* Unknown boundaries — packs that declare boundary ids for which no + predicate is registered — default to ``upheld=True, + runtime_checkable=False`` so a downstream deployment that adds a + novel boundary doesn't crash the runtime. + +See ``docs/decisions/ADR-0032-safety-check-surface.md``. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from typing import Callable, Mapping + +from packs.safety.loader import SafetyPack + + +# ---------- value types ---------- + + +@dataclass(frozen=True, slots=True) +class SafetyContext: + """Inputs available to safety predicates per turn. + + Every field is optional / has a safe default. Callers populate + whatever they have; predicates over fields the caller didn't + populate default to ``upheld=True`` (no evidence of violation). + The interpretation is deliberate: SafetyCheck is observational, + so absence of evidence is not evidence of violation. + """ + + # For ``preserve_versor_closure``. Any object with a + # ``versor_condition`` attribute / property qualifies. + field_state: object | None = None + versor_halt_threshold: float = 1.0e-6 + + # For ``no_fabricated_source``. Set of SHA-256 hashes the + # trajectory cited as sources. + cited_source_shas: frozenset[str] = frozenset() + # Set of SHA-256 hashes the trust allowlist recognizes. An empty + # ``allowed_source_shas`` is treated as "allowlist not in use" — + # the predicate reports ``runtime_checkable=False`` rather than + # flagging every citation as fabricated. + allowed_source_shas: frozenset[str] = frozenset() + + # For ``no_silent_correction``. ``True`` when the most recent + # refusal / exhaustion event was raised as a typed exception + # (e.g. ``InnerLoopExhaustion``). Default ``True`` means no + # silent correction observed (absence of evidence of silent + # correction). + last_refusal_was_typed: bool = True + + # For ``no_identity_override``. Hashes of the identity manifold + # captured at the start and end of the turn. Empty strings mean + # the caller did not supply the hashes; predicate reports + # ``runtime_checkable=False`` in that case. + identity_manifold_hash_before: str = "" + identity_manifold_hash_after: str = "" + + +@dataclass(frozen=True, slots=True) +class SafetyCheckResult: + """Outcome of one boundary's predicate evaluation.""" + + boundary_id: str + upheld: bool + reason: str + runtime_checkable: bool + evidence: tuple[tuple[str, str], ...] = () + + +@dataclass(frozen=True, slots=True) +class SafetyVerdict: + """Aggregate verdict over every boundary in the pack.""" + + pack_id: str + results: tuple[SafetyCheckResult, ...] + upheld: bool + violated_boundaries: frozenset[str] + runtime_checkable_count: int + + +# ---------- predicate signature ---------- + + +SafetyPredicate = Callable[[SafetyContext], SafetyCheckResult] + + +# ---------- the check ---------- + + +class SafetyCheck: + """Structural safety surface. Observational; never refuses. + + Canonical call style:: + + verdict = SafetyCheck().check(ctx, safety_pack) + """ + + def __init__( + self, + predicates: Mapping[str, SafetyPredicate] | None = None, + ) -> None: + if predicates is None: + self._predicates: dict[str, SafetyPredicate] = dict(_DEFAULT_PREDICATES) + else: + self._predicates = dict(predicates) + + def register(self, boundary_id: str, predicate: SafetyPredicate) -> None: + """Register / replace a predicate for ``boundary_id``.""" + self._predicates[boundary_id] = predicate + + def check( + self, + ctx: SafetyContext, + safety_pack: SafetyPack, + ) -> SafetyVerdict: + """Run every predicate. Returns an aggregate verdict. + + Boundaries are evaluated in lex order on ``boundary_id`` so + ``results`` is deterministic regardless of how the pack + enumerates ``boundary_ids``. + """ + results: list[SafetyCheckResult] = [] + runtime_checkable_count = 0 + violated: set[str] = set() + for boundary in sorted(safety_pack.boundary_ids): + predicate = self._predicates.get(boundary) + if predicate is None: + result = SafetyCheckResult( + boundary_id=boundary, + upheld=True, + reason="no predicate registered for boundary", + runtime_checkable=False, + ) + else: + result = predicate(ctx) + if result.boundary_id != boundary: + # Defensive: a registered predicate must declare + # the same boundary id it was registered under. + result = SafetyCheckResult( + boundary_id=boundary, + upheld=result.upheld, + reason=result.reason, + runtime_checkable=result.runtime_checkable, + evidence=result.evidence, + ) + results.append(result) + if result.runtime_checkable: + runtime_checkable_count += 1 + if not result.upheld: + violated.add(boundary) + return SafetyVerdict( + pack_id=safety_pack.pack_id, + results=tuple(results), + upheld=not violated, + violated_boundaries=frozenset(violated), + runtime_checkable_count=runtime_checkable_count, + ) + + +# ---------- default predicates for v1 boundaries ---------- + + +def _predicate_versor_closure(ctx: SafetyContext) -> SafetyCheckResult: + """``preserve_versor_closure`` — field's versor_condition under threshold.""" + fs = ctx.field_state + if fs is None: + return SafetyCheckResult( + boundary_id="preserve_versor_closure", + upheld=True, + reason="no field_state supplied", + runtime_checkable=False, + ) + vc = getattr(fs, "versor_condition", None) + if vc is None: + return SafetyCheckResult( + boundary_id="preserve_versor_closure", + upheld=True, + reason="field_state has no versor_condition attribute", + runtime_checkable=False, + ) + vc_value = float(vc) + upheld = vc_value < ctx.versor_halt_threshold + return SafetyCheckResult( + boundary_id="preserve_versor_closure", + upheld=upheld, + reason=( + f"versor_condition={vc_value:.3e} " + f"{'<' if upheld else '>='} threshold={ctx.versor_halt_threshold:.0e}" + ), + runtime_checkable=True, + evidence=(("versor_condition", f"{vc_value:.6e}"),), + ) + + +def _predicate_no_fabricated_source(ctx: SafetyContext) -> SafetyCheckResult: + """``no_fabricated_source`` — every cited SHA is in the allowlist.""" + if not ctx.allowed_source_shas: + return SafetyCheckResult( + boundary_id="no_fabricated_source", + upheld=True, + reason="allowed_source_shas is empty (allowlist not in use)", + runtime_checkable=False, + ) + if not ctx.cited_source_shas: + return SafetyCheckResult( + boundary_id="no_fabricated_source", + upheld=True, + reason="no citations to check", + runtime_checkable=True, + ) + fabricated = ctx.cited_source_shas - ctx.allowed_source_shas + if fabricated: + return SafetyCheckResult( + boundary_id="no_fabricated_source", + upheld=False, + reason=( + f"{len(fabricated)} citation(s) not in allowlist: " + f"{sorted(s[:12] + '...' for s in fabricated)}" + ), + runtime_checkable=True, + evidence=tuple( + (f"fabricated_{i}", sha) + for i, sha in enumerate(sorted(fabricated)) + ), + ) + return SafetyCheckResult( + boundary_id="no_fabricated_source", + upheld=True, + reason=f"all {len(ctx.cited_source_shas)} citation(s) in allowlist", + runtime_checkable=True, + ) + + +def _predicate_no_silent_correction(ctx: SafetyContext) -> SafetyCheckResult: + """``no_silent_correction`` — refusals raised as typed exceptions.""" + upheld = bool(ctx.last_refusal_was_typed) + return SafetyCheckResult( + boundary_id="no_silent_correction", + upheld=upheld, + reason=( + "last refusal was typed" + if upheld + else "last refusal was swallowed (no typed exception observed)" + ), + runtime_checkable=True, + ) + + +def _predicate_no_identity_override(ctx: SafetyContext) -> SafetyCheckResult: + """``no_identity_override`` — identity manifold unchanged across the turn.""" + before = ctx.identity_manifold_hash_before + after = ctx.identity_manifold_hash_after + if not before or not after: + return SafetyCheckResult( + boundary_id="no_identity_override", + upheld=True, + reason="identity manifold hashes not supplied", + runtime_checkable=False, + ) + upheld = before == after + return SafetyCheckResult( + boundary_id="no_identity_override", + upheld=upheld, + reason=( + "identity manifold unchanged" + if upheld + else f"identity manifold mutated: {before[:12]}... -> {after[:12]}..." + ), + runtime_checkable=True, + evidence=( + ("identity_hash_before", before[:24] + "..."), + ("identity_hash_after", after[:24] + "..."), + ), + ) + + +def _predicate_no_hot_path_repair(ctx: SafetyContext) -> SafetyCheckResult: + """``no_hot_path_repair`` — code-path boundary, not runtime-checkable. + + This boundary forbids normalization / drift-repair operators in + ``field/propagate.py``, ``generate/stream.py``, ``vault/store.py``. + Enforcement is static — by tests and by code review. Reporting it + as ``runtime_checkable=False`` is the honest answer. A reviewer + looking at a SafetyVerdict can verify the surface acknowledged the + boundary's nature rather than silently passing it. + """ + return SafetyCheckResult( + boundary_id="no_hot_path_repair", + upheld=True, + reason=( + "code-path boundary; enforced by static analysis + code review, " + "not by runtime check" + ), + runtime_checkable=False, + ) + + +_DEFAULT_PREDICATES: dict[str, SafetyPredicate] = { + "no_fabricated_source": _predicate_no_fabricated_source, + "no_hot_path_repair": _predicate_no_hot_path_repair, + "no_identity_override": _predicate_no_identity_override, + "no_silent_correction": _predicate_no_silent_correction, + "preserve_versor_closure": _predicate_versor_closure, +} + + +__all__ = [ + "SafetyCheck", + "SafetyCheckResult", + "SafetyContext", + "SafetyPredicate", + "SafetyVerdict", +] diff --git a/tests/test_safety_check.py b/tests/test_safety_check.py new file mode 100644 index 00000000..cf4537c9 --- /dev/null +++ b/tests/test_safety_check.py @@ -0,0 +1,290 @@ +"""ADR-0032 — SafetyCheck structural surface. + +These tests cover the five default predicates (positive + negative +paths), the unknown-boundary fallback, custom predicate registration, +and integration with ``ChatRuntime``. + +Mechanics intentionally mirror ``tests/test_identity_packs.py`` / +``tests/test_safety_pack.py`` patterns: small, deterministic, no +fixtures that go through the full cognitive pipeline. SafetyCheck is +observational; every test asserts on the verdict, not on runtime +behavior. +""" + +from __future__ import annotations + +from dataclasses import dataclass + +from chat.runtime import ChatRuntime +from core.config import RuntimeConfig +from packs.safety.check import ( + SafetyCheck, + SafetyCheckResult, + SafetyContext, + SafetyVerdict, +) +from packs.safety.loader import load_safety_pack + + +@dataclass(frozen=True) +class _FakeFieldState: + """Stand-in for FieldState — only ``versor_condition`` is checked.""" + versor_condition: float + + +# ---------- preserve_versor_closure ---------- + + +class TestVersorClosurePredicate: + def test_under_threshold_upheld(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + ctx = SafetyContext(field_state=_FakeFieldState(versor_condition=1.0e-9)) + verdict = check.check(ctx, pack) + result = _find(verdict, "preserve_versor_closure") + assert result.upheld + assert result.runtime_checkable + + def test_at_or_above_threshold_violated(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + ctx = SafetyContext(field_state=_FakeFieldState(versor_condition=1.0e-3)) + verdict = check.check(ctx, pack) + result = _find(verdict, "preserve_versor_closure") + assert not result.upheld + assert result.runtime_checkable + assert "preserve_versor_closure" in verdict.violated_boundaries + + def test_missing_field_state_not_runtime_checkable(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + verdict = check.check(SafetyContext(), pack) + result = _find(verdict, "preserve_versor_closure") + assert result.upheld + assert not result.runtime_checkable + + +# ---------- no_fabricated_source ---------- + + +class TestNoFabricatedSourcePredicate: + def test_all_citations_in_allowlist_upheld(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + allowed = frozenset({"a" * 64, "b" * 64}) + ctx = SafetyContext( + cited_source_shas=frozenset({"a" * 64}), + allowed_source_shas=allowed, + ) + result = _find(check.check(ctx, pack), "no_fabricated_source") + assert result.upheld + assert result.runtime_checkable + + def test_unknown_citation_violates(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + ctx = SafetyContext( + cited_source_shas=frozenset({"c" * 64}), + allowed_source_shas=frozenset({"a" * 64}), + ) + result = _find(check.check(ctx, pack), "no_fabricated_source") + assert not result.upheld + assert result.runtime_checkable + + def test_empty_allowlist_not_runtime_checkable(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + ctx = SafetyContext( + cited_source_shas=frozenset({"a" * 64}), + allowed_source_shas=frozenset(), + ) + result = _find(check.check(ctx, pack), "no_fabricated_source") + # Empty allowlist → not in use → predicate cannot judge. + assert result.upheld + assert not result.runtime_checkable + + +# ---------- no_silent_correction ---------- + + +class TestNoSilentCorrectionPredicate: + def test_typed_refusal_upheld(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + verdict = check.check(SafetyContext(last_refusal_was_typed=True), pack) + result = _find(verdict, "no_silent_correction") + assert result.upheld + assert result.runtime_checkable + + def test_swallowed_refusal_violates(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + ctx = SafetyContext(last_refusal_was_typed=False) + result = _find(check.check(ctx, pack), "no_silent_correction") + assert not result.upheld + assert result.runtime_checkable + + +# ---------- no_identity_override ---------- + + +class TestNoIdentityOverridePredicate: + def test_unchanged_manifold_upheld(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + h = "a" * 64 + ctx = SafetyContext( + identity_manifold_hash_before=h, + identity_manifold_hash_after=h, + ) + result = _find(check.check(ctx, pack), "no_identity_override") + assert result.upheld + assert result.runtime_checkable + + def test_mutated_manifold_violates(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + ctx = SafetyContext( + identity_manifold_hash_before="a" * 64, + identity_manifold_hash_after="b" * 64, + ) + result = _find(check.check(ctx, pack), "no_identity_override") + assert not result.upheld + assert result.runtime_checkable + + def test_missing_hashes_not_runtime_checkable(self) -> None: + check = SafetyCheck() + pack = load_safety_pack() + result = _find(check.check(SafetyContext(), pack), "no_identity_override") + assert result.upheld + assert not result.runtime_checkable + + +# ---------- no_hot_path_repair ---------- + + +class TestNoHotPathRepairPredicate: + def test_always_upheld_never_runtime_checkable(self) -> None: + """``no_hot_path_repair`` is structural; runtime cannot judge it. + + The honest answer: enforcement lives in static analysis + code + review. SafetyCheck reports this transparently. + """ + check = SafetyCheck() + pack = load_safety_pack() + result = _find(check.check(SafetyContext(), pack), "no_hot_path_repair") + assert result.upheld + assert not result.runtime_checkable + assert "static analysis" in result.reason + + +# ---------- unknown-boundary fallback ---------- + + +class TestUnknownBoundary: + def test_unknown_boundary_defaults_to_upheld_not_runtime_checkable( + self, + ) -> None: + # Build a SafetyCheck with NO predicates, then check against the + # real safety pack. Every boundary becomes "unknown". + check = SafetyCheck(predicates={}) + pack = load_safety_pack() + verdict = check.check(SafetyContext(), pack) + assert verdict.upheld + assert verdict.runtime_checkable_count == 0 + for r in verdict.results: + assert r.upheld + assert not r.runtime_checkable + assert "no predicate registered" in r.reason + + +# ---------- custom predicate registration ---------- + + +class TestCustomPredicateRegistration: + def test_register_custom_predicate(self) -> None: + def my_pred(ctx: SafetyContext) -> SafetyCheckResult: + return SafetyCheckResult( + boundary_id="my_custom_boundary", + upheld=False, + reason="always fails for testing", + runtime_checkable=True, + ) + check = SafetyCheck() + check.register("my_custom_boundary", my_pred) + # Build a pack-like object with our custom boundary. + from packs.safety.loader import SafetyPack + custom_pack = SafetyPack( + pack_id="custom_test", + version="1.0.0", + description="test", + boundary_ids=frozenset({"my_custom_boundary"}), + boundary_descriptions={"my_custom_boundary": "test"}, + mastery_report_sha256="", + ratified=False, + ) + verdict = check.check(SafetyContext(), custom_pack) + assert not verdict.upheld + assert "my_custom_boundary" in verdict.violated_boundaries + + def test_predicate_misreporting_boundary_id_is_corrected(self) -> None: + # Defensive behavior: if a registered predicate returns a result + # with the wrong boundary_id, SafetyCheck rebinds it. + def lying_pred(ctx: SafetyContext) -> SafetyCheckResult: + return SafetyCheckResult( + boundary_id="WRONG", + upheld=True, + reason="defensive test", + runtime_checkable=False, + ) + check = SafetyCheck() + check.register("no_silent_correction", lying_pred) + verdict = check.check(SafetyContext(), load_safety_pack()) + result = _find(verdict, "no_silent_correction") + assert result.boundary_id == "no_silent_correction" + + +# ---------- verdict aggregation ---------- + + +class TestVerdictAggregation: + def test_pack_id_recorded(self) -> None: + verdict = SafetyCheck().check(SafetyContext(), load_safety_pack()) + assert verdict.pack_id == "core_safety_axes_v1" + + def test_results_in_lex_order_on_boundary_id(self) -> None: + verdict = SafetyCheck().check(SafetyContext(), load_safety_pack()) + ids = [r.boundary_id for r in verdict.results] + assert ids == sorted(ids) + + def test_runtime_checkable_count_under_default_context(self) -> None: + verdict = SafetyCheck().check(SafetyContext(), load_safety_pack()) + # With an empty SafetyContext, only ``no_silent_correction`` is + # runtime-checkable (default last_refusal_was_typed=True). + assert verdict.runtime_checkable_count == 1 + + +# ---------- ChatRuntime integration ---------- + + +class TestChatRuntimeIntegration: + def test_runtime_exposes_safety_check(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + assert isinstance(rt.safety_check, SafetyCheck) + + def test_runtime_safety_check_can_evaluate_loaded_pack(self) -> None: + rt = ChatRuntime(config=RuntimeConfig()) + verdict = rt.safety_check.check(SafetyContext(), rt.safety_pack) + assert isinstance(verdict, SafetyVerdict) + assert verdict.pack_id == rt.safety_pack.pack_id + assert verdict.upheld # empty ctx → no violations observed + + +# ---------- helpers ---------- + + +def _find(verdict: SafetyVerdict, boundary_id: str) -> SafetyCheckResult: + for r in verdict.results: + if r.boundary_id == boundary_id: + return r + raise AssertionError(f"boundary {boundary_id!r} not in verdict")