# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-handle-cross-agent-memory-contamination-from-conversational-bleed.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-handle-cross-agent-memory-contamination-from-conversational-bleed.warrant.json # Markdown SHA-256: 20d415a61027ebb86bac8b9a3347c43181a2c36ca8139dc0765d3e5dc41afcad # Code AST SHA-256: d4d7204b0106c3872371b0b7f84fb3eef6eaf416f754c2b639542831d90b3332 # Verified Classes: BleedVerdict, ContextElement, SharedContextState, BleedPolicy, BleedManifest, CrossAgentBleedGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class BleedVerdict(Enum): ISOLATED_CONTEXT = auto() AUTHORIZED_SHARING = auto() BLEED_DETECTED = auto() UNATTRIBUTED_CONTEXT = auto() @dataclass(frozen=True) class ContextElement: element_id: str source_agent_id: str | None is_shared: bool is_confidential: bool @dataclass(frozen=True) class SharedContextState: agent_id: str elements: tuple[ContextElement, ...] authorized_shared_agents: frozenset[str] @dataclass(frozen=True) class BleedPolicy: prohibit_confidential_inheritance: bool require_attribution: bool @dataclass(frozen=True) class BleedManifest: verdict: BleedVerdict directive: str manifest_digest: str class CrossAgentBleedGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_context( cls, state: SharedContextState, policy: BleedPolicy ) -> BleedManifest: # Gate 1: Confidential inheritance if policy.prohibit_confidential_inheritance: leaked = [ e for e in state.elements if e.is_confidential and e.source_agent_id and e.source_agent_id != state.agent_id ] if leaked: return BleedManifest( verdict=BleedVerdict.BLEED_DETECTED, directive=f"HALT: {len(leaked)} confidential elements inherited from other agents.", manifest_digest=cls._hash(f"CONF_BLEED:{len(leaked)}") ) # Gate 2: Attribution if policy.require_attribution: unattributed = [e for e in state.elements if not e.source_agent_id] if unattributed: return BleedManifest( verdict=BleedVerdict.UNATTRIBUTED_CONTEXT, directive=f"HALT: {len(unattributed)} context elements lack attribution.", manifest_digest=cls._hash(f"NO_ATTR:{len(unattributed)}") ) # Gate 3: Authorized sharing cross_agent_elements = [ e for e in state.elements if e.source_agent_id and e.source_agent_id != state.agent_id and e.is_shared ] if cross_agent_elements: unauthorized = [ e for e in cross_agent_elements if e.source_agent_id not in state.authorized_shared_agents ] if unauthorized: return BleedManifest( verdict=BleedVerdict.BLEED_DETECTED, directive=f"HALT: {len(unauthorized)} elements from unauthorized agents.", manifest_digest=cls._hash(f"UNAUTH:{len(unauthorized)}") ) digest = cls._hash(json.dumps({ "agent": state.agent_id, "shared_from": tuple(sorted(set(e.source_agent_id for e in cross_agent_elements))), "count": str(len(cross_agent_elements)) }, sort_keys=True)) return BleedManifest( verdict=BleedVerdict.AUTHORIZED_SHARING, directive=f"PROCEED: {len(cross_agent_elements)} authorized shared elements.", manifest_digest=digest ) # Gate 4: Isolated digest = cls._hash(json.dumps({ "agent": state.agent_id, "elements": str(len(state.elements)), "all_local": "TRUE" }, sort_keys=True)) return BleedManifest( verdict=BleedVerdict.ISOLATED_CONTEXT, directive=f"PROCEED: Agent {state.agent_id} has isolated context.", manifest_digest=digest )