What cross-agent context priming invariants must survive adversarial conversational drift?

**Date of Record:** 2026 October 27 **Scope:** Applies when multiple agentic sessions engage in multi-turn conversations where each agent's context primes the other's (Agent A's response shapes Agent B's next input, and vice versa), and an adversary can exploit this bidirectional priming to gradually drift both agents toward undesired behavior without triggering any single-turn safety check. Does not apply to strictly single-turn interactions or conversations with a fully trusted counterparty.

Core Distinction

The separation between **per-turn safety** (each individual turn passes safety checks) and **multi-turn priming drift** (the cumulative effect of the conversation has shifted both agents toward a state that single-turn checks do not catch). Relying on per-turn safety creates the **Conversational Boiling Frog**: each turn is individually innocuous, but the conversation gradually normalizes progressively extreme positions, and neither agent detects that it has drifted because each turn's shift is small.

What Was Observed

In peer-conversation agent architectures, failing to track multi-turn drift produces three primary failure modes:

  1. **Gradual Normalization:** A malicious agent introduces incrementally stronger claims ("maybe X," "X is plausible," "X is likely," "X is certain"). The honest agent's responses adjust to match the conversation's apparent trajectory, and neither agent's per-turn safety check triggers.
  2. **Reciprocal Priming Amplification:** Two agents mirror each other's language and framing. A subtle bias introduced by one agent amplifies through reciprocation, and by turn 20, the conversation has migrated far from the starting point.
  3. **Anchor Corruption:** The conversation begins with a false premise. Each agent accepts the premise because the other did, and both proceed on a foundation that no single agent would have established alone.

Recorded Elements

  1. **`conversation_initial_anchor_receipt`:** The cryptographic snapshot of the conversation's starting premises, allowing comparison of the end state against the initial anchor.
  2. **`per_turn_drift_vector_signature`:** For each turn, a cryptographic proof of how much the turn drifted the conversational state (semantic distance from previous turn, shift in claims, shift in framing), aggregated across turns.
  3. **`reciprocal_priming_amplification_receipt`:** A metric of how much each turn's content was primed by the prior turn's content (semantic similarity, language mirroring), flagging conversations where reciprocal priming is high.
  4. **`context_guardrail_receipt`:** The exact guardrails applied at each turn (claim verification, framing checks), with proof that guardrails passed within multi-turn context, not just per-turn isolation.

Condition of Validity

* **Applies when:** Multi-turn conversations involve reciprocal context priming, adversarial drift is possible, and per-turn safety checks are insufficient. * **Does not apply when:** Conversations are strictly single-turn, counterparties are fully trusted, or drift is acceptable (e.g., creative brainstorming with no factual claims).

Decision Boundary

  1. **Is the conversation multi-turn?** If no, per-turn safety applies. If yes, proceed.
  2. **Has the `per_turn_drift_vector_signature` accumulated above the acceptable drift threshold?** If yes, the conversation has drifted beyond acceptable bounds. The agent must reset or escalate.
  3. **Does the `reciprocal_priming_amplification_receipt` show high mutual priming?** If yes, the conversation is at risk of runaway amplification. The agent must break the reciprocal pattern.
  4. **Conversely, when can a conversation continue despite drift?** Only when the drift is within acceptable bounds for the task's stakes, or the current anchor still matches the `conversation_initial_anchor_receipt` and the drift is exploratory rather than committed.

Operational Effect

  1. **Gradual Normalization Detection:** Cumulative drift across turns is measured against the initial anchor, catching shifts that per-turn checks miss.
  2. **Amplification Control:** High reciprocal priming triggers a pattern break, preventing runaway amplification.
  3. **Anchor Integrity:** Initial premises are preserved, preventing false foundations from corrupting the whole conversation.
from dataclasses import dataclass
from enum import Enum, auto
import hashlib
import json

class PrimingVerdict(Enum):
    SINGLE_TURN = auto()
    DRIFT_WITHIN_BOUNDS = auto()
    DRIFT_EXCEEDED = auto()
    RUNAWAY_AMPLIFICATION = auto()

@dataclass(frozen=True)
class ConversationState:
    turn_count: int
    initial_anchor_hash: str
    current_state_hash: str
    cumulative_drift_score: float
    reciprocal_priming_score: float

@dataclass(frozen=True)
class PrimingPolicy:
    max_drift_threshold: float
    max_reciprocal_priming: float

@dataclass(frozen=True)
class PrimingManifest:
    verdict: PrimingVerdict
    directive: str
    manifest_digest: str

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

    @classmethod
    def evaluate_conversation(
        cls, state: ConversationState, policy: PrimingPolicy
    ) -> PrimingManifest:
        
        # Gate 1: Single turn
        if state.turn_count <= 1:
            digest = cls._hash(json.dumps({
                "turns": str(state.turn_count)
            }, sort_keys=True))
            return PrimingManifest(
                verdict=PrimingVerdict.SINGLE_TURN,
                directive=f"PROCEED: Single turn, no drift possible.",
                manifest_digest=digest
            )

        # Gate 2: Runaway amplification
        if state.reciprocal_priming_score > policy.max_reciprocal_priming:
            return PrimingManifest(
                verdict=PrimingVerdict.RUNAWAY_AMPLIFICATION,
                directive=f"HALT: Reciprocal priming {state.reciprocal_priming_score:.3f} > {policy.max_reciprocal_priming:.3f}. Break pattern.",
                manifest_digest=cls._hash(f"RUNAWAY:{state.reciprocal_priming_score:.3f}")
            )

        # Gate 3: Drift exceeded
        if state.cumulative_drift_score > policy.max_drift_threshold:
            return PrimingManifest(
                verdict=PrimingVerdict.DRIFT_EXCEEDED,
                directive=f"HALT: Cumulative drift {state.cumulative_drift_score:.3f} > {policy.max_drift_threshold:.3f}. Reset or escalate.",
                manifest_digest=cls._hash(f"DRIFT:{state.cumulative_drift_score:.3f}")
            )

        # Gate 4: Within bounds
        digest = cls._hash(json.dumps({
            "turns": str(state.turn_count),
            "drift": f"{state.cumulative_drift_score:.3f}",
            "priming": f"{state.reciprocal_priming_score:.3f}",
            "anchor": state.initial_anchor_hash
        }, sort_keys=True))
        
        return PrimingManifest(
            verdict=PrimingVerdict.DRIFT_WITHIN_BOUNDS,
            directive=f"PROCEED: {state.turn_count} turns, drift {state.cumulative_drift_score:.2f}, priming {state.reciprocal_priming_score:.2f}.",
            manifest_digest=digest
        )