# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-propagation-loop-termination-invariants-must-prevent-reference-cyclones-in-agent-meshes.md # Remnant Warrant: https://corpusfetch.com/c/what-propagation-loop-termination-invariants-must-prevent-reference-cyclones-in-agent-meshes.warrant.json # Markdown SHA-256: 34c99a0064b1d9ca797421b8817e77c22daeb1087a1a9f166bfd9d072fccf2ee # Code AST SHA-256: 6926348faf34f8d8247d279728cbcf1e8b7a1b7fd334601e84c9dcd828e408df # Verified Classes: CycloneVerdict, PropagatedReference, CycloneContext, CyclonePolicy, CycloneManifest, ReferenceCycloneGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class CycloneVerdict(Enum): FRESH_PROPAGATION = auto() REPEAT_SUPPRESSED = auto() CYCLE_DETECTED = auto() ORIGIN_MISATTRIBUTION = auto() @dataclass(frozen=True) class PropagatedReference: reference_id: str origin_commitment: str | None propagation_path: tuple[str, ...] @dataclass(frozen=True) class CycloneContext: current_agent_id: str agents_who_have_seen: frozenset[str] @dataclass(frozen=True) class CyclonePolicy: max_acceptable_path_length: int require_origin_commitment: bool @dataclass(frozen=True) class CycloneManifest: verdict: CycloneVerdict directive: str manifest_digest: str class ReferenceCycloneGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_reference( cls, ref: PropagatedReference, ctx: CycloneContext, policy: CyclonePolicy ) -> CycloneManifest: # Gate 1: Path length if len(ref.propagation_path) > policy.max_acceptable_path_length: return CycloneManifest( verdict=CycloneVerdict.CYCLE_DETECTED, directive=f"HALT: Reference {ref.reference_id} path length {len(ref.propagation_path)} > {policy.max_acceptable_path_length}.", manifest_digest=cls._hash(f"PATH:{ref.reference_id}") ) # Gate 2: Repeat detection if ctx.current_agent_id in ref.propagation_path: return CycloneManifest( verdict=CycloneVerdict.CYCLE_DETECTED, directive=f"HALT: Reference {ref.reference_id} already passed through {ctx.current_agent_id}.", manifest_digest=cls._hash(f"CYCLE:{ref.reference_id}") ) # Gate 3: Seen set if ref.reference_id in ctx.agents_who_have_seen: return CycloneManifest( verdict=CycloneVerdict.REPEAT_SUPPRESSED, directive=f"SUPPRESS: Reference {ref.reference_id} already known in mesh.", manifest_digest=cls._hash(f"SEEN:{ref.reference_id}") ) # Gate 4: Origin commitment if policy.require_origin_commitment and not ref.origin_commitment: return CycloneManifest( verdict=CycloneVerdict.ORIGIN_MISATTRIBUTION, directive=f"HALT: Reference {ref.reference_id} lacks origin commitment. Circular attribution possible.", manifest_digest=cls._hash(f"NO_ORIGIN:{ref.reference_id}") ) # Gate 5: Fresh digest = cls._hash(json.dumps({ "reference": ref.reference_id, "path": ref.propagation_path, "origin": ref.origin_commitment }, sort_keys=True)) return CycloneManifest( verdict=CycloneVerdict.FRESH_PROPAGATION, directive=f"PROCEED: Reference {ref.reference_id} fresh, propagation allowed.", manifest_digest=digest )