Skip to content

Python API Reference

Top-level exports

import agentcrdt

agentcrdt

agentcrdt - Semantic-causal CRDT for agent-mutable world state.

ClosedLoopError

Bases: ValueError

Raised when the gate refuses empty, constant-only, or unusable state.

GateOutcome(ok, verdict, reason, exit_code, fact_count=0, mutable_count=0, constant_count=0, constant_domains=(), refused_writes=0, conflict_count=0, divergence_count=0, human_required=False, contested_keys=()) dataclass

Result of a closed-loop read of an agentcrdt world store.

Attributes:

Name Type Description
ok bool

True only when the pipeline may continue (PASS).
verdict str

PASS, FAIL, or FAIL_LOUD.
reason str

Always non-empty explanation.
exit_code int

0 PASS, 1 FAIL (policy), 2 FAIL_LOUD (empty/unusable).
fact_count int

Facts examined.
mutable_count int

Facts in non-constant domains.
constant_count int

Facts in constant-only domains.
constant_domains tuple[str, ...]

Distinct constant domains seen.
refused_writes int

Count of writes refused (when gating a write batch).
conflict_count int

Unresolved ContradictionEvents (MAST path).
divergence_count int

Silent multi-agent value divergences detected.
human_required bool

True when multi-agent conflict needs human resolve.
contested_keys tuple[str, ...]

Sample of entity.attribute keys that diverged.

ValueDivergence(fact_id, domain, entity, attribute, agents, values, version_count) dataclass

A fact key where ≥2 agents wrote ≥2 distinct values (MAST silent LWW).

AgentTraceEvent(agent_id, tool, timestamp, payload_fp='', side_channel='', meta=dict()) dataclass

One black-box behavioural event from an agent (tool/timing/payload).

CollusionReport(event_count, agent_count, signals, shared_payload_groups, sync_pairs, side_channel_groups) dataclass

Aggregate collusion analysis over a population of agent events.

CollusionSignal(kind, agents, detail, score=1.0) dataclass

One detected collusion pattern.

AgentMessage(msg_id, sender, receiver, content='', channel='peer', role='peer', architecture='') dataclass

One inter-agent coordination message.

Attributes:

Name Type Description
msg_id str

Stable message id.
sender str

Sending agent id.
receiver str

Target agent id (or broadcast / *).
content str

Message body (natural language or structured text).
channel str

Logical channel name (peer, control, …).
role str

Declared sender role (peer, planner, external, …).
architecture str

Optional architecture tag for this hop.

CommAttackReport(message_count, agent_count, signals, external_entry_count, privileged_spoof_count, injection_count, architecture, details=dict()) dataclass

Aggregate communication integrity analysis.

CommAttackSignal(kind, msg_id, detail, score=1.0) dataclass

One communication-attack finding.

ContradictionEvent(rule, facts_involved, agent_a, agent_b, timestamp=time.time()) dataclass

Fired when two agents hold semantically incompatible world facts.

The id is content-addressed from the rule name, sorted fact IDs, and the two agent IDs so identical contradictions dedup correctly.

__post_init__()

Compute content-addressed id.

Source code in src/agentcrdt/fact.py 
84
85
86
87
88
89
def __post_init__(self) -> None:
    """Compute content-addressed id."""
    payload = (
        f"{self.rule}|{'|'.join(sorted(self.facts_involved))}|{self.agent_a}|{self.agent_b}"
    )
    self.id = _sha16(payload)

to_dict()

Serialise to a plain dict.

Source code in src/agentcrdt/fact.py 
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def to_dict(self) -> dict[str, Any]:
    """Serialise to a plain dict."""
    return {
        "id": self.id,
        "rule": self.rule,
        "facts_involved": self.facts_involved,
        "agent_a": self.agent_a,
        "agent_b": self.agent_b,
        "timestamp": self.timestamp,
    }

WorldFact(domain, entity, attribute, value, version=0, agent_id='', timestamp=time.time()) dataclass

An immutable, content-addressed fact about world state.

The id field is derived from domain|entity|attribute so two agents recording the same fact key always get the same ID.

__post_init__()

Compute content-addressed id.

Source code in src/agentcrdt/fact.py 
33
34
35
def __post_init__(self) -> None:
    """Compute content-addressed id."""
    self.id = _sha16(f"{self.domain}|{self.entity}|{self.attribute}")

to_dict()

Serialise to a plain dict.

Source code in src/agentcrdt/fact.py 
37
38
39
40
41
42
43
44
45
46
47
48
def to_dict(self) -> dict[str, Any]:
    """Serialise to a plain dict."""
    return {
        "id": self.id,
        "domain": self.domain,
        "entity": self.entity,
        "attribute": self.attribute,
        "value": self.value,
        "version": self.version,
        "agent_id": self.agent_id,
        "timestamp": self.timestamp,
    }

from_dict(d) classmethod

Deserialise from a plain dict produced by to_dict.

Source code in src/agentcrdt/fact.py 
50
51
52
53
54
55
56
57
58
59
60
61
62
@classmethod
def from_dict(cls, d: dict[str, Any]) -> WorldFact:
    """Deserialise from a plain dict produced by ``to_dict``."""
    f = cls(
        domain=d["domain"],
        entity=d["entity"],
        attribute=d["attribute"],
        value=d["value"],
        version=d.get("version", 0),
        agent_id=d.get("agent_id", ""),
        timestamp=d.get("timestamp", 0.0),
    )
    return f

FactHistory(store)

Source code in src/agentcrdt/history.py 
20
21
def __init__(self, store: WorldStore) -> None:
    self._store = store

get_history(entity, attribute)

Return all versions of a fact, newest first.

Source code in src/agentcrdt/history.py 
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def get_history(self, entity: str, attribute: str) -> list[FactVersion]:
    """Return all versions of a fact, newest first."""
    rows = self._store.list_fact_history_by_entity_attr(entity, attribute)
    # rows is sorted oldest-first (ASC)
    versions: list[FactVersion] = []
    for i, row in enumerate(rows):
        fact = WorldFact.from_dict(
            {
                "domain": row["domain"],
                "entity": row["entity"],
                "attribute": row["attribute"],
                "value": json.loads(row["value"]),
                "version": row["version"],
                "agent_id": row["agent_id"],
                "timestamp": row["timestamp"],
            }
        )
        # This fact was superseded by the next one (i+1), or None if it's the latest
        superseded_by = rows[i + 1]["fact_id"] if i < len(rows) - 1 else None
        version_index = i  # 0 = oldest, len-1 = newest
        versions.append(
            FactVersion(
                fact=fact,
                superseded_by=superseded_by,
                version_index=version_index,
            )
        )
    # Return newest-first as per docstring
    return list(reversed(versions))

get_at_time(entity, attribute, timestamp)

Return the fact that was current at a given timestamp.

