core-labs/core
1
0
Fork 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

feat(safety): ADR-0032 — SafetyCheck structural surface

Browse source 
Closes the 'boundaries are checked at scattered call sites' gap noted
in ADR-0029.  Adds a centralized observational surface parallel in
shape to IdentityCheck — produces a verdict, does not refuse.  Wiring
verdicts into refusal paths is a future ADR.

Shape (parallel to IdentityCheck, different in mechanism):

  SafetyContext     — duck-typed input bag (field_state, citations,
                       refusal-was-typed flag, identity manifold hashes
                       before/after).  Every field optional with safe
                       defaults; absence of evidence is not evidence of
                       violation.
  SafetyCheckResult — per-boundary: boundary_id, upheld, reason,
                       runtime_checkable, evidence tuple.
  SafetyVerdict     — aggregate: pack_id, results (lex order on
                       boundary_id), upheld, violated_boundaries,
                       runtime_checkable_count.
  SafetyCheck       — registry of predicates; check(ctx, pack) returns
                       SafetyVerdict.  register(boundary_id, predicate)
                       adds custom predicates.

Five default predicates for v1 boundaries:

  preserve_versor_closure   runtime_checkable=True   field.versor_condition < 1e-6
  no_fabricated_source      runtime_checkable=True*  cited ⊆ allowed
  no_silent_correction      runtime_checkable=True   last refusal was typed
  no_identity_override      runtime_checkable=True*  hash before == hash after
  no_hot_path_repair        runtime_checkable=FALSE  code-path; static-analysis

  *Conditional on the caller supplying the necessary fields.

The honest answer on no_hot_path_repair: it is a code-path boundary
enforced by static analysis + code review.  Runtime cannot judge it.
A predicate that silently reported upheld=True would be a small lie —
exactly the kind of thing CLAUDE.md forbids.  SafetyCheck reports
runtime_checkable=False with a clear reason so auditors see the truth.

ChatRuntime integration:
  ChatRuntime.__init__ now constructs self.safety_check = SafetyCheck()
  alongside self._identity_check.  Turn loop does NOT auto-invoke at
  v1 — operators and future ADRs decide when/where to call it.

Files:
  packs/safety/check.py            new — SafetyCheck + value types +
                                   default predicates
  packs/safety/__init__.py         re-exports the new public surface
  chat/runtime.py                  constructs self.safety_check
  tests/test_safety_check.py       new — 20 tests covering each
                                   default predicate (positive +
                                   negative), unknown-boundary
                                   fallback, custom registration,
                                   defensive boundary-id rebinding,
                                   verdict aggregation, ChatRuntime
                                   integration
  docs/decisions/ADR-0032-safety-check-surface.md  Accepted
  docs/safety_packs.md             §SafetyCheck section added,
                                   known-limit #1 struck through
  memory/safety-pack.md            refreshed; new follow-up about
                                   turn-loop auto-invocation

Suite status (all green):
  cognition 121, teaching 17, runtime 19, formation 182, smoke 67
  identity / safety / surface divergence suites: 108 tests passing
  (was 88 before this ADR; +20 safety-check tests)

Scope limits (documented):
  - No auto-invocation in the turn loop.
  - No refusal wiring on violation.
  - No refactoring of existing scattered enforcement sites.
  - Defensive boundary-id rebinding masks predicate bugs; debug-mode
    surfacing is a future enhancement.
This commit is contained in:
Shay 2026-05-17 20:25:22 -07:00
parent 07ad3af845
commit 6f67e9a616
6 changed files with 814 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

7
chat/runtime.py
View file

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

128
docs/decisions/ADR-0032-safety-check-surface.md Normal file
View file

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

46
docs/safety_packs.md
View file

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

12
packs/safety/__init__.py
View file

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

332
packs/safety/check.py Normal file
View file

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

290
tests/test_safety_check.py Normal file
View file

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