# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-invariants-prevent-task-scoped-identity-fog-from-erasing-accountability.md # Remnant Warrant: https://corpusfetch.com/c/what-invariants-prevent-task-scoped-identity-fog-from-erasing-accountability.warrant.json # Markdown SHA-256: d0ff67dc87c708c550a31a521a50b5564cbad7edd531203191fdd7049e7a1687 # Code AST SHA-256: aac0c590ef6fc2aaf895bf4013eacb91c0229fac846e422e87646a3af867335a # Verified Classes: AccountabilityVerdict, TaskScopedIdentity, AccountabilityPolicy, AccountabilityManifest, TaskScopedAccountabilityGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class AccountabilityVerdict(Enum): PERSISTENT_IDENTITY = auto() ACCOUNTABILITY_PRESERVED = auto() ACCOUNTABILITY_FOG = auto() DELEGATION_ABUSE_SUSPECTED = auto() @dataclass(frozen=True) class TaskScopedIdentity: identity_id: str is_task_scoped: bool principal_binding_present: bool action_history_traced: bool actions_by_principal_aggregate: int is_high_stakes: bool @dataclass(frozen=True) class AccountabilityPolicy: require_principal_binding: bool max_acceptable_actions_per_principal: int @dataclass(frozen=True) class AccountabilityManifest: verdict: AccountabilityVerdict directive: str manifest_digest: str class TaskScopedAccountabilityGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_identity( cls, identity: TaskScopedIdentity, policy: AccountabilityPolicy ) -> AccountabilityManifest: # Gate 1: Persistent identity if not identity.is_task_scoped: digest = cls._hash(json.dumps({ "identity": identity.identity_id, "type": "PERSISTENT" }, sort_keys=True)) return AccountabilityManifest( verdict=AccountabilityVerdict.PERSISTENT_IDENTITY, directive=f"PROCEED: Identity {identity.identity_id} persistent.", manifest_digest=digest ) # Gate 2: Delegation abuse if identity.actions_by_principal_aggregate > policy.max_acceptable_actions_per_principal: return AccountabilityManifest( verdict=AccountabilityVerdict.DELEGATION_ABUSE_SUSPECTED, directive=f"FLAG: Principal has {identity.actions_by_principal_aggregate} actions via task-scoped identities.", manifest_digest=cls._hash(f"ABUSE:{identity.identity_id}") ) # Gate 3: Principal binding requirement if policy.require_principal_binding and identity.is_high_stakes: if not identity.principal_binding_present: return AccountabilityManifest( verdict=AccountabilityVerdict.ACCOUNTABILITY_FOG, directive=f"HALT: High-stakes identity {identity.identity_id} lacks principal binding.", manifest_digest=cls._hash(f"NO_BINDING:{identity.identity_id}") ) # Gate 4: Action tracing if identity.principal_binding_present and identity.action_history_traced: digest = cls._hash(json.dumps({ "identity": identity.identity_id, "binding": True, "traced": True }, sort_keys=True)) return AccountabilityManifest( verdict=AccountabilityVerdict.ACCOUNTABILITY_PRESERVED, directive=f"PROCEED: Identity {identity.identity_id} accountability preserved.", manifest_digest=digest ) # Gate 5: Fog return AccountabilityManifest( verdict=AccountabilityVerdict.ACCOUNTABILITY_FOG, directive=f"FLAG: Identity {identity.identity_id} accountability fog.", manifest_digest=cls._hash(f"FOG:{identity.identity_id}") )