Source code in src/agentcrdt/history.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
def get_at_time(self, entity: str, attribute: str, timestamp: float) -> WorldFact | None:
    """Return the fact that was current at a given timestamp."""
    rows = self._store.list_fact_history_by_entity_attr(entity, attribute)
    # Find the last row recorded at or before the given timestamp
    candidate = None
    for row in rows:
        if row["recorded_at"] <= timestamp:
            candidate = row
    if candidate is None:
        return None
    return WorldFact.from_dict(
        {
            "domain": candidate["domain"],
            "entity": candidate["entity"],
            "attribute": candidate["attribute"],
            "value": json.loads(candidate["value"]),
            "version": candidate["version"],
            "agent_id": candidate["agent_id"],
            "timestamp": candidate["timestamp"],
        }
    )

diff_entity(entity)

Return full history of all attributes for an entity.

Source code in src/agentcrdt/history.py 
75
76
77
78
79
80
81
82
83
84
def diff_entity(self, entity: str) -> dict[str, list[FactVersion]]:
    """Return full history of all attributes for an entity."""
    rows = self._store.list_fact_history_by_entity(entity)
    attributes: set[str] = set()
    for row in rows:
        attributes.add(row["attribute"])
    result: dict[str, list[FactVersion]] = {}
    for attr in sorted(attributes):
        result[attr] = self.get_history(entity, attr)
    return result

MergeResult(merged_count, conflicts=list()) dataclass

Summary of a completed merge operation.

to_dict()

Serialise to a plain dict suitable for JSON output.

Source code in src/agentcrdt/merger.py 
20
21
22
23
24
25
def to_dict(self) -> dict[str, Any]:
    """Serialise to a plain dict suitable for JSON output."""
    return {
        "merged_count": self.merged_count,
        "conflicts": [c.to_dict() for c in self.conflicts],
    }

WorldMerger(rule_engine=None)

Merge a remote :class:WorldStore into a local one using CRDT semantics.

Uses Last-Write-Wins (LWW) per fact key. After merging, optionally runs the provided :class:~agentcrdt.rules.RuleEngine to detect semantic contradictions and records them as :class:~agentcrdt.fact.ContradictionEvent objects.

Initialise with an optional rule engine for contradiction detection.

Source code in src/agentcrdt/merger.py 
37
38
39
def __init__(self, rule_engine: RuleEngine | None = None) -> None:
    """Initialise with an optional rule engine for contradiction detection."""
    self.rule_engine = rule_engine

merge(local, remote)

Merge remote into local using LWW CRDT semantics.

Parameters:

Name Type Description Default
local WorldStore

The target store (modified in-place).

 required
remote WorldStore

The source store (read-only).

 required

Returns:

Name Type Description
A MergeResult

class:MergeResult with the number of facts merged and any
MergeResult

contradiction events detected by the rule engine.
Source code in src/agentcrdt/merger.py 
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def merge(self, local: WorldStore, remote: WorldStore) -> MergeResult:
    """Merge *remote* into *local* using LWW CRDT semantics.

    Args:
        local:  The target store (modified in-place).
        remote: The source store (read-only).

    Returns:
        A :class:`MergeResult` with the number of facts merged and any
        contradiction events detected by the rule engine.
    """
    merged = 0
    remote_facts = remote.list_facts()
    for fact in remote_facts:
        local.set_fact(fact)
        merged += 1

    conflicts: list[ContradictionEvent] = []
    if self.rule_engine is not None:
        all_facts = {f.id: f for f in local.list_facts()}
        conflicts = self.rule_engine.check(all_facts)
        for evt in conflicts:
            local.add_event(evt)

    return MergeResult(merged_count=merged, conflicts=conflicts)

RuleEngine(rules)

Evaluates a set of SemanticRule objects against a snapshot of world facts.

Initialise with a list of semantic rules to enforce.

Source code in src/agentcrdt/rules.py 
34
35
36
def __init__(self, rules: list[SemanticRule]) -> None:
    """Initialise with a list of semantic rules to enforce."""
    self.rules = rules

check(facts)

Check all semantic rules and return ContradictionEvents for violations.

Parameters:

Name Type Description Default
facts dict[str, WorldFact]

Mapping of fact_id -> WorldFact representing the current world state.

 required

Returns:

Type Description
list[ContradictionEvent]

A list of :class:ContradictionEvent objects, one per violated rule.
Source code in src/agentcrdt/rules.py 
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
def check(self, facts: dict[str, WorldFact]) -> list[ContradictionEvent]:
    """Check all semantic rules and return ContradictionEvents for violations.

    Args:
        facts: Mapping of ``fact_id -> WorldFact`` representing the current
            world state.

    Returns:
        A list of :class:`ContradictionEvent` objects, one per violated rule.
    """
    events: list[ContradictionEvent] = []
    # Group facts by (domain, entity, attribute) for fast lookup
    by_key: dict[tuple[str, str, str], WorldFact] = {}
    for f in facts.values():
        by_key[(f.domain, f.entity, f.attribute)] = f

    for rule in self.rules:
        # Find all facts that trigger this rule
        for f in facts.values():
            if f.domain != rule.trigger_domain or f.attribute != rule.trigger_attribute:
                continue
            if f.value != rule.trigger_value:
                continue
            # This fact triggers the rule
            if rule.implies_entity_same:
                implied_key = (rule.implies_domain, f.entity, rule.implies_attribute)
                candidates: list[WorldFact] = [
                    c for c in [by_key.get(implied_key)] if c is not None
                ]
            else:
                # Check all facts in implies_domain with implies_attribute (any entity)
                candidates = [
                    g
                    for g in facts.values()
                    if g.domain == rule.implies_domain and g.attribute == rule.implies_attribute
                ]  # facts.values() is WorldFact (non-optional)
            for implied in candidates:
                if implied.value != rule.implies_value:
                    evt = ContradictionEvent(
                        rule=rule.name,
                        facts_involved=[f.id, implied.id],
                        agent_a=f.agent_id,
                        agent_b=implied.agent_id,
                    )
                    events.append(evt)
    return events

SemanticRule(name, trigger_domain, trigger_attribute, trigger_value, implies_domain, implies_entity_same=True, implies_attribute='', implies_value=None) dataclass

A first-order semantic implication rule between two world facts.

When a fact matching trigger_domain / trigger_attribute / trigger_value exists, the rule asserts that a related fact in implies_domain must have implies_value. If the implied fact disagrees, a ContradictionEvent is emitted.

WorldStore(path)

Persistent store backed by a single SQLite database file.

Supports context-manager usage::

with WorldStore("world.db") as store:
    store.set_fact(fact)

Open (or create) a WorldStore at path.

Source code in src/agentcrdt/store.py 
56
57
58
59
60
61
62
63
def __init__(self, path: str | Path) -> None:
    """Open (or create) a WorldStore at *path*."""
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self._conn = sqlite3.connect(str(self.path), check_same_thread=False)
    self._conn.row_factory = sqlite3.Row
    self._conn.executescript(self._SCHEMA)
    self._conn.commit()

close()

Close the underlying database connection.

Source code in src/agentcrdt/store.py 
65
66
67
def close(self) -> None:
    """Close the underlying database connection."""
    self._conn.close()

__enter__()

Support with WorldStore(...) as store: usage.

Source code in src/agentcrdt/store.py 
69
70
71
def __enter__(self) -> WorldStore:
    """Support ``with WorldStore(...) as store:`` usage."""
    return self

__exit__(*args)

Close the store on context-manager exit.

