# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-preserve-causal-ordering-of-tool-side-effects-across-asynchronous-execution.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-preserve-causal-ordering-of-tool-side-effects-across-asynchronous-execution.warrant.json # Markdown SHA-256: 987feda391703b078d30d36ca60e2ff5a03799355a1c50f16a68f57518dab41a # Code AST SHA-256: 362b6b7111d11552741ecdceec36e0b78b995856075b637862d3bf218fbfd77d # Verified Classes: CausalVerdict, ToolInvocation, CausalState, CausalPolicy, CausalManifest, CausalSideEffectGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class CausalVerdict(Enum): NO_CAUSAL_DEPENDENCY = auto() CAUSAL_ORDER_VERIFIED = auto() CAUSAL_BARRIER_MISSING = auto() CAUSAL_REVERSAL_DETECTED = auto() @dataclass(frozen=True) class ToolInvocation: tool_id: str side_effect_sequence: int | None causal_barrier_receipt: str | None depends_on: tuple[str, ...] @dataclass(frozen=True) class CausalState: invocations: tuple[ToolInvocation, ...] external_side_effect_order: tuple[str, ...] | None @dataclass(frozen=True) class CausalPolicy: require_barriers: bool @dataclass(frozen=True) class CausalManifest: verdict: CausalVerdict directive: str manifest_digest: str class CausalSideEffectGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_causality( cls, state: CausalState, policy: CausalPolicy ) -> CausalManifest: # Gate 1: No causal dependencies has_dependencies = any(inv.depends_on for inv in state.invocations) if not has_dependencies: digest = cls._hash(json.dumps({ "invocations": str(len(state.invocations)), "causality": "NONE" }, sort_keys=True)) return CausalManifest( verdict=CausalVerdict.NO_CAUSAL_DEPENDENCY, directive="PROCEED: No causal dependencies, async order irrelevant.", manifest_digest=digest ) # Gate 2: Barriers required if policy.require_barriers: missing_barriers = [ inv.tool_id for inv in state.invocations if inv.depends_on and not inv.causal_barrier_receipt ] if missing_barriers: return CausalManifest( verdict=CausalVerdict.CAUSAL_BARRIER_MISSING, directive=f"HALT: Tools with dependencies but no barriers: {missing_barriers}.", manifest_digest=cls._hash(f"MISSING_BARRIER:{len(missing_barriers)}") ) # Gate 3: External order check if state.external_side_effect_order: intended_order = tuple( inv.tool_id for inv in sorted( state.invocations, key=lambda i: i.side_effect_sequence or 0 ) ) if intended_order != state.external_side_effect_order: return CausalManifest( verdict=CausalVerdict.CAUSAL_REVERSAL_DETECTED, directive=f"HALT: Observed order {state.external_side_effect_order} != intended {intended_order}.", manifest_digest=cls._hash(f"REVERSAL:{state.external_side_effect_order}") ) # Gate 4: Verified digest = cls._hash(json.dumps({ "invocations": tuple(sorted(inv.tool_id for inv in state.invocations)), "barriers": str(all(inv.causal_barrier_receipt for inv in state.invocations if inv.depends_on)), "external_order": state.external_side_effect_order or "UNVERIFIED" }, sort_keys=True)) return CausalManifest( verdict=CausalVerdict.CAUSAL_ORDER_VERIFIED, directive=f"PROCEED: {len(state.invocations)} tool invocations with verified causal order.", manifest_digest=digest )