russell@unturf.com
883289d00e
Pre-Phase-1e: falsification-v1.jsonl covered 10 motif tags across 50 fixtures (the high-traffic warrant/title/anchor/format set). Phase 1e adds 12 fixtures (5f-fal-051..062) for the previously-uncovered motifs from the verifier+soft-demote registries: CITATION_MISMATCH DEFLECTION_DETECTED MANUAL_QUOTE_VIOLATION SCHEMA_INVALID SOURCE_ROLE_BLOCKED SUBJECT_TOKENS_ABSENT TOO_MANY_EVIDENCE_IDS UNKNOWN_EVIDENCE_ID BROAD_QUANTIFIER_RUNAWAY BROAD_QUANTIFIER_CAP_APPLIED BROAD_QUANTIFIER_SCOPE_UNBOUND BROAD_QUANTIFIER_REJECTED Coverage now: 22 unique motif tags across 62 fixtures. Harness changes: - test_bench_batteries.py: bump pass_count assertion 50 → 62 in both falsification tests; add test_5f_falsification_covers_every_documented_motif that pins the motif set against the verifier+soft-demote registries so adding a new violation upstream surfaces here as a missing fixture (loud signal, no silent drift). - test_session_integration.py: bump full-suite total 662 → 674. Closes #000025 §10.12 (every documented failure-motif tag). Still open in Phase 1b: §10.11 (real shard finetuning chains), §10.13 (Feedback Loop latency/efficiency against real workload), §10.14 (threshold handoff to #000012).
367 lines
13 KiB
Python
367 lines
13 KiB
Python
"""Cross-module integration tests for this session's surface.
|
|
|
|
End-to-end flows that exercise multiple modules talking to each
|
|
other — beyond what per-module unit tests cover.
|
|
|
|
Scope:
|
|
- ingest emits a capital_ledger row keyed to its audit event
|
|
- SelfModel.snapshot reads memory_root from memory_records when present
|
|
- π* registry stays read-only after import (re-registering raises)
|
|
- Lattice runners chain back to the π* registry
|
|
- The complete Dav1DPrometheus suite (`runner --all`) returns 0 with
|
|
all 312 fixtures passing
|
|
- Battery runtime_digest reflects the live π* registry fingerprint
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Ingest → audit_events → capital_ledger
|
|
# ---------------------------------------------------------------------
|
|
|
|
|
|
def test_ingest_emits_capital_ledger_row(tmp_path):
|
|
"""A real ingest pass writes one capital_ledger row per batch tied
|
|
to the last audit event in that batch."""
|
|
from arborist.document import Document
|
|
from arborist.ingest import ingest_source
|
|
from arborist.source import Source
|
|
from arborist.store import connect
|
|
|
|
class _StubSource:
|
|
source_type = "stub"
|
|
|
|
def iter_documents(self):
|
|
yield Document(
|
|
uri="https://example.com/a",
|
|
content="Hello world. This is a test document for ingest.",
|
|
title="A",
|
|
source_type="stub",
|
|
)
|
|
yield Document(
|
|
uri="https://example.com/b",
|
|
content="Another document with different content.",
|
|
title="B",
|
|
source_type="stub",
|
|
)
|
|
|
|
db = tmp_path / "shard.db"
|
|
conn = connect(db)
|
|
try:
|
|
stats = ingest_source(conn, _StubSource(), batch_size=10)
|
|
assert stats.inserted == 2
|
|
|
|
# Audit events for the two ingested docs.
|
|
n_audit = conn.execute(
|
|
"SELECT COUNT(*) FROM audit_events WHERE event_type = 'ingest'"
|
|
).fetchone()[0]
|
|
assert n_audit == 2
|
|
|
|
# One capital_ledger row per batch — attached to last event hash.
|
|
rows = conn.execute(
|
|
"SELECT op_type, material, intellectual, audit_event_hash"
|
|
" FROM capital_ledger WHERE op_type = 'ingest'"
|
|
).fetchall()
|
|
assert len(rows) == 1
|
|
assert rows[0]["material"] > 0
|
|
assert rows[0]["intellectual"] > 0
|
|
|
|
# The audit_event_hash on the capital row must match a real audit row.
|
|
last_event = conn.execute(
|
|
"SELECT event_hash FROM audit_events"
|
|
" WHERE event_type = 'ingest' ORDER BY seq DESC LIMIT 1"
|
|
).fetchone()
|
|
assert rows[0]["audit_event_hash"] == last_event["event_hash"]
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------
|
|
# SelfModel ↔ memory_root cite-chain
|
|
# ---------------------------------------------------------------------
|
|
|
|
|
|
def test_selfmodel_snapshot_picks_up_memory_root(tmp_path):
|
|
"""When a memory_root exists, SelfModel.snapshot() folds it in."""
|
|
from arborist.memory import snapshot as memory_snapshot, store_snapshot as memory_store
|
|
from arborist.selfmodel import snapshot as sm_snapshot
|
|
from arborist.store import connect, transaction
|
|
|
|
db = tmp_path / "shard.db"
|
|
conn = connect(db)
|
|
try:
|
|
# 1) Store a memory snapshot first.
|
|
with transaction(conn):
|
|
ms = memory_snapshot(conn)
|
|
mem_root = memory_store(conn, ms)
|
|
|
|
# 2) SelfModel snapshot now reads it.
|
|
with transaction(conn):
|
|
sm = sm_snapshot(conn)
|
|
assert sm.memory_root == mem_root
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
def test_selfmodel_snapshot_works_without_memory_root(tmp_path):
|
|
"""SelfModel.snapshot() returns memory_root=None when memory table
|
|
has no live rows (table exists from migration)."""
|
|
from arborist.selfmodel import snapshot as sm_snapshot
|
|
from arborist.store import connect, transaction
|
|
|
|
db = tmp_path / "shard.db"
|
|
conn = connect(db)
|
|
try:
|
|
with transaction(conn):
|
|
sm = sm_snapshot(conn)
|
|
assert sm.memory_root is None
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
# ---------------------------------------------------------------------
|
|
# π* registry guarantees
|
|
# ---------------------------------------------------------------------
|
|
|
|
|
|
def test_pi_star_registry_rejects_conflicting_registration():
|
|
"""Re-registering the same key with a different instance must raise."""
|
|
from arborist.pi_star import register
|
|
from arborist.pi_star.protocol import PiStarError
|
|
from arborist.pi_star.text import WikitextBaseV1
|
|
|
|
rogue = WikitextBaseV1(name="wikitext-base", version="v1", domain="other")
|
|
with pytest.raises(PiStarError):
|
|
register(rogue)
|
|
|
|
|
|
def test_pi_star_registry_idempotent_on_same_instance():
|
|
"""Registering the SAME instance twice is allowed (no-op)."""
|
|
from arborist.pi_star import REGISTRY, register
|
|
|
|
existing = REGISTRY["wikitext-base@v1"]
|
|
# No exception expected — same instance.
|
|
register(existing)
|
|
assert REGISTRY["wikitext-base@v1"] is existing
|
|
|
|
|
|
def test_pi_star_get_unknown_raises():
|
|
from arborist.pi_star import get
|
|
|
|
with pytest.raises(KeyError):
|
|
get("nonexistent@v999")
|
|
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Battery runners chain to π* registry
|
|
# ---------------------------------------------------------------------
|
|
|
|
|
|
def test_battery_runtime_digest_changes_when_pi_star_added(tmp_path):
|
|
"""The runtime_digest in BatteryResult should reflect the active π*
|
|
registry. Adding a new π* changes the fingerprint."""
|
|
from bench.batteries.b_5s import _runtime_digest
|
|
from arborist.pi_star import REGISTRY, register
|
|
from arborist.pi_star.protocol import PiStar
|
|
from dataclasses import dataclass
|
|
|
|
digest_before = _runtime_digest()
|
|
|
|
@dataclass
|
|
class _TestPiStar:
|
|
name: str = "_test-runtime-digest"
|
|
version: str = "v1"
|
|
domain: str = "text"
|
|
|
|
def canonicalize(self, raw: bytes) -> bytes:
|
|
return raw
|
|
|
|
fake = _TestPiStar()
|
|
register(fake)
|
|
try:
|
|
digest_after = _runtime_digest()
|
|
assert digest_before != digest_after
|
|
finally:
|
|
# Cleanup so other tests don't see the rogue π*.
|
|
del REGISTRY["_test-runtime-digest@v1"]
|
|
digest_restored = _runtime_digest()
|
|
assert digest_restored == digest_before
|
|
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Bench suite end-to-end
|
|
# ---------------------------------------------------------------------
|
|
|
|
|
|
def test_full_dav1dprometheus_suite_runs_end_to_end(tmp_path, capsys):
|
|
"""`bench.batteries.runner --all` runs every Phase-1 sub-battery
|
|
and exits 0 when all 312 fixtures pass."""
|
|
from bench.batteries.runner import main
|
|
|
|
rc = main(["--all", "--out", str(tmp_path / "result.json")])
|
|
assert rc == 0
|
|
|
|
payload = json.loads((tmp_path / "result.json").read_text())
|
|
assert payload["schema_version"] == "bench-result-v1"
|
|
sub_batteries = {
|
|
(r["battery"], r["sub_battery"]) for r in payload["results"]
|
|
}
|
|
# All five 5S sub-batteries are present.
|
|
for sub in ("syntax", "semantics", "syllogism", "synthesis", "semiotics"):
|
|
assert ("5s", sub) in sub_batteries
|
|
# All Phase-1b 5T plus legacy transfer.
|
|
for sub in ("transfer", "transfer-learning", "triangulation",
|
|
"truthtables", "transitivity", "time"):
|
|
assert ("5t", sub) in sub_batteries
|
|
# All five 5F sub-batteries.
|
|
for sub in ("function", "finetuning", "falsification",
|
|
"formulate", "feedback-loop"):
|
|
assert ("5f", sub) in sub_batteries
|
|
# All five 5R sub-batteries (Phase 2 of #000021).
|
|
for sub in ("react", "rearrange", "restore", "replicate", "resonate"):
|
|
assert ("5r", sub) in sub_batteries
|
|
|
|
# Aggregate pass counts: every sub-battery must have zero failures.
|
|
for r in payload["results"]:
|
|
assert r["fail_count"] == 0, (
|
|
f"{r['battery']}/{r['sub_battery']} failed "
|
|
f"{r['fail_count']} fixtures"
|
|
)
|
|
|
|
|
|
def test_full_suite_total_fixture_count():
|
|
"""Sanity check: the complete Dav1DPrometheus suite executes 674
|
|
deterministic tasks across 21 sub-batteries (5S+5T+5F+5R).
|
|
|
|
History:
|
|
- Phase 1a baseline: 462 tasks.
|
|
- Phase 1c (#000025, 2026-05-09): 5F synthetic side expanded
|
|
10 → 30; +100 → 562.
|
|
- Phase 1d (#000025, 2026-05-09): 5F synthetic 30 → 50; +100 → 662.
|
|
(Live side ALSO went 30 → 50 but lives in *-live-v1 files
|
|
that the default-fixture-set doesn't load — those run via
|
|
the dedicated Makefile targets.)
|
|
- Phase 1e (#000025, 2026-05-10): 5F falsification expanded
|
|
50 → 62 to cover every documented failure-motif tag; +12 → 674.
|
|
(Closes #000025 §10.12.)
|
|
"""
|
|
from bench.batteries.runner import _DEFAULT_FIXTURES, _run_one
|
|
|
|
total = 0
|
|
for (battery, sub), fx in _DEFAULT_FIXTURES.items():
|
|
result = _run_one(battery, sub, Path(fx))
|
|
total += result.pass_count + result.fail_count
|
|
assert total == 674
|
|
|
|
|
|
def test_5s_phase1a_digests_unchanged_after_phase1b():
|
|
"""Closure-criterion guard: 5S Phase 1a fixture digests stay pinned."""
|
|
from bench.batteries.base import fixture_digest
|
|
|
|
# Hashes computed on the committed fixture files. If you intentionally
|
|
# change those fixtures, update this list — but Phase 1b explicitly
|
|
# forbids it (per ticket #000023 §1).
|
|
syntax_digest = fixture_digest(
|
|
REPO_ROOT / "bench" / "fixtures" / "5s" / "syntax-v1.jsonl"
|
|
)
|
|
semantics_digest = fixture_digest(
|
|
REPO_ROOT / "bench" / "fixtures" / "5s" / "semantics-v1.jsonl"
|
|
)
|
|
transfer_digest = fixture_digest(
|
|
REPO_ROOT / "bench" / "fixtures" / "5t" / "transfer-v1.jsonl"
|
|
)
|
|
# Two reads must be byte-equal (digest is just a content hash).
|
|
assert syntax_digest == fixture_digest(
|
|
REPO_ROOT / "bench" / "fixtures" / "5s" / "syntax-v1.jsonl"
|
|
)
|
|
assert semantics_digest == fixture_digest(
|
|
REPO_ROOT / "bench" / "fixtures" / "5s" / "semantics-v1.jsonl"
|
|
)
|
|
assert transfer_digest == fixture_digest(
|
|
REPO_ROOT / "bench" / "fixtures" / "5t" / "transfer-v1.jsonl"
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------
|
|
# Cross-cutting: full state-space round-trip
|
|
# ---------------------------------------------------------------------
|
|
|
|
|
|
def test_full_session_state_round_trip(tmp_path):
|
|
"""End-to-end: ingest doc → snapshot SelfModel → snapshot Memory →
|
|
inspect via CLI surfaces. All three v8-substrate components present
|
|
+ chain stays clean."""
|
|
import hashlib
|
|
from arborist.cli import build_parser
|
|
from arborist.document import Document
|
|
from arborist.ingest import ingest_source
|
|
from arborist.store import connect
|
|
|
|
class _Source:
|
|
source_type = "round-trip-stub"
|
|
|
|
def iter_documents(self):
|
|
yield Document(
|
|
uri="https://example.com/round-trip",
|
|
content="Round-trip integration content.",
|
|
title="Round Trip",
|
|
source_type="round-trip-stub",
|
|
)
|
|
|
|
db = tmp_path / "shard.db"
|
|
conn = connect(db)
|
|
try:
|
|
ingest_source(conn, _Source(), batch_size=10)
|
|
finally:
|
|
conn.close()
|
|
|
|
parser = build_parser()
|
|
# Capture: argparse's args.func calls our handlers, all of which
|
|
# print JSON on stdout.
|
|
import io
|
|
import sys
|
|
|
|
def _capture(argv: list[str]) -> dict:
|
|
old = sys.stdout
|
|
sys.stdout = io.StringIO()
|
|
try:
|
|
args = parser.parse_args(argv)
|
|
args.func(args)
|
|
return json.loads(sys.stdout.getvalue())
|
|
finally:
|
|
sys.stdout = old
|
|
|
|
sm = _capture(["--db", str(db), "selfmodel", "snapshot"])
|
|
mem = _capture(["--db", str(db), "memory", "snapshot"])
|
|
cap = _capture(["--db", str(db), "capital", "summary"])
|
|
|
|
assert len(sm["selfmodel_root"]) == 64
|
|
assert len(mem["memory_root"]) == 64
|
|
# Capital row from ingest should be visible.
|
|
assert cap["row_count"] >= 1
|
|
|
|
# Audit chain still verifies.
|
|
conn = connect(db)
|
|
try:
|
|
rows = conn.execute(
|
|
"SELECT event_hash, prev_event_hash, body FROM audit_events ORDER BY seq"
|
|
).fetchall()
|
|
prev = None
|
|
for row in rows:
|
|
h = hashlib.sha256()
|
|
if prev is not None:
|
|
h.update(bytes.fromhex(prev))
|
|
h.update(row["body"].encode("utf-8", errors="surrogatepass"))
|
|
assert h.hexdigest() == row["event_hash"]
|
|
prev = row["event_hash"]
|
|
finally:
|
|
conn.close()
|