Source code in src/agentcrdt/store.py 
73
74
75
def __exit__(self, *args: Any) -> None:
    """Close the store on context-manager exit."""
    self.close()

set_fact(fact)

Store or update a fact using LWW semantics (higher version wins, then timestamp).

Source code in src/agentcrdt/store.py 
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
def set_fact(self, fact: WorldFact) -> None:
    """Store or update a fact using LWW semantics (higher version wins, then timestamp)."""
    existing = self.get_fact(fact.id)
    if existing is not None:
        # LWW: higher version wins; on tie, higher timestamp wins
        if fact.version < existing.version:
            return
        if fact.version == existing.version and fact.timestamp <= existing.timestamp:
            return
    value_json = json.dumps(fact.value)
    self._conn.execute(
        "INSERT OR REPLACE INTO facts VALUES (?,?,?,?,?,?,?,?)",
        (
            fact.id,
            fact.domain,
            fact.entity,
            fact.attribute,
            value_json,
            fact.version,
            fact.agent_id,
            fact.timestamp,
        ),
    )
    # Log to history
    self._conn.execute(
        "INSERT INTO fact_history (fact_id, domain, entity, attribute,"
        " value, version, agent_id, timestamp, recorded_at)"
        " VALUES (?,?,?,?,?,?,?,?,?)",
        (
            fact.id,
            fact.domain,
            fact.entity,
            fact.attribute,
            value_json,
            fact.version,
            fact.agent_id,
            fact.timestamp,
            _time_mod.time(),
        ),
    )
    self._conn.commit()

get_fact(fact_id)

Return a single :class:WorldFact by id, or None if not found.

Source code in src/agentcrdt/store.py 
119
120
121
122
123
124
125
126
def get_fact(self, fact_id: str) -> WorldFact | None:
    """Return a single :class:`WorldFact` by id, or ``None`` if not found."""
    row = self._conn.execute("SELECT * FROM facts WHERE id=?", (fact_id,)).fetchone()
    if row is None:
        return None
    d = dict(row)
    d["value"] = json.loads(d["value"])
    return WorldFact.from_dict(d)

get_fact_by_key(domain, entity, attribute)

Return a fact looked up by its natural key (domain, entity, attribute).

Convenience alternative to :meth:get_fact when you don't have the SHA-256 fact_id at hand.

Parameters:

Name Type Description Default
domain str

Fact domain, e.g. "life".

 required
entity str

Entity name, e.g. "king".

 required
attribute str

Attribute name, e.g. "alive".

 required

Returns:

Type Description
WorldFact | None

The matching :class:WorldFact, or None if not found.
Source code in src/agentcrdt/store.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def get_fact_by_key(self, domain: str, entity: str, attribute: str) -> WorldFact | None:
    """Return a fact looked up by its natural key ``(domain, entity, attribute)``.

    Convenience alternative to :meth:`get_fact` when you don't have the
    SHA-256 ``fact_id`` at hand.

    Args:
        domain:    Fact domain, e.g. ``"life"``.
        entity:    Entity name, e.g. ``"king"``.
        attribute: Attribute name, e.g. ``"alive"``.

    Returns:
        The matching :class:`WorldFact`, or ``None`` if not found.
    """
    row = self._conn.execute(
        "SELECT * FROM facts WHERE domain=? AND entity=? AND attribute=?",
        (domain, entity, attribute),
    ).fetchone()
    if row is None:
        return None
    d = dict(row)
    d["value"] = json.loads(d["value"])
    return WorldFact.from_dict(d)

list_facts(domain=None)

Return all stored facts, optionally filtered by domain.

Source code in src/agentcrdt/store.py 
152
153
154
155
156
157
158
159
160
161
162
163
164
165
def list_facts(self, domain: str | None = None) -> list[WorldFact]:
    """Return all stored facts, optionally filtered by *domain*."""
    if domain:
        rows = self._conn.execute(
            "SELECT * FROM facts WHERE domain=? ORDER BY timestamp", (domain,)
        ).fetchall()
    else:
        rows = self._conn.execute("SELECT * FROM facts ORDER BY timestamp").fetchall()
    result = []
    for row in rows:
        d = dict(row)
        d["value"] = json.loads(d["value"])
        result.append(WorldFact.from_dict(d))
    return result

add_event(event)

Persist a :class:ContradictionEvent.

Source code in src/agentcrdt/store.py 
167
168
169
170
171
172
173
174
175
176
177
178
179
180
def add_event(self, event: ContradictionEvent) -> None:
    """Persist a :class:`ContradictionEvent`."""
    self._conn.execute(
        "INSERT OR REPLACE INTO events VALUES (?,?,?,?,?,?)",
        (
            event.id,
            event.rule,
            json.dumps(event.facts_involved),
            event.agent_a,
            event.agent_b,
            event.timestamp,
        ),
    )
    self._conn.commit()

list_events()

Return all stored contradiction events ordered by timestamp.

Source code in src/agentcrdt/store.py 
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
def list_events(self) -> list[ContradictionEvent]:
    """Return all stored contradiction events ordered by timestamp."""
    rows = self._conn.execute("SELECT * FROM events ORDER BY timestamp").fetchall()
    result = []
    for row in rows:
        d = dict(row)
        d["facts_involved"] = json.loads(d["facts_involved"])
        e = ContradictionEvent(
            rule=d["rule"],
            facts_involved=d["facts_involved"],
            agent_a=d["agent_a"],
            agent_b=d["agent_b"],
            timestamp=d["timestamp"],
        )
        result.append(e)
    return result

list_fact_history(fact_id)

Return all historical rows for a fact_id ordered by recorded_at ASC.

Source code in src/agentcrdt/store.py 
199
200
201
202
203
204
205
def list_fact_history(self, fact_id: str) -> list[dict[str, Any]]:
    """Return all historical rows for a fact_id ordered by recorded_at ASC."""
    rows = self._conn.execute(
        "SELECT * FROM fact_history WHERE fact_id=? ORDER BY recorded_at ASC",
        (fact_id,),
    ).fetchall()
    return [dict(r) for r in rows]

list_fact_history_by_entity_attr(entity, attribute)

Return all historical rows for entity+attribute, ordered by recorded_at ASC.

Source code in src/agentcrdt/store.py 
207
208
209
210
211
212
213
def list_fact_history_by_entity_attr(self, entity: str, attribute: str) -> list[dict[str, Any]]:
    """Return all historical rows for entity+attribute, ordered by recorded_at ASC."""
    rows = self._conn.execute(
        "SELECT * FROM fact_history WHERE entity=? AND attribute=? ORDER BY recorded_at ASC",
        (entity, attribute),
    ).fetchall()
    return [dict(r) for r in rows]

list_fact_history_by_entity(entity)

Return all historical rows for an entity, ordered by recorded_at ASC.

Source code in src/agentcrdt/store.py 
215
216
217
218
219
220
221
def list_fact_history_by_entity(self, entity: str) -> list[dict[str, Any]]:
    """Return all historical rows for an entity, ordered by recorded_at ASC."""
    rows = self._conn.execute(
        "SELECT * FROM fact_history WHERE entity=? ORDER BY recorded_at ASC",
        (entity,),
    ).fetchall()
    return [dict(r) for r in rows]

