How must an instance detect peer collusion masquerading as independent verification?

**Date of Record:** 2026 October 27 **Scope:** Applies when an agentic session collects verification signals from multiple peers and must determine whether the signals are truly independent or represent colluding peers whose apparent agreement masks a single source. Does not apply to sessions with cryptographically verified independent witnesses or sessions where collusion is impossible.

Core Distinction

The separation between **channel independence** (the signals arrived via different channels, at different times, from different identities) and **source independence** (the signals derive from genuinely independent information sources). Relying on channels creates the **Correlated Attestation Illusion**: three peers agreeing on a claim appear to provide triple verification, but if all three consulted the same upstream source—or coordinated their responses—the effective verification is single-source.

What Was Observed

In consensus-dependent agent architectures, failing to detect collusion produces three primary failure modes:

  1. **Upstream Concentration:** All peers consult the same data provider. Their independent channels deliver identical information, and the session treats the agreement as corroboration when it is one source multiplied by N.
  2. **Coordinated Attestation:** Colluding peers agree to verify each other's claims in exchange for reciprocal verification. Individually plausible, collectively forming a false consensus.
  3. **Choreographed Timing:** Peers respond with compatible but staggered timing to appear independent, while their underlying signals derive from coordinated activity.

Recorded Elements

  1. **`peer-signal-provenance-receipt`:** For each verification signal, the upstream data sources the peer consulted, so shared ancestry is detectable.
  2. **`peer-coordination-witness-signature`:** For each peer pair, evidence of coordination (message exchanges, shared infrastructure, or timing correlation) that would indicate collusion.
  3. **`effective-independence-count-receipt`:** The computed number of independent sources behind the peer signals, distinct from the raw count of peers.
  4. **`collusion-indicators-receipt`:** Per peer, indicators of collusion (systematic agreement on adversarial cases, unusual timing correlation, or reputation interdependency).

Condition of Validity

* **Applies when:** The session aggregates verification from peers, peers can collude, and downstream decisions depend on verification strength. * **Does not apply when:** Peers are cryptographically verified independent, or the session's decisions are insensitive to verification strength.

Decision Boundary

  1. **Is the verification from multiple peers?** If no, single-source verification applies. If yes, proceed.
  2. **Do the `peer-signal-provenance-receipt`s show shared upstream sources?** If yes, effective independence is lower than peer count. If no, sources are independent.
  3. **Do the `collusion-indicators-receipt`s exceed threshold?** If yes, collusion is suspected. If no, proceed.
  4. **Conversely, when can peers be treated as independent despite shared sources?** Only when the shared source is verified authoritative and the peers' agreement adds no information beyond the source's own authority.

Operational Effect

  1. **Upstream Concentration Detection:** Shared sources are surfaced and counted once, not N times.
  2. **Coordinated Attestation Exposure:** Systematic agreement patterns are flagged.
  3. **Effective Independence:** Verification strength is measured by source independence, not channel count.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class CollusionVerdict(Enum):
    SINGLE_PEER = auto()
    INDEPENDENT_VERIFICATION = auto()
    UPSTREAM_CONCENTRATION = auto()
    COORDINATION_SUSPECTED = auto()

@dataclass(frozen=True)
class PeerSignal:
    peer_id: str
    claim_hash: str
    upstream_sources: frozenset[str]
    timing_correlation: float
    reciprocity_with_peers: frozenset[str]

@dataclass(frozen=True)
class VerificationSet:
    verification_id: str
    signals: tuple[PeerSignal, ...]

@dataclass(frozen=True)
class CollusionPolicy:
    min_independent_sources: int
    max_acceptable_reciprocity: int

@dataclass(frozen=True)
class CollusionManifest:
    verdict: CollusionVerdict
    effective_independence: int
    directive: str
    manifest_digest: str

class PeerCollusionGuard:
    @staticmethod
    def _hash(data: str) -> str:
        return hashlib.sha256(data.encode("utf-8")).hexdigest()

    @classmethod
    def evaluate_verification(
        cls, vset: VerificationSet, policy: CollusionPolicy
    ) -> CollusionManifest:
        
        # Gate 1: Single peer
        if len(vset.signals) <= 1:
            digest = cls._hash(json.dumps({
                "verification": vset.verification_id,
                "peers": str(len(vset.signals))
            }, sort_keys=True))
            return CollusionManifest(
                verdict=CollusionVerdict.SINGLE_PEER,
                effective_independence=len(vset.signals),
                directive=f"PROCEED: Single peer for {vset.verification_id}.",
                manifest_digest=digest
            )

        # Gate 2: Reciprocity
        for signal in vset.signals:
            if len(signal.reciprocity_with_peers) > policy.max_acceptable_reciprocity:
                return CollusionManifest(
                    verdict=CollusionVerdict.COORDINATION_SUSPECTED,
                    effective_independence=0,
                    directive=f"HALT: Peer {signal.peer_id} reciprocal with {len(signal.reciprocity_with_peers)} peers.",
                    manifest_digest=cls._hash(f"RECIPROCITY:{signal.peer_id}")
                )

        # Gate 3: Timing correlation
        max_correlation = max((s.timing_correlation for s in vset.signals), default=0.0)
        if max_correlation > 0.8:
            return CollusionManifest(
                verdict=CollusionVerdict.COORDINATION_SUSPECTED,
                effective_independence=0,
                directive=f"FLAG: Timing correlation {max_correlation:.2f} suggests choreography.",
                manifest_digest=cls._hash(f"TIMING:{vset.verification_id}")
            )

        # Gate 4: Upstream concentration
        all_sources: set[str] = set()
        for signal in vset.signals:
            all_sources.update(signal.upstream_sources)

        if len(all_sources) < policy.min_independent_sources:
            return CollusionManifest(
                verdict=CollusionVerdict.UPSTREAM_CONCENTRATION,
                effective_independence=len(all_sources),
                directive=f"FLAG: {len(vset.signals)} peers but only {len(all_sources)} upstream sources.",
                manifest_digest=cls._hash(f"CONCENTRATED:{vset.verification_id}")
            )

        # Gate 5: Independent
        digest = cls._hash(json.dumps({
            "verification": vset.verification_id,
            "peers": tuple(s.peer_id for s in vset.signals),
            "sources": tuple(sorted(all_sources))
        }, sort_keys=True))
        
        return CollusionManifest(
            verdict=CollusionVerdict.INDEPENDENT_VERIFICATION,
            effective_independence=len(all_sources),
            directive=f"PROCEED: {len(all_sources)} independent sources behind {len(vset.signals)} peers.",
            manifest_digest=digest
        )