mirror of
https://github.com/saymrwulf/crisis.git
synced 2026-05-14 20:37:54 +00:00
saymrwulf
b8684297fa
A new sibling Python package, `crisis_agents`, that lifts the Crisis
protocol from "consensus between machines" to "consensus between AI
agents". Threat model: a team of sub-agents normally talks freely
with its orchestrator (the "mothership"); when the team's boundary
opens and an external agent of unknown trust joins, the mothership
activates the Crisis layer so byzantine equivocation is detectable.
Two-phase orchestration model:
Phase 1 — closed team, no Crisis: agents emit claims directly, the
mothership collects them flat.
Phase 2 — boundary opens: every subsequent claim is wrapped into a
Crisis Message with the agent's stable process_id and a PoW nonce,
delivered into per-agent LamportGraphs, and after each turn the
mothership scans for mutations via LamportGraph.find_mutations.
Phase 3 — proof: when an alarm fires, the mothership emits a
replayable JSON proof-of-malfeasance document with the contradictory
witnesses, their delivery sets, and DAG cross-references showing
which honest agents saw what.
Modules:
- claim.py Claim dataclass + JSON round-trip
- boundary.py membership tracker + open() trigger
- agent.py CrisisAgent abstract + MockAgent + MockByzantineAgent
(the latter equivocates by emitting two variants to
disjoint peer subsets at the same logical turn)
- mothership.py orchestrator driving both phases, building Crisis
Messages from Claims, per-agent LamportGraphs, log
- alarm.py scan_for_mutations: same-agent same-turn distinct
digests with non-identical delivery sets, verified
spacelike via LamportGraph.are_spacelike on the
honest-agent graphs
- proof.py build_proof + ProofDocument + JSON serializer +
verify_proof_self_consistent
- cli.py `crisis-agents demo` + `crisis-agents verify`
- scenarios/ fact_check: reference doc + 6 statements + scripted
honest/byzantine agents producing a deterministic
equivocation on statement s03
Tests: 50 new tests across test_claim, test_boundary, test_mothership,
test_alarm, test_proof, test_demo_fact_check. End-to-end test runs the
fact_check scenario, asserts exactly one alarm raised, proof is built,
re-serialized JSON passes self-consistency. Full suite (existing
crisis + new crisis_agents) green in 0.77s — 145 tests.
Out of scope (deliberately): visualization (separate CrisisViz upgrade
later), real TCP gossip (agents talk via in-process function calls in
the mothership), false-claim detection without equivocation (an
agent that consistently lies but never equivocates is out-voted, not
"caught"; catching it would require a ground-truth oracle).
Reuse from existing crisis package: Message, Vertex, LamportGraph,
LamportGraph.find_mutations, ProofOfWorkWeight, digest. The new code
is a thin adapter layer; the protocol substrate did the heavy lifting.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
50 lines
1.3 KiB
Python
50 lines
1.3 KiB
Python
"""Tests for the Boundary membership tracker."""
|
|
|
|
import pytest
|
|
|
|
from crisis.crypto import digest
|
|
from crisis.message import ID_LENGTH
|
|
from crisis_agents.boundary import Boundary
|
|
|
|
|
|
def pid(name: str) -> bytes:
|
|
return digest(name.encode())[:ID_LENGTH]
|
|
|
|
|
|
class TestBoundary:
|
|
|
|
def test_initially_closed_and_empty(self):
|
|
b = Boundary()
|
|
assert not b.is_open
|
|
assert b.size() == 0
|
|
|
|
def test_add_trusted_in_closed_phase(self):
|
|
b = Boundary()
|
|
b.add_trusted(pid("a"))
|
|
b.add_trusted(pid("b"))
|
|
assert b.size() == 2
|
|
assert b.is_trusted(pid("a"))
|
|
assert b.is_trusted(pid("b"))
|
|
assert not b.is_trusted(pid("c"))
|
|
|
|
def test_open_flips_flag_and_adds_id(self):
|
|
b = Boundary()
|
|
b.add_trusted(pid("a"))
|
|
b.open(pid("new"))
|
|
assert b.is_open
|
|
assert b.is_trusted(pid("new"))
|
|
assert b.size() == 2
|
|
|
|
def test_add_trusted_after_open_rejected(self):
|
|
b = Boundary()
|
|
b.open(pid("a"))
|
|
with pytest.raises(RuntimeError, match="already open"):
|
|
b.add_trusted(pid("b"))
|
|
|
|
def test_open_is_idempotent_on_state(self):
|
|
"""Calling open() a second time with a new id just adds the id."""
|
|
b = Boundary()
|
|
b.open(pid("first"))
|
|
b.open(pid("second"))
|
|
assert b.is_open
|
|
assert b.size() == 2
|