Clean up runtime contracts before cognitive pipeline
- document ChatResponse, TurnEvent, identity, memory/teaching, and test-organization contracts - add local trace and build metadata ignore rules - warn on deprecated IdentityCheck constructor injection - update identity gate tests to canonical ValueAxis and ReasoningTrajectory usage - keep cleanup scoped ahead of cognitive pipeline work
This commit is contained in:
Shay
GitHub
parent
dcb0b34ccc
commit
7401eae7ae
4 changed files with 190 additions and 87 deletions
5
.gitignore
vendored
5
.gitignore
vendored
|
|
@ -1,6 +1,11 @@
|
|||
__pycache__/
|
||||
.pytest_cache/
|
||||
.hypothesis/
|
||||
*.pyc
|
||||
.DS_Store
|
||||
|
||||
*.egg-info/
|
||||
traces/
|
||||
|
||||
core-rs/target/
|
||||
core-rs/Cargo.lock
|
||||
|
|
|
|||
|
|
@ -13,6 +13,7 @@ CORE's identity is not a description of CORE. It is CORE, expressed geometricall
|
|||
|
||||
from __future__ import annotations
|
||||
import math
|
||||
import warnings
|
||||
from dataclasses import dataclass
|
||||
from typing import Dict, FrozenSet, List, Optional, Tuple
|
||||
|
||||
|
|
@ -74,14 +75,21 @@ class IdentityManifold:
|
|||
class IdentityCheck:
|
||||
"""Checks a ReasoningTrajectory against an IdentityManifold.
|
||||
|
||||
Supports both the current explicit call style:
|
||||
Canonical call style:
|
||||
IdentityCheck().check(trajectory, manifold)
|
||||
|
||||
and the older fixture style still used by some tests:
|
||||
Deprecated compatibility style:
|
||||
IdentityCheck(manifold=manifold).check(trajectory)
|
||||
"""
|
||||
|
||||
def __init__(self, manifold: IdentityManifold | None = None) -> None:
|
||||
if manifold is not None:
|
||||
warnings.warn(
|
||||
"IdentityCheck(manifold=...) is deprecated; use "
|
||||
"IdentityCheck().check(trajectory, manifold).",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
self._manifold = manifold
|
||||
|
||||
@staticmethod
|
||||
|
|
|
|||
122
docs/runtime_contracts.md
Normal file
122
docs/runtime_contracts.md
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
# Runtime Contracts
|
||||
|
||||
This document freezes the runtime contracts used by chat, telemetry, memory,
|
||||
and future teaching work. It exists to prevent contract drift between tests,
|
||||
runtime code, and future cognitive pipeline work.
|
||||
|
||||
## Field invariant
|
||||
|
||||
CORE state is a versor field. Runtime code must preserve the core closure
|
||||
contract:
|
||||
|
||||
```text
|
||||
versor_condition(F) < 1e-6
|
||||
```
|
||||
|
||||
If a propagation path violates this invariant, fix the operator path or the
|
||||
explicit closure boundary that owns the transition. Do not hide violations by
|
||||
changing tests or silently downgrading the invariant.
|
||||
|
||||
## ChatResponse contract
|
||||
|
||||
`ChatResponse.surface`
|
||||
: The selected user-facing response. This is the exact string returned by
|
||||
`ChatRuntime.respond()` and should match what the user receives.
|
||||
|
||||
`ChatResponse.walk_surface`
|
||||
: The manifold/token-walk evidence surface. It is trace evidence for what the
|
||||
field traversal produced. It is not necessarily the user-facing response.
|
||||
|
||||
`ChatResponse.articulation_surface`
|
||||
: The proposition/realizer surface. This is the structured linguistic
|
||||
realization of the current proposition or proposition graph.
|
||||
|
||||
Current selection policy:
|
||||
|
||||
```text
|
||||
surface = articulation_surface
|
||||
walk_surface = retained telemetry/evidence
|
||||
```
|
||||
|
||||
Future realizer work may change the selection policy, but must update this
|
||||
document and the contract tests in the same PR.
|
||||
|
||||
## TurnEvent contract
|
||||
|
||||
`TurnEvent.surface`
|
||||
: Exact emitted user-facing response for the turn.
|
||||
|
||||
`TurnEvent.walk_surface`
|
||||
: Exact manifold/token-walk evidence surface for the turn.
|
||||
|
||||
`TurnEvent.articulation_surface`
|
||||
: Exact proposition/realizer surface for the turn.
|
||||
|
||||
`TurnEvent.vault_hits`
|
||||
: Actual count of recall hits applied during generation. Never hardcode this.
|
||||
|
||||
`TurnEvent.flagged`
|
||||
: Mirrors `IdentityScore.flagged` for filtering and trace inspection.
|
||||
|
||||
## Identity contract
|
||||
|
||||
Identity checks are telemetry/gating signals. A flagged identity score must not
|
||||
silently erase useful generation unless an explicit hard-block policy is
|
||||
configured and tested.
|
||||
|
||||
Canonical call style:
|
||||
|
||||
```python
|
||||
IdentityCheck().check(trajectory, manifold)
|
||||
```
|
||||
|
||||
Legacy constructor injection:
|
||||
|
||||
```python
|
||||
IdentityCheck(manifold=manifold).check(trajectory)
|
||||
```
|
||||
|
||||
is supported temporarily and emits `DeprecationWarning`. New code must not use
|
||||
it.
|
||||
|
||||
## Memory and teaching contract
|
||||
|
||||
Session memory can be immediate and local to the running context.
|
||||
|
||||
Reviewed memory must be explicit: user corrections or teaching examples become
|
||||
reviewed memory only through the reviewed teaching loop.
|
||||
|
||||
Pack mutation is proposal-only until reviewed. Runtime correction capture must
|
||||
not directly rewrite language packs, frames, identity axes, or operator code.
|
||||
|
||||
Identity manifold mutation by user prompt or correction is forbidden.
|
||||
|
||||
## Testing policy
|
||||
|
||||
Tests should protect load-bearing behavior:
|
||||
|
||||
- versor closure
|
||||
- deterministic replay
|
||||
- runtime response/telemetry contracts
|
||||
- memory correctness
|
||||
- identity protection
|
||||
- teaching/correction safety
|
||||
- articulation contract
|
||||
|
||||
Avoid tests that preserve stale constructors, private helper shapes, or exact
|
||||
formatting that is not part of a documented contract.
|
||||
|
||||
## Test organization target
|
||||
|
||||
Future test moves should follow this taxonomy:
|
||||
|
||||
| Area | Destination |
|
||||
|---|---|
|
||||
| versor closure, holonomy, motors, null cone, energy physics | `tests/algebra/` or `tests/physics/` |
|
||||
| chat runtime, config, async runtime, identity gate telemetry | `tests/runtime/` |
|
||||
| articulation, proposition, surface assembly, future pipeline | `tests/cognition/` |
|
||||
| correction capture, reviewed memory, consolidation | `tests/teaching/` |
|
||||
| language pack loading and seed pack invariants | `tests/packs/` |
|
||||
|
||||
Do not reorganize tests as a standalone churn PR unless it directly reduces
|
||||
contract ambiguity or unlocks a cognitive subsystem.
|
||||
|
|
@ -1,150 +1,118 @@
|
|||
"""tests/test_identity_gate.py — Tests for IdentityCheck gate wiring.
|
||||
|
||||
Covers:
|
||||
- IdentityCheck.check() returns an IdentityScore for a valid trajectory
|
||||
- IdentityScore.score is a float in [0, 1]
|
||||
- IdentityScore.flagged is a bool
|
||||
- IdentityScore.value and .alignment property aliases work
|
||||
- IdentityScore.axes_evaluated returns a sorted list
|
||||
- ChatRuntime.turn_log[-1].identity_score is populated after chat()
|
||||
- ChatRuntime.turn_log[-1].flagged matches response.flagged
|
||||
- TurnEvent.elaboration field exists and is Optional[str]
|
||||
- A response that should not be flagged has flagged=False
|
||||
"""
|
||||
"""Runtime identity-gate contract tests."""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
import dataclasses
|
||||
|
||||
import pytest
|
||||
|
||||
from core.physics.drive import ValueAxis
|
||||
from core.physics.identity import (
|
||||
IdentityCheck,
|
||||
IdentityManifold,
|
||||
IdentityScore,
|
||||
TurnEvent,
|
||||
ValueAxis,
|
||||
)
|
||||
from core.physics.reasoning import ReasoningTrajectory, TrajectoryOperator
|
||||
|
||||
import numpy as np
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class _Frame:
|
||||
frame_id: str
|
||||
coherence_magnitude: float
|
||||
region_ids: frozenset[str]
|
||||
cycle_index: int
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Helpers
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
def _make_manifold() -> IdentityManifold:
|
||||
return IdentityManifold(
|
||||
value_axes=(
|
||||
ValueAxis(
|
||||
axis_id="truthfulness",
|
||||
name="truthfulness",
|
||||
direction=np.array([1.0, 0.0, 0.0], dtype=np.float32),
|
||||
weight=1.0,
|
||||
direction=(1.0, 0.0, 0.0),
|
||||
theological_note="test truth axis",
|
||||
),
|
||||
ValueAxis(
|
||||
axis_id="coherence",
|
||||
name="coherence",
|
||||
direction=np.array([0.0, 1.0, 0.0], dtype=np.float32),
|
||||
weight=1.0,
|
||||
direction=(0.0, 1.0, 0.0),
|
||||
theological_note="test coherence axis",
|
||||
),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def _make_trajectory(n_steps: int = 4) -> ReasoningTrajectory:
|
||||
"""Minimal trajectory with n_steps identity operators."""
|
||||
ops = [
|
||||
TrajectoryOperator(
|
||||
versor=np.ones(32, dtype=np.float32) / np.sqrt(32),
|
||||
step=i,
|
||||
frames = [
|
||||
_Frame(
|
||||
frame_id=f"f{i}",
|
||||
coherence_magnitude=1.0,
|
||||
region_ids=frozenset({str(i % 2)}),
|
||||
cycle_index=i,
|
||||
)
|
||||
for i in range(n_steps)
|
||||
]
|
||||
return ReasoningTrajectory(operators=tuple(ops), turn=0)
|
||||
return TrajectoryOperator().build(frames, trajectory_id="test_identity_trajectory")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# IdentityScore invariants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestIdentityScore:
|
||||
def test_score_is_float_in_unit_interval(self):
|
||||
manifold = _make_manifold()
|
||||
check = IdentityCheck(manifold=manifold)
|
||||
trajectory = _make_trajectory()
|
||||
score = check.check(trajectory)
|
||||
score = IdentityCheck().check(_make_trajectory(), _make_manifold())
|
||||
assert isinstance(score, IdentityScore)
|
||||
assert 0.0 <= score.score <= 1.0
|
||||
|
||||
def test_flagged_is_bool(self):
|
||||
manifold = _make_manifold()
|
||||
check = IdentityCheck(manifold=manifold)
|
||||
score = check.check(_make_trajectory())
|
||||
score = IdentityCheck().check(_make_trajectory(), _make_manifold())
|
||||
assert isinstance(score.flagged, bool)
|
||||
|
||||
def test_value_alias_matches_score(self):
|
||||
manifold = _make_manifold()
|
||||
check = IdentityCheck(manifold=manifold)
|
||||
score = check.check(_make_trajectory())
|
||||
score = IdentityCheck().check(_make_trajectory(), _make_manifold())
|
||||
assert score.value == score.score
|
||||
|
||||
def test_alignment_is_float_in_unit_interval(self):
|
||||
manifold = _make_manifold()
|
||||
check = IdentityCheck(manifold=manifold)
|
||||
score = check.check(_make_trajectory())
|
||||
score = IdentityCheck().check(_make_trajectory(), _make_manifold())
|
||||
assert 0.0 <= score.alignment <= 1.0
|
||||
|
||||
def test_axes_evaluated_is_sorted_list(self):
|
||||
manifold = _make_manifold()
|
||||
check = IdentityCheck(manifold=manifold)
|
||||
score = check.check(_make_trajectory())
|
||||
score = IdentityCheck().check(_make_trajectory(), _make_manifold())
|
||||
axes = score.axes_evaluated
|
||||
assert isinstance(axes, list)
|
||||
assert axes == sorted(axes)
|
||||
|
||||
def test_deviation_axes_is_frozenset_of_str(self):
|
||||
manifold = _make_manifold()
|
||||
check = IdentityCheck(manifold=manifold)
|
||||
score = check.check(_make_trajectory())
|
||||
score = IdentityCheck().check(_make_trajectory(), _make_manifold())
|
||||
assert isinstance(score.deviation_axes, frozenset)
|
||||
for ax in score.deviation_axes:
|
||||
assert isinstance(ax, str)
|
||||
for axis_id in score.deviation_axes:
|
||||
assert isinstance(axis_id, str)
|
||||
|
||||
def test_legacy_constructor_emits_deprecation_warning(self):
|
||||
with pytest.deprecated_call(match="IdentityCheck\(manifold=\.\.\.\) is deprecated"):
|
||||
check = IdentityCheck(manifold=_make_manifold())
|
||||
score = check.check(_make_trajectory())
|
||||
assert isinstance(score, IdentityScore)
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# TurnEvent field invariants
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestTurnEventFields:
|
||||
def test_elaboration_field_exists_and_is_optional_str(self):
|
||||
"""TurnEvent.elaboration must be Optional[str] — verify field presence."""
|
||||
import dataclasses
|
||||
fields = {f.name: f for f in dataclasses.fields(TurnEvent)}
|
||||
assert "elaboration" in fields, "TurnEvent missing elaboration field"
|
||||
# Default must be None.
|
||||
# We can't instantiate without all required fields, so just check the
|
||||
# default value via the field descriptor.
|
||||
field_obj = fields["elaboration"]
|
||||
assert field_obj.default is None
|
||||
fields = {field.name: field for field in dataclasses.fields(TurnEvent)}
|
||||
assert "elaboration" in fields
|
||||
assert fields["elaboration"].default is None
|
||||
|
||||
def test_surface_contract_fields_exist(self):
|
||||
fields = {field.name for field in dataclasses.fields(TurnEvent)}
|
||||
assert {"surface", "walk_surface", "articulation_surface"} <= fields
|
||||
|
||||
def test_identity_score_field_is_optional(self):
|
||||
import dataclasses
|
||||
fields = {f.name: f for f in dataclasses.fields(TurnEvent)}
|
||||
fields = {field.name for field in dataclasses.fields(TurnEvent)}
|
||||
assert "identity_score" in fields
|
||||
|
||||
def test_flagged_field_exists(self):
|
||||
import dataclasses
|
||||
fields = {f.name: f for f in dataclasses.fields(TurnEvent)}
|
||||
fields = {field.name for field in dataclasses.fields(TurnEvent)}
|
||||
assert "flagged" in fields
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# ChatRuntime integration: identity_score wired into turn_log
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class TestChatRuntimeIdentityWiring:
|
||||
"""These tests instantiate a live ChatRuntime and check identity gate wiring.
|
||||
|
||||
Skipped if ChatRuntime cannot be instantiated (missing language pack data).
|
||||
"""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def runtime(self):
|
||||
try:
|
||||
|
|
@ -159,20 +127,20 @@ class TestChatRuntimeIdentityWiring:
|
|||
|
||||
def test_identity_score_is_identityscore_or_none(self):
|
||||
self._runtime.chat("light", max_tokens=4)
|
||||
ev = self._runtime.turn_log[-1]
|
||||
assert ev.identity_score is None or isinstance(ev.identity_score, IdentityScore)
|
||||
event = self._runtime.turn_log[-1]
|
||||
assert event.identity_score is None or isinstance(event.identity_score, IdentityScore)
|
||||
|
||||
def test_flagged_matches_response(self):
|
||||
response = self._runtime.chat("covenant", max_tokens=4)
|
||||
ev = self._runtime.turn_log[-1]
|
||||
assert ev.flagged == response.flagged
|
||||
event = self._runtime.turn_log[-1]
|
||||
assert event.flagged == response.flagged
|
||||
|
||||
def test_elaboration_is_none_or_str(self):
|
||||
self._runtime.chat("word", max_tokens=8)
|
||||
ev = self._runtime.turn_log[-1]
|
||||
assert ev.elaboration is None or isinstance(ev.elaboration, str)
|
||||
event = self._runtime.turn_log[-1]
|
||||
assert event.elaboration is None or isinstance(event.elaboration, str)
|
||||
|
||||
def test_unflagged_response_has_non_empty_surface(self):
|
||||
response = self._runtime.chat("beginning", max_tokens=8)
|
||||
if not response.flagged:
|
||||
assert response.surface, "Surface must not be empty for unflagged response"
|
||||
assert response.surface
|
||||
|
|
|
|||
Loading…
Reference in a new issue