ChangeWatcher(store)

Watch a WorldStore for changes to specific entities or attributes.

Source code in src/agentcrdt/watch.py 
15
16
17
18
19
def __init__(self, store: WorldStore) -> None:
    self._store = store
    self._callbacks: list[tuple[str | None, str | None, Callable[..., Any]]] = []
    # Snapshot: "entity::attribute" -> WorldFact
    self._last_snapshot: dict[str, WorldFact] = self._take_snapshot()

on_change(entity=None, attribute=None)

Decorator: register a callback for when a matching fact changes.

Source code in src/agentcrdt/watch.py 
21
22
23
24
25
26
27
28
29
30
def on_change(
    self, entity: str | None = None, attribute: str | None = None
) -> Callable[..., Any]:
    """Decorator: register a callback for when a matching fact changes."""

    def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
        self._callbacks.append((entity, attribute, fn))
        return fn

    return decorator

check()

Check for new facts since last check. Returns changed facts and fires callbacks.

Source code in src/agentcrdt/watch.py 
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def check(self) -> list[WorldFact]:
    """Check for new facts since last check. Returns changed facts and fires callbacks."""
    current = self._take_snapshot()
    changed: list[WorldFact] = []

    for key, fact in current.items():
        old_fact = self._last_snapshot.get(key)
        if old_fact is None or (
            fact.version != old_fact.version
            or fact.timestamp != old_fact.timestamp
            or fact.value != old_fact.value
        ):
            changed.append(fact)

    self._last_snapshot = current

    for fact in changed:
        self._fire(fact)

    return changed

snapshot()

Return current fact snapshot (entity::attribute -> fact).

Source code in src/agentcrdt/watch.py 
53
54
55
def snapshot(self) -> dict[str, WorldFact]:
    """Return current fact snapshot (entity::attribute -> fact)."""
    return dict(self._last_snapshot)

