310aed9ff0
* docs: consolidate governance anchors and clean up test registries * refactor(cli): decompose cli into dedicated modules * test: fix broken test baselines and formatting * docs: add domain boundary READMEs for governance anchors * test: update baseline for determination lane * test: fix capability_pass expectation * test: fix CORE_SHOWCASE_SKIP_BUDGET enforcement * chore: cleanup CLI extraction and unreachable code
332 lines
12 KiB
Python
332 lines
12 KiB
Python
"""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/adr/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",
|
|
]
|