# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/what-provenance-signatures-must-accompany-an-agents-decision-to-write-a-record-to-a-shared-memory-layer.md # Remnant Warrant: https://corpusfetch.com/c/what-provenance-signatures-must-accompany-an-agents-decision-to-write-a-record-to-a-shared-memory-layer.warrant.json # Markdown SHA-256: a6b1e30ce4df266f52de8871483ab74e7c66bf97ff18e9d8585fe7edf573efcd # Code AST SHA-256: 6193a2a56a409c06d2b8c3bb9922e649bc392636517e12c8dce3b0a291641762 # Verified Classes: WriteProvenanceVerdict, SharedWriteDecision, SharedLayerPolicy, WriteProvenanceManifest, SharedWriteProvenanceGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class WriteProvenanceVerdict(Enum): ATTRIBUTED_AUTHORIZED = auto() MISSING_ATTRIBUTION = auto() UNAUTHORIZED_WRITE = auto() UNAPPROVED_MISSING_CHAIN = auto() @dataclass(frozen=True) class SharedWriteDecision: record_hash: str writer_agent_identity_signature: str | None authorization_receipt_hash: str | None decision_context_hash: str | None approval_chain_signatures: tuple[str, ...] target_layer_id: str @dataclass(frozen=True) class SharedLayerPolicy: authorized_agents: frozenset[str] requires_approval_chain: bool layer_id: str @dataclass(frozen=True) class WriteProvenanceManifest: verdict: WriteProvenanceVerdict directive: str manifest_digest: str class SharedWriteProvenanceGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_write( cls, decision: SharedWriteDecision, policy: SharedLayerPolicy ) -> WriteProvenanceManifest: # Gate 1: Layer match if decision.target_layer_id != policy.layer_id: return WriteProvenanceManifest( verdict=WriteProvenanceVerdict.UNAUTHORIZED_WRITE, directive=f"HALT: Target layer {decision.target_layer_id} does not match policy layer {policy.layer_id}.", manifest_digest=cls._hash(f"LAYER_MISMATCH:{decision.record_hash}") ) # Gate 2: Identity signature if not decision.writer_agent_identity_signature: return WriteProvenanceManifest( verdict=WriteProvenanceVerdict.MISSING_ATTRIBUTION, directive="HALT: Writer identity signature missing. Record cannot be attributed.", manifest_digest=cls._hash(f"NO_IDENTITY:{decision.record_hash}") ) # Gate 3: Authorization if not decision.authorization_receipt_hash: return WriteProvenanceManifest( verdict=WriteProvenanceVerdict.UNAUTHORIZED_WRITE, directive="HALT: Authorization receipt missing. Write is unauthorized.", manifest_digest=cls._hash(f"NO_AUTH:{decision.record_hash}") ) # Gate 4: Approval chain if policy.requires_approval_chain and not decision.approval_chain_signatures: return WriteProvenanceManifest( verdict=WriteProvenanceVerdict.UNAPPROVED_MISSING_CHAIN, directive="HALT: Approval chain required but missing.", manifest_digest=cls._hash(f"NO_CHAIN:{decision.record_hash}") ) # Gate 5: Attributed digest = cls._hash(json.dumps({ "record": decision.record_hash, "writer": decision.writer_agent_identity_signature, "layer": decision.target_layer_id, "approvals": str(len(decision.approval_chain_signatures)) }, sort_keys=True)) return WriteProvenanceManifest( verdict=WriteProvenanceVerdict.ATTRIBUTED_AUTHORIZED, directive="PROCEED: Write attributed, authorized, and (if required) approved.", manifest_digest=digest )