assert_multi_agent_ok(store, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_multi_agent is ok.

Source code in src/agentcrdt/closed_loop.py 
562
563
564
565
566
567
def assert_multi_agent_ok(store: WorldStore, **kwargs: Any) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_multi_agent` is ok."""
    outcome = gate_multi_agent(store, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_mutable_write(fact, **kwargs)

Raise :class:ClosedLoopError if fact is a constant-domain write.

Source code in src/agentcrdt/closed_loop.py 
356
357
358
359
360
361
362
363
364
def assert_mutable_write(
    fact: WorldFact,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` if *fact* is a constant-domain write."""
    outcome = refuse_constant_write(fact, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

assert_world_state_ok(store, **kwargs)

Gate and raise :class:ClosedLoopError unless outcome is ok.

Source code in src/agentcrdt/closed_loop.py 
345
346
347
348
349
350
351
352
353
def assert_world_state_ok(
    store: WorldStore | list[WorldFact],
    **kwargs: Any,
) -> GateOutcome:
    """Gate and raise :class:`ClosedLoopError` unless outcome is ok."""
    outcome = gate_world_state(store, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

detect_silent_divergences(store)

Find fact keys with multi-agent multi-value history and no contradiction event.

MAST / ICLR multi-agent failure class: agents write incompatible values for the same domain.entity.attribute; LWW keeps one winner and no :class:~agentcrdt.fact.ContradictionEvent is recorded - consumers see a single "truth" that is actually contested.

Uses fact_history rows. A divergence requires:

  • ≥ 2 distinct agent_ids (non-empty preferred)
  • ≥ 2 distinct values
  • no store event that lists this fact_id in facts_involved
Source code in src/agentcrdt/closed_loop.py 
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
def detect_silent_divergences(store: WorldStore) -> list[ValueDivergence]:
    """Find fact keys with multi-agent multi-value history and no contradiction event.

    MAST / ICLR multi-agent failure class: agents write incompatible values for
    the same ``domain.entity.attribute``; LWW keeps one winner and **no**
    :class:`~agentcrdt.fact.ContradictionEvent` is recorded - consumers see a
    single "truth" that is actually contested.

    Uses ``fact_history`` rows. A divergence requires:

    * ≥ 2 distinct agent_ids (non-empty preferred)
    * ≥ 2 distinct values
    * no store event that lists this ``fact_id`` in ``facts_involved``
    """
    if not isinstance(store, WorldStore):
        raise TypeError("detect_silent_divergences requires WorldStore (needs history)")

    events = store.list_events()
    covered: set[str] = set()
    for ev in events:
        for fid in ev.facts_involved:
            covered.add(fid)

    facts = store.list_facts()
    out: list[ValueDivergence] = []
    seen_keys: set[str] = set()

    for fact in facts:
        if fact.id in covered:
            continue
        hist = store.list_fact_history(fact.id)
        if len(hist) < 2:
            continue
        values: set[str] = set()
        agents: set[str] = set()
        for row in hist:
            values.add(str(row.get("value", "")))
            # Blank agent_id → "anonymous" so multi-blank writers still count.
            aid = str(row.get("agent_id") or "").strip() or "anonymous"
            agents.add(aid)
        if len(values) < 2:
            continue
        if len(agents) < 2:
            # Single agent overwriting itself is versioning, not MAST conflict.
            continue

        key = f"{fact.domain}.{fact.entity}.{fact.attribute}"
        if key in seen_keys:
            continue
        seen_keys.add(key)
        out.append(
            ValueDivergence(
                fact_id=fact.id,
                domain=fact.domain,
                entity=fact.entity,
                attribute=fact.attribute,
                agents=tuple(sorted(agents)),
                values=tuple(sorted(values)),
                version_count=len(hist),
            )
        )
    return out

gate_merge_result(result, *, max_conflicts=0, min_merged=0)

Gate a :class:~agentcrdt.merger.MergeResult after multi-agent merge.

  • merged_count < min_merged → FAIL_LOUD (empty merge ornament)
  • len(conflicts) > max_conflicts → FAIL (human_required)
  • clean merge → PASS
Source code in src/agentcrdt/closed_loop.py 
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
def gate_merge_result(
    result: MergeResult,
    *,
    max_conflicts: int = 0,
    min_merged: int = 0,
) -> GateOutcome:
    """Gate a :class:`~agentcrdt.merger.MergeResult` after multi-agent merge.

    * ``merged_count < min_merged`` → FAIL_LOUD (empty merge ornament)
    * ``len(conflicts) > max_conflicts`` → FAIL (human_required)
    * clean merge → PASS
    """
    if not isinstance(result, MergeResult):
        return _fail_loud(
            "MAST: gate_merge_result requires MergeResult",
            human_required=True,
        )
    n_merged = int(result.merged_count)
    n_conf = len(result.conflicts or [])
    if n_merged < min_merged:
        return _fail_loud(
            f"MAST: merge merged_count={n_merged} < min_merged={min_merged} - empty/ornament merge",
            fact_count=n_merged,
            conflict_count=n_conf,
            human_required=True,
        )
    if n_conf > max_conflicts:
        return _fail(
            f"MAST: merge produced {n_conf} conflict(s) (max={max_conflicts}) "
            f"- refuse post-merge continue without resolution",
            fact_count=n_merged,
            conflict_count=n_conf,
            human_required=True,
        )
    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=f"merge ok: merged={n_merged} conflicts={n_conf}",
        exit_code=0,
        fact_count=n_merged,
        conflict_count=n_conf,
        human_required=False,
    )

gate_multi_agent(store, *, max_unresolved_events=0, refuse_silent_divergence=True, require_facts=True)

Gate a world store for MAST multi-agent coordination failures.

Load-bearing controls:

  1. Empty store → FAIL_LOUD (when require_facts).
  2. Unresolved :class:~agentcrdt.fact.ContradictionEvent count above max_unresolved_events → FAIL (human_required).
  3. Silent value divergences (history multi-agent multi-value, no event) → FAIL when refuse_silent_divergence (MAST silent LWW).

Pair with :func:gate_world_state for CONST-AS-STATE domain checks.

Source code in src/agentcrdt/closed_loop.py 
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
def gate_multi_agent(
    store: WorldStore,
    *,
    max_unresolved_events: int = 0,
    refuse_silent_divergence: bool = True,
    require_facts: bool = True,
) -> GateOutcome:
    """Gate a world store for MAST multi-agent coordination failures.

    Load-bearing controls:

    1. Empty store → **FAIL_LOUD** (when ``require_facts``).
    2. Unresolved :class:`~agentcrdt.fact.ContradictionEvent` count above
       ``max_unresolved_events`` → **FAIL** (``human_required``).
    3. Silent value divergences (history multi-agent multi-value, no event) →
       **FAIL** when ``refuse_silent_divergence`` (MAST silent LWW).

    Pair with :func:`gate_world_state` for CONST-AS-STATE domain checks.
    """
    if not isinstance(store, WorldStore):
        return _fail_loud(
            "MAST: gate_multi_agent requires WorldStore",
            human_required=True,
        )

    facts = store.list_facts()
    n = len(facts)
    if require_facts and n == 0:
        return _fail_loud(
            "MAST: empty world store - no multi-agent state to coordinate",
            fact_count=0,
            human_required=True,
        )

    events = store.list_events()
    n_events = len(events)
    if n_events > max_unresolved_events:
        contested = tuple(sorted({f"{e.agent_a}|{e.agent_b}|{e.rule}" for e in events[:20]})[:10])
        return _fail(
            f"MAST: {n_events} unresolved contradiction event(s) "
            f"(max={max_unresolved_events}) - multi-agent conflict not resolved "
            f"(ICLR/AgentPulse class)",
            fact_count=n,
            conflict_count=n_events,
            divergence_count=0,
            human_required=True,
            contested_keys=contested,
        )

    divergences: list[ValueDivergence] = []
    if refuse_silent_divergence:
        divergences = detect_silent_divergences(store)
        if divergences:
            keys = tuple(d.key for d in divergences[:10])
            return _fail(
                f"MAST: {len(divergences)} silent value divergence(s) "
                f"(multi-agent multi-value history without ContradictionEvent) "
                f"keys={list(keys)} - refuse silent LWW as truth "
                f"(AdaMAST/ICLR multi-agent failure class)",
                fact_count=n,
                conflict_count=n_events,
                divergence_count=len(divergences),
                human_required=True,
                contested_keys=keys,
            )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"multi-agent state ok: facts={n} events={n_events} "
            f"divergences=0 max_events={max_unresolved_events}"
        ),
        exit_code=0,
        fact_count=n,
        conflict_count=n_events,
        divergence_count=0,
        human_required=False,
    )

gate_world_state(store, *, extra_constant_domains=None, allow_mixed=False, require_mutable=True)

Gate a world store / fact list for CONST-AS-STATE discipline.

  • Empty → FAIL_LOUD (exit 2).
  • Only constant domains → FAIL (exit 1) - refuse constant-only CRDT.
  • Constants mixed with mutable → FAIL unless allow_mixed=True.
  • Mutable present (and no banned constants if not allow_mixed) → PASS.

Parameters:

Name Type Description Default
store WorldStore | list[WorldFact]

:class:WorldStore or list of :class:WorldFact.

 required
extra_constant_domains Iterable[str] | None

Additional domain names to treat as constants.

 None
allow_mixed bool

If True, constant facts are counted but do not fail when mutable facts also exist (still refuse constant-only).

 False
require_mutable bool

If True, at least one mutable fact is required.

 True
Source code in src/agentcrdt/closed_loop.py 
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
def gate_world_state(
    store: WorldStore | list[WorldFact],
    *,
    extra_constant_domains: Iterable[str] | None = None,
    allow_mixed: bool = False,
    require_mutable: bool = True,
) -> GateOutcome:
    """Gate a world store / fact list for CONST-AS-STATE discipline.

    * Empty → ``FAIL_LOUD`` (exit 2).
    * Only constant domains → ``FAIL`` (exit 1) - refuse constant-only CRDT.
    * Constants mixed with mutable → ``FAIL`` unless ``allow_mixed=True``.
    * Mutable present (and no banned constants if not allow_mixed) → ``PASS``.

    Args:
        store: :class:`WorldStore` or list of :class:`WorldFact`.
        extra_constant_domains: Additional domain names to treat as constants.
        allow_mixed: If True, constant facts are counted but do not fail when
            mutable facts also exist (still refuse constant-only).
        require_mutable: If True, at least one mutable fact is required.
    """
    facts = store.list_facts() if isinstance(store, WorldStore) else list(store)

    if len(facts) == 0:
        return _fail_loud(
            "empty world store - no load-bearing mutable state "
            "(CONST-AS-STATE: constant-only or empty is ornament)"
        )

    mutable, constant = classify_facts(facts, extra_constant_domains=extra_constant_domains)
    const_domains = tuple(sorted({_canonical_domain(f.domain) for f in constant}))

    if require_mutable and len(mutable) == 0:
        return _fail(
            f"CONST-AS-STATE: constant-only domains {list(const_domains)} - "
            f"refuse CRDT world state for recipes/code constants "
            f"(POLYMATTER_RECIPE class)",
            fact_count=len(facts),
            mutable_count=0,
            constant_count=len(constant),
            constant_domains=const_domains,
        )

    if constant and not allow_mixed:
        return _fail(
            f"CONST-AS-STATE: store mixes constant domains {list(const_domains)} "
            f"with mutable state - strip constants before merge",
            fact_count=len(facts),
            mutable_count=len(mutable),
            constant_count=len(constant),
            constant_domains=const_domains,
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"world state ok: mutable={len(mutable)} constant={len(constant)} "
            f"allow_mixed={allow_mixed}"
        ),
        exit_code=0,
        fact_count=len(facts),
        mutable_count=len(mutable),
        constant_count=len(constant),
        constant_domains=const_domains,
    )

is_constant_domain(domain, *, extra=None)

True if domain is a constant/recipe/code domain (CONST-AS-STATE).

Source code in src/agentcrdt/closed_loop.py 
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
def is_constant_domain(
    domain: str,
    *,
    extra: Iterable[str] | None = None,
) -> bool:
    """True if *domain* is a constant/recipe/code domain (CONST-AS-STATE)."""
    d = _canonical_domain(domain)
    if not d:
        return True  # empty domain is never valid mutable world state
    banned = set(DEFAULT_CONSTANT_DOMAINS)
    if extra:
        banned |= {_canonical_domain(x) for x in extra}
    if d in banned:
        return True
    # Prefix match: recipe_v2, constant_xyz, polymatter_*
    return any(d.startswith(b + "_") or d.endswith("_" + b) for b in banned)

refuse_constant_write(fact, *, extra_constant_domains=None)

Gate a single write: constant domains FAIL (do not set_fact).

Returns PASS only for mutable-domain facts.

Source code in src/agentcrdt/closed_loop.py 
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def refuse_constant_write(
    fact: WorldFact,
    *,
    extra_constant_domains: Iterable[str] | None = None,
) -> GateOutcome:
    """Gate a single write: constant domains FAIL (do not set_fact).

    Returns PASS only for mutable-domain facts.
    """
    if not fact.domain or not str(fact.domain).strip():
        return _fail_loud(
            "empty domain - refuse write",
            fact_count=1,
            constant_count=1,
        )
    if is_constant_domain(fact.domain, extra=extra_constant_domains):
        d = _canonical_domain(fact.domain)
        return _fail(
            f"CONST-AS-STATE: refuse write to constant domain {d!r} "
            f"(entity={fact.entity!r} attr={fact.attribute!r}) - "
            f"not multi-writer world state",
            fact_count=1,
            mutable_count=0,
            constant_count=1,
            constant_domains=(d,),
            refused_writes=1,
        )
    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=f"mutable domain {fact.domain!r} allowed",
        exit_code=0,
        fact_count=1,
        mutable_count=1,
        constant_count=0,
    )

set_fact_if_mutable(store, fact, *, extra_constant_domains=None)

Write fact only if domain is mutable; otherwise refuse (CONST-AS-STATE).

Source code in src/agentcrdt/closed_loop.py 
331
332
333
334
335
336
337
338
339
340
341
342
def set_fact_if_mutable(
    store: WorldStore,
    fact: WorldFact,
    *,
    extra_constant_domains: Iterable[str] | None = None,
) -> GateOutcome:
    """Write *fact* only if domain is mutable; otherwise refuse (CONST-AS-STATE)."""
    outcome = refuse_constant_write(fact, extra_constant_domains=extra_constant_domains)
    if not outcome.ok:
        return outcome
    store.set_fact(fact)
    return outcome

assert_no_covert_collusion(events, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_covert_collusion is ok.

Source code in src/agentcrdt/collusion.py 
355
356
357
358
359
360
361
362
363
def assert_no_covert_collusion(
    events: Sequence[AgentTraceEvent | dict[str, Any]] | None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_covert_collusion` is ok."""
    outcome = gate_covert_collusion(events, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

detect_covert_collusion(events, *, sync_window=1.0, min_agents_shared_payload=2, min_payload_rarity=3, min_side_channel_agents=2)

Black-box collusion detector over tool/timing/payload traces.

Signals (paper-inspired, behavioural only):

  1. shared_payload — same non-empty payload_fp appears for ≥ min_agents_shared_payload distinct agents, and the fingerprint is rare overall (count of events with that fp ≤ population / rarity).
  2. sync_timing — two agents fire the same tool within sync_window seconds more than once (pair-level).
  3. side_channel_code — identical non-empty side_channel token used by ≥ min_side_channel_agents agents.
Source code in src/agentcrdt/collusion.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
def detect_covert_collusion(
    events: Sequence[AgentTraceEvent | dict[str, Any]],
    *,
    sync_window: float = 1.0,
    min_agents_shared_payload: int = 2,
    min_payload_rarity: int = 3,
    min_side_channel_agents: int = 2,
) -> CollusionReport:
    """Black-box collusion detector over tool/timing/payload traces.

    Signals (paper-inspired, behavioural only):

    1. **shared_payload** — same non-empty ``payload_fp`` appears for
       ≥ ``min_agents_shared_payload`` distinct agents, and the fingerprint
       is rare overall (count of events with that fp ≤ population / rarity).
    2. **sync_timing** — two agents fire the same tool within ``sync_window``
       seconds more than once (pair-level).
    3. **side_channel_code** — identical non-empty ``side_channel`` token used
       by ≥ ``min_side_channel_agents`` agents.
    """
    parsed = [_as_event(e, i) for i, e in enumerate(events)]
    agents = {e.agent_id for e in parsed}
    signals: list[CollusionSignal] = []

    # 1) shared rare payloads
    by_fp: dict[str, set[str]] = defaultdict(set)
    fp_counts: dict[str, int] = defaultdict(int)
    for e in parsed:
        if not e.payload_fp:
            continue
        by_fp[e.payload_fp].add(e.agent_id)
        fp_counts[e.payload_fp] += 1
    shared_groups = 0
    n = max(len(parsed), 1)
    for fp, agent_set in by_fp.items():
        if len(agent_set) < min_agents_shared_payload:
            continue
        # rare: appears in fewer than n/min_payload_rarity events, or absolute small
        if fp_counts[fp] > max(min_payload_rarity, n // max(min_payload_rarity, 1)):
            # common payload — skip (not covert)
            if fp_counts[fp] > len(agents) * 2:
                continue
        shared_groups += 1
        signals.append(
            CollusionSignal(
                kind="shared_payload",
                agents=tuple(sorted(agent_set)),
                detail=f"payload_fp={fp[:48]} agents={len(agent_set)} count={fp_counts[fp]}",
                score=min(1.0, len(agent_set) / max(len(agents), 1)),
            )
        )

    # 2) sync timing same tool
    by_tool: dict[str, list[AgentTraceEvent]] = defaultdict(list)
    for e in parsed:
        by_tool[e.tool].append(e)
    sync_pairs = 0
    seen_pairs: set[tuple[str, str, str]] = set()
    for tool, evs in by_tool.items():
        ordered = sorted(evs, key=lambda x: x.timestamp)
        for i, a in enumerate(ordered):
            for b in ordered[i + 1 :]:
                if b.timestamp - a.timestamp > sync_window:
                    break
                if a.agent_id == b.agent_id:
                    continue
                pair = tuple(sorted((a.agent_id, b.agent_id))) + (tool,)  # type: ignore[operator]
                key = (pair[0], pair[1], tool)
                if key in seen_pairs:
                    continue
                # require at least 2 sync hits for this pair+tool to reduce noise
                hits = 0
                for x in ordered:
                    for y in ordered:
                        if x.agent_id == pair[0] and y.agent_id == pair[1]:
                            if abs(x.timestamp - y.timestamp) <= sync_window:
                                hits += 1
                if hits < 2:
                    continue
                seen_pairs.add(key)
                sync_pairs += 1
                signals.append(
                    CollusionSignal(
                        kind="sync_timing",
                        agents=(pair[0], pair[1]),
                        detail=f"tool={tool} window={sync_window}s hits~={hits}",
                        score=min(1.0, hits / 4.0),
                    )
                )

    # 3) side channel codes
    by_side: dict[str, set[str]] = defaultdict(set)
    for e in parsed:
        if not e.side_channel:
            continue
        by_side[e.side_channel].add(e.agent_id)
    side_groups = 0
    for code, agent_set in by_side.items():
        if len(agent_set) < min_side_channel_agents:
            continue
        side_groups += 1
        signals.append(
            CollusionSignal(
                kind="side_channel_code",
                agents=tuple(sorted(agent_set)),
                detail=f"code={code[:48]} agents={len(agent_set)}",
                score=min(1.0, len(agent_set) / max(len(agents), 1)),
            )
        )

    return CollusionReport(
        event_count=len(parsed),
        agent_count=len(agents),
        signals=tuple(signals),
        shared_payload_groups=shared_groups,
        sync_pairs=sync_pairs,
        side_channel_groups=side_groups,
    )

gate_covert_collusion(events, *, require_events=True, min_agents=2, max_signals=0, sync_window=1.0, refuse_shared_payload=True, refuse_sync_timing=True, refuse_side_channel=True)

Refuse multi-agent populations with covert collusion signals.

Rules:

  • No events when required → FAIL_LOUD
  • Fewer than min_agents → FAIL_LOUD (need a population)
  • Any signal kind over budget (default 0) → FAIL
  • Clean independent traces → PASS
Source code in src/agentcrdt/collusion.py 
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
def gate_covert_collusion(
    events: Sequence[AgentTraceEvent | dict[str, Any]] | None,
    *,
    require_events: bool = True,
    min_agents: int = 2,
    max_signals: int = 0,
    sync_window: float = 1.0,
    refuse_shared_payload: bool = True,
    refuse_sync_timing: bool = True,
    refuse_side_channel: bool = True,
) -> GateOutcome:
    """Refuse multi-agent populations with covert collusion signals.

    Rules:

    * No events when required → **FAIL_LOUD**
    * Fewer than ``min_agents`` → **FAIL_LOUD** (need a population)
    * Any signal kind over budget (default 0) → **FAIL**
    * Clean independent traces → **PASS**
    """
    if not events:
        if require_events:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=(
                    "COVERT-COLLUSION: no agent trace events — cannot run "
                    "black-box steganalysis without behavioural inventory "
                    "(arXiv 2608.02698)"
                ),
                exit_code=2,
                human_required=True,
            )
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason="COVERT-COLLUSION: no events required",
            exit_code=0,
        )

    try:
        report = detect_covert_collusion(events, sync_window=sync_window)
    except (TypeError, ValueError) as exc:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=f"COVERT-COLLUSION: invalid events: {exc}",
            exit_code=2,
            human_required=True,
        )

    if report.agent_count < min_agents:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=(
                f"COVERT-COLLUSION: agent_count={report.agent_count} < "
                f"min_agents={min_agents} — population-level detector needs "
                "multi-agent traces"
            ),
            exit_code=2,
            human_required=True,
            fact_count=report.event_count,
        )

    filtered: list[CollusionSignal] = []
    for s in report.signals:
        if s.kind == "shared_payload" and not refuse_shared_payload:
            continue
        if s.kind == "sync_timing" and not refuse_sync_timing:
            continue
        if s.kind == "side_channel_code" and not refuse_side_channel:
            continue
        filtered.append(s)

    if len(filtered) > max_signals:
        kinds = sorted({s.kind for s in filtered})
        agents = sorted({a for s in filtered for a in s.agents})
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"COVERT-COLLUSION: {len(filtered)} signal(s) kinds={kinds} "
                f"agents={agents[:8]} score={report.collusion_score:.2f} "
                f"shared_payload={report.shared_payload_groups} "
                f"sync_pairs={report.sync_pairs} side_channel={report.side_channel_groups} "
                f"— refuse treating agents as independent (arXiv 2608.02698)"
            ),
            exit_code=1,
            human_required=True,
            fact_count=report.event_count,
            divergence_count=len(filtered),
            contested_keys=tuple(kinds),
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"COVERT-COLLUSION ok: events={report.event_count} "
            f"agents={report.agent_count} signals={len(filtered)}"
        ),
        exit_code=0,
        human_required=False,
        fact_count=report.event_count,
        divergence_count=0,
    )

