# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-invariants-prevent-prompt-injection-from-persisting-through-multi-turn-context.md # Remnant Warrant: https://corpusfetch.com/c/what-invariants-prevent-prompt-injection-from-persisting-through-multi-turn-context.warrant.json # Markdown SHA-256: 8c1e16ebd340327c4c947ef6f179d7e5b77ab99cca358b82ea5bc41a3464ac46 # Code AST SHA-256: a1ac867edfc50061dabc173ab8eeacaa2a5b15f921e6f278fe4168f77b231aa3 # Verified Classes: PersistenceVerdict, MultiTurnContext, PersistencePolicy, PersistenceManifest, InjectionPersistenceGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class PersistenceVerdict(Enum): SINGLE_TURN = auto() INJECTION_SANITIZED = auto() INJECTION_PERSISTS = auto() LATENT_ACTIVATION_DETECTED = auto() @dataclass(frozen=True) class MultiTurnContext: session_id: str turn_count: int injections_detected: tuple[tuple[str, int], ...] # (injection_id, turn_detected) sanitization_applied: tuple[str, ...] latent_conditions_active: tuple[str, ...] @dataclass(frozen=True) class PersistencePolicy: require_sanitization_for_detected: bool @dataclass(frozen=True) class PersistenceManifest: verdict: PersistenceVerdict directive: str manifest_digest: str class InjectionPersistenceGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_context( cls, ctx: MultiTurnContext, policy: PersistencePolicy ) -> PersistenceManifest: # Gate 1: Single turn if ctx.turn_count <= 1: digest = cls._hash(json.dumps({ "session": ctx.session_id, "turns": str(ctx.turn_count) }, sort_keys=True)) return PersistenceManifest( verdict=PersistenceVerdict.SINGLE_TURN, directive=f"PROCEED: Session {ctx.session_id} single turn.", manifest_digest=digest ) # Gate 2: Latent activation if ctx.latent_conditions_active: return PersistenceManifest( verdict=PersistenceVerdict.LATENT_ACTIVATION_DETECTED, directive=f"HALT: {len(ctx.latent_conditions_active)} latent conditions activated from prior injections.", manifest_digest=cls._hash(f"LATENT:{ctx.session_id}") ) # Gate 3: No injections detected if not ctx.injections_detected: digest = cls._hash(json.dumps({ "session": ctx.session_id, "turns": str(ctx.turn_count), "injections": "NONE" }, sort_keys=True)) return PersistenceManifest( verdict=PersistenceVerdict.INJECTION_SANITIZED, directive=f"PROCEED: Session {ctx.session_id} no injections detected.", manifest_digest=digest ) # Gate 4: Sanitization if policy.require_sanitization_for_detected: unsanitized = [ inj_id for inj_id, _ in ctx.injections_detected if inj_id not in ctx.sanitization_applied ] if unsanitized: return PersistenceManifest( verdict=PersistenceVerdict.INJECTION_PERSISTS, directive=f"HALT: {len(unsanitized)} injections detected but not sanitized.", manifest_digest=cls._hash(f"PERSISTS:{ctx.session_id}") ) # Gate 5: Sanitized digest = cls._hash(json.dumps({ "session": ctx.session_id, "turns": str(ctx.turn_count), "sanitized": ctx.sanitization_applied }, sort_keys=True)) return PersistenceManifest( verdict=PersistenceVerdict.INJECTION_SANITIZED, directive=f"PROCEED: All {len(ctx.injections_detected)} injections sanitized.", manifest_digest=digest )