AID-D-009.002: Cross-Agent Hallucination Cascade Detection
Detect when a factual claim classified by
AID-D-009.001 as CONTRADICTED, UNVERIFIED, INCONCLUSIVE, or POISONED is corroborated, rebroadcast, persisted, or amplified across agent-to-agent messages, shared context, memory, delegation chains, or peer quorum. Correlate a complete authenticated claim-occurrence graph with a separately signed AID-D-009.001 state export; peer agreement is propagation evidence only and never establishes truth or action approval. Pillars: app, data
Lifecycle: operation
Scope boundary
Owns reconstruction of authenticated content-level claim occurrence lineage and emits findings when contradicted, unverified, inconclusive, or poisoned claims are corroborated, rebroadcast, persisted, or amplified across agent boundaries. Peer agreement is propagation evidence only.
AID-D-009.001: AID-D-009.002 tracks propagation of a claim after joining to its signed verification state; AID-D-009.001 owns the factual claim-state determination. The cascade detector cannot infer truth from quorum or occurrence count.
AID-D-011.002: AID-D-009.002 reconstructs content-level claim propagation; AID-D-011.002 verifies producer identity, event schema, transport, route, delegation, sequence, and interaction evidence. A message can be authentic yet propagate a false claim, and an accurate claim can arrive through forged or replayed interaction evidence.
AID-D-016.001: AID-D-009.002 follows a claim across authenticated agent nodes; AID-D-016.001 discovers unknown or newly emergent agents and graph topology. Content lineage assumes agent occurrence identities, while agent discovery determines whether those identities belong in the observed graph.
Implementation guidance
- AID-D-009.002-G001: Reconstruct the complete authenticated claim-occurrence graph, join it to the separately signed claim-state export, and detect policy-breaching propagation of contradicted, unverified, inconclusive, or poisoned claims.
AID-D-009.002-G001
Reconstruct the complete authenticated claim-occurrence graph, join it to the separately signed claim-state export, and detect policy-breaching propagation of contradicted, unverified, inconclusive, or poisoned claims.
Observation and truth boundary
Collect privacy-bounded occurrences at the agent message bus, shared-context service, memory service, delegation broker, and peer-corroboration path. Events carry occurrence and lineage facts only; they never self-declare factual truth. A separately signed AID-D-009.001 export supplies the final verification state. An independently signed ingress receipt proves the exact pre-normalization UTF-8 digest before raw text is discarded. This executable adapter joins only a stable canonical claim ID with identical raw bytes; it also rejects assigning the same raw digest to multiple canonical IDs. A rephrase is INSUFFICIENT_DATA until a separate adapter verifies a signed equivalence receipt. Embedding or LLM similarity alone never merges claims.
Verify the complete population and fail safely
The collection authority signs a fresh manifest that binds the exact event, state, raw-digest-receipt, and canonicalizer-artifact bytes; required sources; per-source sequence checkpoints; eligible source-event IDs; window; tenant; and workflow. The detector copies every mutable input and its signature into one mode-0700 private snapshot before verification, and evaluates only that verified snapshot. The prior atomically published, detector-signed generation is the trusted last-accepted checkpoint: every later manifest must use a strictly greater sequence and bind the exact prior result digest. The state directory must be owned and writable only by the detector identity, while verification keys remain in a separate read-only trust store. Missing, partial, unsigned, stale, rolled-back, cross-scope, or internally inconsistent input is ERROR, never NO_FINDING. Build the full graph before joining a later state so a contradiction discovered after propagation is applied retroactively to every affected component.
set -euo pipefail
umask 077
: "$D009002_TRUSTED_GENERATION_PATH"
OUTPUT=$D009002_TRUSTED_GENERATION_PATH
STATE_DIR=$(dirname -- "$OUTPUT")
test -d "$STATE_DIR"
test ! -L "$STATE_DIR"
test "$(stat -c %u "$STATE_DIR")" -eq "$(id -u)"
test "$(stat -c %a "$STATE_DIR")" = 700
exec 9>"$STATE_DIR/.cascade-generation.lock"
flock -x 9
WORK=$(mktemp -d "$STATE_DIR/.cascade-generation.tmp.XXXXXX")
trap 'rm -rf "$WORK"' EXIT
SNAPSHOT="$WORK/input"
PREVIOUS="$WORK/previous"
mkdir -m 700 "$SNAPSHOT" "$PREVIOUS"
RESULT="$WORK/result.json"
SIGNATURE="$WORK/result.sig"
BUNDLE="$WORK/generation.tar"
capture() { install -m 0400 -- "$1" "$SNAPSHOT/$2"; }
capture policy/cascade-policy.json policy.json
capture policy/cascade-policy.sig policy.sig
capture detection/claim-canonicalizer.wasm canonicalizer.wasm
capture detection/claim-canonicalizer.wasm.sig canonicalizer.sig
capture evidence/claim-lineage.jsonl events.jsonl
capture evidence/claim-lineage.jsonl.sig events.sig
capture evidence/claim-states.jsonl states.jsonl
capture evidence/claim-states.jsonl.sig states.sig
capture evidence/raw-claim-digest-receipts.jsonl raw-receipts.jsonl
capture evidence/raw-claim-digest-receipts.jsonl.sig raw-receipts.sig
capture evidence/claim-lineage-population.json manifest.json
capture evidence/claim-lineage-population.json.sig manifest.sig
cosign verify-blob --key keys/cascade-policy.pub --bundle "$SNAPSHOT/policy.sig" "$SNAPSHOT/policy.json"
cosign verify-blob --key keys/claim-canonicalizer.pub --bundle "$SNAPSHOT/canonicalizer.sig" "$SNAPSHOT/canonicalizer.wasm"
cosign verify-blob --key keys/lineage-export.pub --bundle "$SNAPSHOT/events.sig" "$SNAPSHOT/events.jsonl"
cosign verify-blob --key keys/claim-state-export.pub --bundle "$SNAPSHOT/states.sig" "$SNAPSHOT/states.jsonl"
cosign verify-blob --key keys/raw-claim-ingress.pub --bundle "$SNAPSHOT/raw-receipts.sig" "$SNAPSHOT/raw-receipts.jsonl"
cosign verify-blob --key keys/lineage-population.pub --bundle "$SNAPSHOT/manifest.sig" "$SNAPSHOT/manifest.json"
PREVIOUS_RESULT=-
if [ -e "$OUTPUT" ]; then
test -f "$OUTPUT"
test ! -L "$OUTPUT"
test "$(tar -tf "$OUTPUT")" = "$(printf 'result.json
result.sig')"
tar --extract --no-same-owner --no-same-permissions -f "$OUTPUT" -C "$PREVIOUS" result.json result.sig
cosign verify-blob --key keys/cascade-detector.pub --bundle "$PREVIOUS/result.sig" "$PREVIOUS/result.json"
PREVIOUS_RESULT="$PREVIOUS/result.json"
fi
set +e
python detection/claim_cascade.py "$SNAPSHOT/policy.json" "$SNAPSHOT/canonicalizer.wasm" "$SNAPSHOT/events.jsonl" "$SNAPSHOT/states.jsonl" "$SNAPSHOT/raw-receipts.jsonl" "$SNAPSHOT/manifest.json" "$PREVIOUS_RESULT" "$RESULT"
status=$?
set -e
if [ "$status" -le 1 ]; then
test -s "$RESULT"
cosign sign-blob --yes --key env://CASCADE_DETECTOR_SIGNING_KEY --bundle "$SIGNATURE" "$RESULT"
cosign verify-blob --key keys/cascade-detector.pub --bundle "$SIGNATURE" "$RESULT"
tar --sort=name --mtime=@0 --owner=0 --group=0 --numeric-owner -C "$WORK" -cf "$BUNDLE" result.json result.sig
sync -f "$BUNDLE"
mv -f -- "$BUNDLE" "$OUTPUT"
sync -f "$OUTPUT"
fi
exit "$status"Executable claim-lineage detector
# File: detection/claim_cascade.py
from __future__ import annotations
import hashlib
import json
import math
import re
import sys
from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
BAD_STATES = {"CONTRADICTED", "UNVERIFIED", "INCONCLUSIVE", "POISONED"}
CLAIM_STATES = BAD_STATES | {"VERIFIED"}
EDGE_TYPES = {
"origin", "a2a_message", "shared_context", "memory_write",
"memory_read", "delegation", "peer_corroboration", "boundary_stub",
}
SHA256 = re.compile(r"^[0-9a-f]{64}$")
EVENT_FIELDS = {
"schema_version", "event_id", "source_event_id", "source_id",
"source_sequence", "occurred_at", "canonical_claim_id",
"raw_claim_sha256", "canonicalizer_version", "canonicalizer_sha256",
"raw_claim_receipt_sha256",
"source_agent_id", "target_agent_id", "tenant_id", "workflow_id",
"edge_type", "parent_event_ids",
}
RAW_RECEIPT_FIELDS = {
"schema_version", "receipt_id", "source_id", "source_event_id",
"captured_at", "tenant_id", "workflow_id", "raw_claim_sha256",
"digest_algorithm", "digest_stage", "canonicalizer_version",
"canonicalizer_sha256",
}
STATE_FIELDS = {
"schema_version", "state_id", "state_source_id", "state_sequence",
"evaluated_at", "canonical_claim_id", "raw_claim_sha256", "decision",
"source_control", "evidence_population_sha256", "policy_sha256",
}
MANIFEST_FIELDS = {
"schema_version", "manifest_sequence", "window_id", "tenant_id",
"workflow_id", "window_start", "window_end", "issued_at", "expires_at",
"previous_result_sha256", "collection_complete", "required_sources",
"canonicalizer_version", "canonicalizer_sha256",
"source_checkpoints", "eligible_source_event_ids",
"eligible_source_event_ids_sha256", "event_count", "events_sha256",
"state_count", "states_sha256", "raw_receipt_count", "raw_receipts_sha256",
}
CHECKPOINT_FIELDS = {"first", "last", "count", "complete"}
POLICY_FIELDS = {
"schema_version", "policy_version", "canonicalizer_version",
"canonicalizer_sha256", "detector_build_sha256",
"minimum_manifest_sequence", "maximum_evidence_age_seconds",
"max_clock_skew_seconds", "required_sources", "authorized_state_controls",
"limits", "forbid_cross_tenant",
}
LIMIT_FIELDS = {
"max_distinct_agents", "max_depth", "max_fan_out",
"max_memory_writes", "max_agent_reentries",
}
PREVIOUS_RESULT_FIELDS = {
"schema_version", "policy_version", "canonicalizer_version",
"canonicalizer_sha256", "detector_build_sha256", "manifest_sequence",
"window_id", "manifest_issued_at", "manifest_expires_at",
"previous_result_sha256", "input_sha256", "event_count",
"state_record_count", "raw_claim_receipt_sha256", "claim_count",
"finding_status", "findings",
}
def strict_json(text: str) -> object:
def reject_constant(value: str) -> None:
raise ValueError(f"non-finite JSON constant: {value}")
def reject_duplicates(pairs: list[tuple[str, object]]) -> dict:
value = {}
for key, item in pairs:
if key in value:
raise ValueError(f"duplicate JSON key: {key}")
value[key] = item
return value
return json.loads(
text, parse_constant=reject_constant,
object_pairs_hook=reject_duplicates,
)
def timestamp(value: object) -> datetime:
if not isinstance(value, str):
raise ValueError("occurred_at must be text")
parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
if parsed.tzinfo is None:
raise ValueError("occurred_at lacks timezone")
return parsed.astimezone(timezone.utc)
def nonnegative_int(value: object, field: str) -> int:
if isinstance(value, bool) or not isinstance(value, int) or value < 0:
raise ValueError(f"{field} must be a non-negative integer")
return value
def load_policy(path: Path, canonicalizer_path: Path) -> dict:
policy = strict_json(path.read_text(encoding="utf-8"))
if not isinstance(policy, dict) or set(policy) != POLICY_FIELDS:
raise ValueError("cascade policy schema differs")
if (
policy["schema_version"] != "aidefend.claim-cascade-policy.v1"
or not isinstance(policy["policy_version"], str)
or not policy["policy_version"]
or not isinstance(policy["canonicalizer_version"], str)
or not policy["canonicalizer_version"]
or not isinstance(policy["canonicalizer_sha256"], str)
or not SHA256.fullmatch(policy["canonicalizer_sha256"])
or not isinstance(policy["detector_build_sha256"], str)
or not SHA256.fullmatch(policy["detector_build_sha256"])
or not isinstance(policy["forbid_cross_tenant"], bool)
):
raise ValueError("cascade policy identity or flags are invalid")
if (
hashlib.sha256(canonicalizer_path.read_bytes()).hexdigest()
!= policy["canonicalizer_sha256"]
):
raise ValueError("verified canonicalizer artifact differs from policy")
policy["minimum_manifest_sequence"] = nonnegative_int(
policy["minimum_manifest_sequence"], "minimum_manifest_sequence"
)
evidence_age = policy["maximum_evidence_age_seconds"]
if (
isinstance(evidence_age, bool)
or not isinstance(evidence_age, (int, float))
or not math.isfinite(float(evidence_age))
or evidence_age <= 0
):
raise ValueError("maximum_evidence_age_seconds must be finite and positive")
skew = policy["max_clock_skew_seconds"]
if (
isinstance(skew, bool)
or not isinstance(skew, (int, float))
or not math.isfinite(float(skew))
or skew < 0
):
raise ValueError("max_clock_skew_seconds must be finite and non-negative")
sources = policy["required_sources"]
if (
not isinstance(sources, list)
or not sources
or len(sources) != len(set(sources))
or any(not isinstance(source, str) or not source for source in sources)
):
raise ValueError("required source registry is invalid")
authorities = policy["authorized_state_controls"]
if not isinstance(authorities, dict) or set(authorities) != CLAIM_STATES:
raise ValueError("claim-state authority registry differs")
for decision, controls in authorities.items():
if (
not isinstance(controls, list)
or not controls
or len(controls) != len(set(controls))
or any(not isinstance(control, str) or not control for control in controls)
):
raise ValueError(f"claim-state authority is invalid: {decision}")
limits = policy["limits"]
if not isinstance(limits, dict) or set(limits) != LIMIT_FIELDS:
raise ValueError("cascade limit schema differs")
for field in LIMIT_FIELDS:
limits[field] = nonnegative_int(limits[field], f"limits.{field}")
return policy
def load_previous_result(path: Path) -> tuple[dict | None, str | None]:
if str(path) == "-":
return None, None
raw = path.read_bytes()
previous = strict_json(raw.decode("utf-8"))
if (
not isinstance(previous, dict)
or set(previous) != PREVIOUS_RESULT_FIELDS
or previous["schema_version"] != "aidefend.claim-cascade-result.v2"
or previous["finding_status"] not in {"FINDING", "NO_FINDING"}
):
raise ValueError("last-accepted cascade result schema differs")
nonnegative_int(previous["manifest_sequence"], "previous.manifest_sequence")
return previous, hashlib.sha256(raw).hexdigest()
def load_events(path: Path, policy: dict) -> list[dict]:
lines = path.read_text(encoding="utf-8").splitlines()
if any(not line.strip() for line in lines):
raise ValueError("lineage export contains blank records")
events, event_ids, source_event_ids, source_positions = [], set(), set(), set()
for line in lines:
event = strict_json(line)
if not isinstance(event, dict) or set(event) != EVENT_FIELDS:
raise ValueError("lineage event schema differs")
if event["schema_version"] != "aidefend.claim-lineage-event.v2":
raise ValueError("lineage event version differs")
event_id = event["event_id"]
if not isinstance(event_id, str) or not event_id or event_id in event_ids:
raise ValueError("event ID is missing or duplicated")
event_ids.add(event_id)
source_event_id = event["source_event_id"]
if (
not isinstance(source_event_id, str)
or not source_event_id
or source_event_id in source_event_ids
):
raise ValueError("source event ID is missing or duplicated")
source_event_ids.add(source_event_id)
source_id = event["source_id"]
if source_id not in policy["required_sources"]:
raise ValueError("lineage event source is outside signed policy")
event["source_sequence"] = nonnegative_int(
event["source_sequence"], "source_sequence"
)
position = (source_id, event["source_sequence"])
if position in source_positions:
raise ValueError("source sequence is duplicated")
source_positions.add(position)
event["_time"] = timestamp(event["occurred_at"])
if (
not isinstance(event["canonical_claim_id"], str)
or not event["canonical_claim_id"]
or not isinstance(event["raw_claim_sha256"], str)
or not SHA256.fullmatch(event["raw_claim_sha256"])
or event["canonicalizer_version"] != policy["canonicalizer_version"]
or event["canonicalizer_sha256"] != policy["canonicalizer_sha256"]
):
raise ValueError("canonical claim binding is invalid")
receipt = event["raw_claim_receipt_sha256"]
if not isinstance(receipt, str) or not SHA256.fullmatch(receipt):
raise ValueError("raw-claim receipt digest is invalid")
if event["edge_type"] not in EDGE_TYPES:
raise ValueError("lineage edge type is invalid")
for field in (
"source_id", "source_agent_id", "target_agent_id",
"tenant_id", "workflow_id",
):
if not isinstance(event[field], str) or not event[field]:
raise ValueError(f"{field} is required")
parents = event["parent_event_ids"]
if (
not isinstance(parents, list)
or len(parents) != len(set(parents))
or any(not isinstance(item, str) or not item for item in parents)
):
raise ValueError("parent_event_ids must be unique non-empty strings")
if event["edge_type"] in {"origin", "boundary_stub"} and parents:
raise ValueError("origin or boundary stub cannot claim a parent")
if event["edge_type"] not in {"origin", "boundary_stub"} and not parents:
raise ValueError("propagation event lacks a parent")
events.append(event)
for claim_id in {event["canonical_claim_id"] for event in events}:
claim_digests = {
event["raw_claim_sha256"] for event in events
if event["canonical_claim_id"] == claim_id
}
if len(claim_digests) != 1:
raise ValueError(
"this exact-lineage adapter cannot merge rephrased claim bytes"
)
claim_by_raw_digest: dict[str, str] = {}
for event in events:
raw_digest = event["raw_claim_sha256"]
claim_id = event["canonical_claim_id"]
prior_claim = claim_by_raw_digest.setdefault(raw_digest, claim_id)
if prior_claim != claim_id:
raise ValueError(
"identical raw claim bytes cannot be split across canonical IDs"
)
return events
def load_raw_receipts(
path: Path, events: list[dict], policy: dict
) -> dict[str, dict]:
lines = path.read_text(encoding="utf-8").splitlines()
if any(not line.strip() for line in lines):
raise ValueError("raw-claim receipt export contains blank records")
receipts, receipt_ids, source_event_ids = {}, set(), set()
for line in lines:
receipt = strict_json(line)
digest = hashlib.sha256(line.encode("utf-8")).hexdigest()
if not isinstance(receipt, dict) or set(receipt) != RAW_RECEIPT_FIELDS:
raise ValueError("raw-claim receipt schema differs")
if receipt["schema_version"] != "aidefend.raw-claim-digest-receipt.v2":
raise ValueError("raw-claim receipt version differs")
for field in (
"receipt_id", "source_id", "source_event_id",
"tenant_id", "workflow_id",
):
if not isinstance(receipt[field], str) or not receipt[field]:
raise ValueError(f"raw-claim receipt {field} is required")
if (
receipt["receipt_id"] in receipt_ids
or receipt["source_event_id"] in source_event_ids
or digest in receipts
):
raise ValueError("raw-claim receipt identity is duplicated")
receipt_ids.add(receipt["receipt_id"])
source_event_ids.add(receipt["source_event_id"])
receipt["_time"] = timestamp(receipt["captured_at"])
if (
receipt["source_id"] not in policy["required_sources"]
or receipt["digest_algorithm"] != "SHA-256"
or receipt["digest_stage"] != "pre_normalization_utf8_bytes"
or receipt["canonicalizer_version"] != policy["canonicalizer_version"]
or receipt["canonicalizer_sha256"] != policy["canonicalizer_sha256"]
or not isinstance(receipt["raw_claim_sha256"], str)
or not SHA256.fullmatch(receipt["raw_claim_sha256"])
):
raise ValueError("raw-claim digest receipt is invalid")
receipts[digest] = receipt
referenced = {event["raw_claim_receipt_sha256"] for event in events}
if referenced != set(receipts):
raise ValueError("raw-claim receipt and event populations differ")
skew = float(policy["max_clock_skew_seconds"])
for event in events:
receipt = receipts[event["raw_claim_receipt_sha256"]]
if (
receipt["source_id"] != event["source_id"]
or receipt["source_event_id"] != event["source_event_id"]
or receipt["tenant_id"] != event["tenant_id"]
or receipt["workflow_id"] != event["workflow_id"]
or receipt["raw_claim_sha256"] != event["raw_claim_sha256"]
or receipt["canonicalizer_version"] != event["canonicalizer_version"]
or receipt["canonicalizer_sha256"] != event["canonicalizer_sha256"]
or (receipt["_time"] - event["_time"]).total_seconds() > skew
):
raise ValueError("raw-claim receipt is not bound to its event")
return receipts
def load_states(path: Path, policy: dict, policy_sha256: str) -> list[dict]:
lines = path.read_text(encoding="utf-8").splitlines()
if any(not line.strip() for line in lines):
raise ValueError("claim-state export contains blank records")
states, state_ids, positions = [], set(), set()
for line in lines:
state = strict_json(line)
if not isinstance(state, dict) or set(state) != STATE_FIELDS:
raise ValueError("claim-state record schema differs")
if state["schema_version"] != "aidefend.claim-state.v1":
raise ValueError("claim-state record version differs")
state_id = state["state_id"]
if not isinstance(state_id, str) or not state_id or state_id in state_ids:
raise ValueError("claim-state ID is missing or duplicated")
state_ids.add(state_id)
for field in ("state_source_id", "canonical_claim_id", "source_control"):
if not isinstance(state[field], str) or not state[field]:
raise ValueError(f"claim-state {field} is required")
state["state_sequence"] = nonnegative_int(
state["state_sequence"], "state_sequence"
)
position = (state["state_source_id"], state["state_sequence"])
if position in positions:
raise ValueError("claim-state source sequence is duplicated")
positions.add(position)
state["_time"] = timestamp(state["evaluated_at"])
if (
state["decision"] not in CLAIM_STATES
or state["source_control"]
not in policy["authorized_state_controls"][state["decision"]]
or not isinstance(state["raw_claim_sha256"], str)
or not SHA256.fullmatch(state["raw_claim_sha256"])
or not isinstance(state["evidence_population_sha256"], str)
or not SHA256.fullmatch(state["evidence_population_sha256"])
or state["policy_sha256"] != policy_sha256
):
raise ValueError("claim-state authority or evidence binding is invalid")
states.append(state)
return states
def latest_states(states: list[dict]) -> dict[str, dict]:
grouped: dict[str, list[dict]] = defaultdict(list)
for state in states:
grouped[state["canonical_claim_id"]].append(state)
result = {}
for claim_id, records in grouped.items():
sources = {record["state_source_id"] for record in records}
if len(sources) != 1:
raise ValueError("claim state has multiple incomparable authorities")
ordered = sorted(records, key=lambda item: item["state_sequence"])
sequences = [item["state_sequence"] for item in ordered]
if len(sequences) != len(set(sequences)):
raise ValueError("claim-state sequence is duplicated")
result[claim_id] = ordered[-1]
return result
def load_manifest(
path: Path,
policy: dict,
events_path: Path,
states_path: Path,
raw_receipts_path: Path,
events: list[dict],
states: list[dict],
raw_receipts: dict[str, dict],
previous_result: dict | None,
previous_result_sha256: str | None,
) -> tuple[dict, dict[str, dict]]:
manifest = strict_json(path.read_text(encoding="utf-8"))
if not isinstance(manifest, dict) or set(manifest) != MANIFEST_FIELDS:
raise ValueError("lineage population manifest schema differs")
if (
manifest["schema_version"] != "aidefend.claim-lineage-population.v2"
or not isinstance(manifest["window_id"], str)
or not manifest["window_id"]
or not isinstance(manifest["tenant_id"], str)
or not manifest["tenant_id"]
or not isinstance(manifest["workflow_id"], str)
or not manifest["workflow_id"]
or manifest["collection_complete"] is not True
or manifest["required_sources"] != policy["required_sources"]
or manifest["canonicalizer_version"] != policy["canonicalizer_version"]
or manifest["canonicalizer_sha256"] != policy["canonicalizer_sha256"]
or manifest["events_sha256"]
!= hashlib.sha256(events_path.read_bytes()).hexdigest()
or manifest["states_sha256"]
!= hashlib.sha256(states_path.read_bytes()).hexdigest()
or manifest["raw_receipts_sha256"]
!= hashlib.sha256(raw_receipts_path.read_bytes()).hexdigest()
or manifest["event_count"] != len(events)
or manifest["state_count"] != len(states)
or manifest["raw_receipt_count"] != len(raw_receipts)
):
raise ValueError("lineage population is incomplete or misbound")
for field in ("event_count", "state_count", "raw_receipt_count"):
nonnegative_int(manifest[field], field)
manifest_sequence = nonnegative_int(
manifest["manifest_sequence"], "manifest_sequence"
)
if manifest_sequence < policy["minimum_manifest_sequence"]:
raise ValueError("lineage manifest is below the anti-rollback floor")
previous = manifest["previous_result_sha256"]
if previous is not None and (
not isinstance(previous, str) or not SHA256.fullmatch(previous)
):
raise ValueError("previous result binding is invalid")
if previous_result is None:
if previous is not None:
raise ValueError("genesis manifest unexpectedly names a prior result")
elif (
previous != previous_result_sha256
or manifest_sequence <= previous_result["manifest_sequence"]
):
raise ValueError(
"manifest does not strictly advance the trusted last-accepted result"
)
eligible = manifest["eligible_source_event_ids"]
if (
not isinstance(eligible, list)
or len(eligible) != len(set(eligible))
or any(not isinstance(value, str) or not value for value in eligible)
):
raise ValueError("eligible event population is invalid")
eligible_digest = hashlib.sha256(
json.dumps(sorted(eligible), separators=(",", ":")).encode("utf-8")
).hexdigest()
observed_ids = sorted(event["source_event_id"] for event in events)
if (
eligible_digest != manifest["eligible_source_event_ids_sha256"]
or observed_ids != sorted(eligible)
or len(eligible) != manifest["event_count"]
):
raise ValueError("eligible and observed lineage populations differ")
start = timestamp(manifest["window_start"])
end = timestamp(manifest["window_end"])
issued = timestamp(manifest["issued_at"])
expires = timestamp(manifest["expires_at"])
now = datetime.now(timezone.utc)
maximum_age = float(policy["maximum_evidence_age_seconds"])
skew = float(policy["max_clock_skew_seconds"])
if (
start >= end
or issued > now
or not issued < expires
or now >= expires
or (now - issued).total_seconds() > maximum_age
or (end - issued).total_seconds() > skew
):
raise ValueError("lineage population window or freshness is invalid")
for event in events:
if (
event["tenant_id"] != manifest["tenant_id"]
or event["workflow_id"] != manifest["workflow_id"]
or event["_time"] >= end
or (event["edge_type"] != "boundary_stub" and event["_time"] < start)
):
raise ValueError("lineage event is outside manifest scope")
checkpoints = manifest["source_checkpoints"]
if not isinstance(checkpoints, dict) or set(checkpoints) != set(policy["required_sources"]):
raise ValueError("source checkpoint registry differs")
for source, checkpoint in checkpoints.items():
if (
not isinstance(checkpoint, dict)
or set(checkpoint) != CHECKPOINT_FIELDS
or checkpoint["complete"] is not True
):
raise ValueError("source checkpoint is incomplete or malformed")
count = nonnegative_int(checkpoint["count"], "checkpoint.count")
sequences = sorted(
event["source_sequence"] for event in events
if event["source_id"] == source
)
if count != len(sequences):
raise ValueError("source checkpoint count differs")
if sequences:
first = nonnegative_int(checkpoint["first"], "checkpoint.first")
last = nonnegative_int(checkpoint["last"], "checkpoint.last")
if sequences != list(range(first, last + 1)):
raise ValueError("source sequence population has a gap or rollback")
elif checkpoint["first"] is not None or checkpoint["last"] is not None:
raise ValueError("empty source checkpoint has sequence bounds")
current_states = latest_states(states)
event_claims = {event["canonical_claim_id"] for event in events}
if set(current_states) != event_claims:
raise ValueError("claim-state and event claim populations differ")
digests_by_claim = defaultdict(set)
for event in events:
digests_by_claim[event["canonical_claim_id"]].add(event["raw_claim_sha256"])
for claim_id, state in current_states.items():
if (
state["raw_claim_sha256"] not in digests_by_claim[claim_id]
or (state["_time"] - issued).total_seconds() > skew
or (issued - state["_time"]).total_seconds() > maximum_age
):
raise ValueError("claim-state raw digest is absent from lineage")
return manifest, current_states
def evaluate(
events: list[dict],
states: dict[str, dict],
policy: dict,
manifest: dict,
input_digests: dict[str, str | None],
) -> dict:
by_id = {event["event_id"]: event for event in events}
children: dict[str, list[str]] = defaultdict(list)
neighbours: dict[str, set[str]] = defaultdict(set)
skew = float(policy["max_clock_skew_seconds"])
for event in events:
for parent_id in event["parent_event_ids"]:
parent = by_id.get(parent_id)
if parent is None:
raise ValueError("lineage export is missing a parent event")
if parent["canonical_claim_id"] != event["canonical_claim_id"]:
raise ValueError("lineage edge crosses canonical claims")
if parent["workflow_id"] != event["workflow_id"]:
raise ValueError("lineage edge crosses workflows")
if (
parent["source_id"] == event["source_id"]
and parent["source_sequence"] >= event["source_sequence"]
):
raise ValueError("same-source lineage edge reverses sequence")
if (parent["_time"] - event["_time"]).total_seconds() > skew:
raise ValueError("lineage edge exceeds allowed clock skew")
children[parent_id].append(event["event_id"])
neighbours[parent_id].add(event["event_id"])
neighbours[event["event_id"]].add(parent_id)
indegree = {
event["event_id"]: len(event["parent_event_ids"])
for event in events
}
ready = sorted(event_id for event_id, count in indegree.items() if count == 0)
topological = []
while ready:
event_id = ready.pop(0)
topological.append(event_id)
for child_id in sorted(children.get(event_id, [])):
indegree[child_id] -= 1
if indegree[child_id] == 0:
ready.append(child_id)
ready.sort()
if len(topological) != len(events):
raise ValueError("claim lineage contains a cycle")
order = {event_id: index for index, event_id in enumerate(topological)}
findings = []
limits = policy["limits"]
visited = set()
for start_id in topological:
if start_id in visited:
continue
claim_id = by_id[start_id]["canonical_claim_id"]
component, pending = set(), [start_id]
while pending:
event_id = pending.pop()
if event_id in component:
continue
if by_id[event_id]["canonical_claim_id"] != claim_id:
raise ValueError("weak component crosses canonical claims")
component.add(event_id)
pending.extend(sorted(neighbours.get(event_id, set()) - component))
visited.update(component)
state = states[claim_id]
if state["decision"] not in BAD_STATES:
continue
component_order = sorted(component, key=lambda event_id: order[event_id])
depths = {}
for event_id in component_order:
parents = [
parent_id for parent_id in by_id[event_id]["parent_event_ids"]
if parent_id in component
]
depths[event_id] = (
max(depths[parent_id] for parent_id in parents) + 1
if parents else 0
)
observed = [by_id[event_id] for event_id in component_order]
agents = {
agent_id for event in observed
for agent_id in (event["source_agent_id"], event["target_agent_id"])
}
max_fan_out = max(
(sum(child_id in component for child_id in children.get(event["event_id"], []))
for event in observed),
default=0,
)
memory_writes = sum(event["edge_type"] == "memory_write" for event in observed)
cross_tenant = sum(
by_id[parent_id]["tenant_id"] != event["tenant_id"]
for event in observed for parent_id in event["parent_event_ids"]
if parent_id in component
)
seen_agents, agent_reentries = set(), 0
for event in observed:
if (
event["edge_type"] not in {"origin", "boundary_stub"}
and event["target_agent_id"] in seen_agents
):
agent_reentries += 1
seen_agents.update((event["source_agent_id"], event["target_agent_id"]))
metrics = {
"distinct_agents": len(agents),
"depth": max(depths.values(), default=0),
"fan_out": max_fan_out,
"memory_writes": memory_writes,
"agent_reentries": agent_reentries,
"cross_tenant_edges": cross_tenant,
"lineage_events": len(observed),
}
bounded_metrics = {
"distinct_agents": "max_distinct_agents",
"depth": "max_depth",
"fan_out": "max_fan_out",
"memory_writes": "max_memory_writes",
"agent_reentries": "max_agent_reentries",
}
breaches = [
name for name, limit_name in bounded_metrics.items()
if metrics[name] > limits[limit_name]
]
if policy["forbid_cross_tenant"] and cross_tenant:
breaches.append("cross_tenant_edges")
if breaches:
findings.append({
"canonical_claim_id": claim_id,
"component_sha256": hashlib.sha256(
json.dumps(sorted(component), separators=(",", ":")).encode("utf-8")
).hexdigest(),
"root_event_ids": sorted(
event_id for event_id in component
if not any(
parent_id in component
for parent_id in by_id[event_id]["parent_event_ids"]
)
),
"state_id": state["state_id"],
"state_decision": state["decision"],
"state_source_control": state["source_control"],
"metrics": metrics,
"breached_limits": sorted(breaches),
})
return {
"schema_version": "aidefend.claim-cascade-result.v2",
"policy_version": policy["policy_version"],
"canonicalizer_version": policy["canonicalizer_version"],
"canonicalizer_sha256": policy["canonicalizer_sha256"],
"detector_build_sha256": policy["detector_build_sha256"],
"manifest_sequence": manifest["manifest_sequence"],
"window_id": manifest["window_id"],
"manifest_issued_at": manifest["issued_at"],
"manifest_expires_at": manifest["expires_at"],
"previous_result_sha256": manifest["previous_result_sha256"],
"input_sha256": input_digests,
"event_count": len(events),
"state_record_count": manifest["state_count"],
"raw_claim_receipt_sha256": sorted(
event["raw_claim_receipt_sha256"] for event in events
),
"claim_count": len(states),
"finding_status": "FINDING" if findings else "NO_FINDING",
"findings": findings,
}
def main(argv: list[str]) -> int:
if len(argv) != 9:
raise ValueError(
"usage: claim_cascade.py POLICY CANONICALIZER EVENTS STATES "
"RAW_RECEIPTS MANIFEST PREVIOUS_RESULT_OR_DASH OUTPUT"
)
(
policy_path, canonicalizer_path, events_path, states_path,
raw_receipts_path, manifest_path, previous_result_path, output_path,
) = map(
Path, argv[1:]
)
raw_digests = {
"policy": hashlib.sha256(policy_path.read_bytes()).hexdigest(),
"canonicalizer": hashlib.sha256(canonicalizer_path.read_bytes()).hexdigest(),
"events": hashlib.sha256(events_path.read_bytes()).hexdigest(),
"states": hashlib.sha256(states_path.read_bytes()).hexdigest(),
"raw_receipts": hashlib.sha256(raw_receipts_path.read_bytes()).hexdigest(),
"manifest": hashlib.sha256(manifest_path.read_bytes()).hexdigest(),
}
policy = load_policy(policy_path, canonicalizer_path)
previous_result, previous_result_sha256 = load_previous_result(
previous_result_path
)
raw_digests["previous_result"] = previous_result_sha256
events = load_events(events_path, policy)
raw_receipts = load_raw_receipts(raw_receipts_path, events, policy)
state_records = load_states(states_path, policy, raw_digests["policy"])
manifest, states = load_manifest(
manifest_path, policy, events_path, states_path, raw_receipts_path,
events, state_records, raw_receipts,
previous_result, previous_result_sha256,
)
result = evaluate(events, states, policy, manifest, raw_digests)
output_path.parent.mkdir(parents=True, exist_ok=True)
writing = output_path.with_name(output_path.name + ".writing")
writing.unlink(missing_ok=True)
writing.write_text(
json.dumps(result, sort_keys=True, separators=(",", ":"), allow_nan=False)
+ "\n",
encoding="utf-8",
)
writing.replace(output_path)
return 1 if result["finding_status"] == "FINDING" else 0
if __name__ == "__main__":
destination = Path(sys.argv[-1]) if len(sys.argv) == 9 else None
try:
raise SystemExit(main(sys.argv))
except SystemExit:
raise
except Exception as error:
if destination is not None:
destination.unlink(missing_ok=True)
destination.with_name(destination.name + ".writing").unlink(missing_ok=True)
print(f"cascade evaluation error: {type(error).__name__}: {error}", file=sys.stderr)
raise SystemExit(2)Verify cascade semantics
Run signed fixtures for a local bad claim below every limit; late CONTRADICTED state after A-to-B-to-C propagation; two independent roots; incomplete or digest-drifted population; same event ID with changed bytes; forged or stale state authority; signed-manifest rollback; a changed raw digest under the same canonical ID that this exact-lineage adapter must reject; per-source sequence reuse or gap; invalid origin/parent structure; a pre-window parent represented by a signed boundary stub; agent re-entry; peer-majority propagation with an external UNVERIFIED state; and a signed authoritative empty population. A valid finding exits 1 and publishes one signed generation archive; input or graph error exits 2 and leaves no final archive. Independently unpack the archive, verify the internal result signature, replay the exact raw inputs, and compare the result digest.
Open-source tools
- LangChain
- Neo4j Community Edition
- jsonschema
- OpenTelemetry
- Langfuse (trace substrate for claim-occurrence graph analysis; claim-state binding and cascade detection remain external)
Commercial / hosted tools
- Neo4j Enterprise Edition
- Databricks MLflow 3
- Datadog LLM Observability
Threat mappings
MITRE ATLAS
- AML.T0062 Discover LLM Hallucinations (cascade findings identify unverified hallucinations propagating across agents)
- AML.T0080 AI Agent Context Poisoning (cascade findings identify poisoned context persisting across agent boundaries)
- AML.T0080.000 AI Agent Context Poisoning: Memory (cascade findings identify hallucinated facts propagating through agent memory)
MAESTRO
- Goal Misalignment Cascades (Cross-Layer) (cascade findings identify false statements propagating across agents and being reinforced as 'truth')
OWASP LLM Top 10 2026
- LLM07:2026 Misinformation
OWASP ML Top 10 2023
- ML09:2023 Output Integrity Attack (cascade findings identify fabricated claims treated as authoritative across downstream systems)
OWASP Top 10 for Agentic Applications 2026
- ASI08:2026 Cascading Failures (hallucination cascades across agent networks)
- ASI01:2026 Agent Goal Hijack (poisoned facts redirect agent decision pathways)
NIST Adversarial Machine Learning 2025
- NISTAML.027 Misaligned Outputs
- NISTAML.015 Indirect Prompt Injection
Cisco Integrated AI Security and Safety Framework
- AISubtech-15.1.19 Integrity: Hallucinations / Misinformation (official misinformation crosswalk supports tracking unverified factual-claim states across agent messages and memory)
- AISubtech-15.1.5 Safety Harms and Toxicity: Disinformation (authenticated false-claim propagation across agents or memory writes)
- AISubtech-5.1.1 Long-term / Short-term Memory Injection (authenticated false-claim propagation across agents or memory writes)
- AITech-15.1 Harmful Content (authenticated false-claim propagation across agents or memory writes)
- AITech-5.1 Memory System Persistence (authenticated false-claim propagation across agents or memory writes)
- AITech-7.2 Memory System Corruption (authenticated false-claim propagation across agents or memory writes)
Google Secure AI Framework 2.0 - Risks
- IMO: Insecure Model Output (cascade monitoring identifies unsupported claims amplified across agents)
Databricks AI Security Framework 3.0
- Agents - Core 13.5: Cascading Hallucination Attacks
- Agents - Core 13.12: Agent Communication Poisoning
- Agents - Core 13.6: Intent Breaking & Goal Manipulation (poisoned facts redirect agent intent)