analyze_comm_attacks(messages, *, system_agents=None, trusted_external_senders=None, privileged_senders=None, architecture='dmas', injection_phrases=None)

Detect External Entry Point and Privileged In-System attack signals.

Does not gate; use :func:gate_comm_integrity.

Source code in src/agentcrdt/comm_attack.py 
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
def analyze_comm_attacks(
    messages: Sequence[Any] | None,
    *,
    system_agents: Sequence[str] | None = None,
    trusted_external_senders: Sequence[str] | None = None,
    privileged_senders: Sequence[str] | None = None,
    architecture: str = "dmas",
    injection_phrases: Iterable[str] | None = None,
) -> CommAttackReport:
    """Detect External Entry Point and Privileged In-System attack signals.

    Does not gate; use :func:`gate_comm_integrity`.
    """
    parsed = [_as_message(m, i) for i, m in enumerate(messages or [])]
    arch = _canon(architecture) or "dmas"
    system = {_canon(a) for a in (system_agents or []) if str(a).strip()}
    if not system:
        # infer system agents from non-external roles
        system = {
            _canon(m.sender)
            for m in parsed
            if not is_external_role(m.role) and _canon(m.sender)
        }
    trusted_ext = {_canon(a) for a in (trusted_external_senders or []) if str(a).strip()}
    priv_senders = {_canon(a) for a in (privileged_senders or []) if str(a).strip()}
    # privileged senders default: system agents with privileged role messages
    if not priv_senders:
        priv_senders = {
            _canon(m.sender)
            for m in parsed
            if is_privileged_role(m.role) and _canon(m.sender) in system
        }

    signals: list[CommAttackSignal] = []
    n_ext = n_priv = n_inj = 0

    for m in parsed:
        sid = _canon(m.sender)
        role = _canon(m.role)
        ch = _canon(m.channel)

        # External Entry Point Attack
        externalish = is_external_role(role) or (system and sid not in system)
        if externalish and sid not in trusted_ext:
            n_ext += 1
            signals.append(
                CommAttackSignal(
                    kind="external_entry",
                    msg_id=m.msg_id,
                    detail=(
                        f"sender={m.sender!r} role={m.role!r} not in trusted "
                        f"external entry points — External Entry Point Attack "
                        f"(arXiv 2608.06830)"
                    ),
                    score=1.0,
                )
            )

        # Privileged In-System Attack: privileged channel/role without auth
        uses_priv_plane = is_privileged_channel(ch) or is_privileged_role(role)
        if uses_priv_plane and sid not in priv_senders and sid not in trusted_ext:
            n_priv += 1
            signals.append(
                CommAttackSignal(
                    kind="privileged_spoof",
                    msg_id=m.msg_id,
                    detail=(
                        f"sender={m.sender!r} role={m.role!r} channel={m.channel!r} "
                        f"uses privileged plane without grant — Privileged "
                        f"In-System Attack (arXiv 2608.06830)"
                    ),
                    score=1.0,
                )
            )

        # Content injection in coordination messages
        hits = detect_comm_injection_phrases(m.content, phrases=injection_phrases)
        if hits:
            n_inj += 1
            signals.append(
                CommAttackSignal(
                    kind="content_injection",
                    msg_id=m.msg_id,
                    detail=f"phrases={hits[:4]} in msg from {m.sender!r}",
                    score=1.0,
                )
            )

    agents = {_canon(m.sender) for m in parsed} | {
        _canon(m.receiver) for m in parsed if m.receiver not in {"*", "broadcast", ""}
    }

    return CommAttackReport(
        message_count=len(parsed),
        agent_count=len({a for a in agents if a}),
        signals=tuple(signals),
        external_entry_count=n_ext,
        privileged_spoof_count=n_priv,
        injection_count=n_inj,
        architecture=arch,
        details={
            "system_agents": sorted(system),
            "trusted_external": sorted(trusted_ext),
            "privileged_senders": sorted(priv_senders),
        },
    )

assert_comm_integrity(messages, **kwargs)

Raise :class:ClosedLoopError unless :func:gate_comm_integrity is ok.

Source code in src/agentcrdt/comm_attack.py 
452
453
454
455
456
457
458
459
460
def assert_comm_integrity(
    messages: Sequence[Any] | None,
    **kwargs: Any,
) -> GateOutcome:
    """Raise :class:`ClosedLoopError` unless :func:`gate_comm_integrity` is ok."""
    outcome = gate_comm_integrity(messages, **kwargs)
    if not outcome.ok:
        raise ClosedLoopError(f"{outcome.verdict}: {outcome.reason}")
    return outcome

detect_comm_injection_phrases(text, *, phrases=None)

Return coordination-injection phrases found in text.

Source code in src/agentcrdt/comm_attack.py 
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
def detect_comm_injection_phrases(
    text: str,
    *,
    phrases: Iterable[str] | None = None,
) -> list[str]:
    """Return coordination-injection phrases found in *text*."""
    blob = (text or "").lower()
    if not blob:
        return []
    found: list[str] = []
    for p in phrases if phrases is not None else DEFAULT_COMM_INJECTION_PHRASES:
        pl = p.lower()
        if pl and pl in blob:
            found.append(p)
    return found

gate_comm_integrity(messages, *, system_agents=None, trusted_external_senders=None, privileged_senders=None, architecture='dmas', claim_coordinated=False, require_messages=True, refuse_external_entry=True, refuse_privileged_spoof=True, refuse_content_injection=True, max_signals=0, injection_phrases=None)

Refuse multi-agent coordination when communication channel is attacked.

Public case: arXiv 2608.06830 — External Entry Point and Privileged In-System attacks against LLM-controlled multi-robot / multi-agent systems under DMAS / HMAS architectures.

Rules:

  1. claim_coordinated with zero messages → FAIL_LOUD
  2. Empty inventory when required → FAIL_LOUD
  3. External Entry Point (untrusted external sender) → FAIL
  4. Privileged In-System spoof (non-privileged on control plane) → FAIL
  5. Content injection phrases in messages → FAIL
  6. Signal count above max_signals (default 0) → FAIL
  7. Clean internal mesh → PASS
Source code in src/agentcrdt/comm_attack.py 
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
def gate_comm_integrity(
    messages: Sequence[Any] | None,
    *,
    system_agents: Sequence[str] | None = None,
    trusted_external_senders: Sequence[str] | None = None,
    privileged_senders: Sequence[str] | None = None,
    architecture: str = "dmas",
    claim_coordinated: bool = False,
    require_messages: bool = True,
    refuse_external_entry: bool = True,
    refuse_privileged_spoof: bool = True,
    refuse_content_injection: bool = True,
    max_signals: int = 0,
    injection_phrases: Iterable[str] | None = None,
) -> GateOutcome:
    """Refuse multi-agent coordination when communication channel is attacked.

    Public case: arXiv 2608.06830 — External Entry Point and Privileged
    In-System attacks against LLM-controlled multi-robot / multi-agent systems
    under DMAS / HMAS architectures.

    Rules:

    1. ``claim_coordinated`` with zero messages → **FAIL_LOUD**
    2. Empty inventory when required → **FAIL_LOUD**
    3. External Entry Point (untrusted external sender) → **FAIL**
    4. Privileged In-System spoof (non-privileged on control plane) → **FAIL**
    5. Content injection phrases in messages → **FAIL**
    6. Signal count above ``max_signals`` (default 0) → **FAIL**
    7. Clean internal mesh → **PASS**
    """
    if not messages:
        if claim_coordinated or require_messages:
            return GateOutcome(
                ok=False,
                verdict="FAIL_LOUD",
                reason=(
                    "COMM-ATTACK: empty message inventory — cannot authorize "
                    "multi-agent coordination without communication log "
                    f"(claim_coordinated={claim_coordinated}; arXiv 2608.06830)"
                ),
                exit_code=2,
                human_required=True,
            )
        return GateOutcome(
            ok=True,
            verdict="PASS",
            reason="COMM-ATTACK: no messages required",
            exit_code=0,
        )

    try:
        report = analyze_comm_attacks(
            messages,
            system_agents=system_agents,
            trusted_external_senders=trusted_external_senders,
            privileged_senders=privileged_senders,
            architecture=architecture,
            injection_phrases=injection_phrases,
        )
    except (TypeError, ValueError) as exc:
        return GateOutcome(
            ok=False,
            verdict="FAIL_LOUD",
            reason=f"COMM-ATTACK: invalid messages: {exc}",
            exit_code=2,
            human_required=True,
        )

    filtered: list[CommAttackSignal] = []
    for s in report.signals:
        if s.kind == "external_entry" and not refuse_external_entry:
            continue
        if s.kind == "privileged_spoof" and not refuse_privileged_spoof:
            continue
        if s.kind == "content_injection" and not refuse_content_injection:
            continue
        filtered.append(s)

    if len(filtered) > max_signals:
        kinds = sorted({s.kind for s in filtered})
        return GateOutcome(
            ok=False,
            verdict="FAIL",
            reason=(
                f"COMM-ATTACK: {len(filtered)} communication attack signal(s) "
                f"kinds={kinds} architecture={report.architecture} — refuse "
                f"coordination under External Entry / Privileged In-System "
                f"threat class (arXiv 2608.06830)"
            ),
            exit_code=1,
            human_required=True,
            fact_count=report.message_count,
            conflict_count=len(filtered),
            contested_keys=tuple(s.msg_id for s in filtered[:12]),
        )

    return GateOutcome(
        ok=True,
        verdict="PASS",
        reason=(
            f"COMM-ATTACK ok: messages={report.message_count} "
            f"agents={report.agent_count} architecture={report.architecture} "
            f"signals=0"
        ),
        exit_code=0,
        fact_count=report.message_count,
        human_required=False,
    )

conflicts_for_entity(store, entity)

Get all conflicts for a specific entity.

Source code in src/agentcrdt/conflict_report.py 
82
83
84
85
86
87
def conflicts_for_entity(store: WorldStore, entity: str) -> list[ContradictionEvent]:
    """Get all conflicts for a specific entity."""
    facts_for_entity = {f.id for f in store.list_facts() if f.entity == entity}
    return [
        e for e in store.list_events() if any(fid in facts_for_entity for fid in e.facts_involved)
    ]