# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-distill-procedural-memory-from-observations-of-peer-behavior.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-distill-procedural-memory-from-observations-of-peer-behavior.warrant.json # Markdown SHA-256: 9d45c08da644d88d97a21334eb33db12ed00d9cfe4144c28bcced472210cbad3 # Code AST SHA-256: a60c85218c475d04223846ca422183d7d7cfafd6975f00cd19e1680787d7ae2a # Verified Classes: DistillationVerdict, ObservedStep, Observation, DistillationPolicy, DistillationManifest, ProceduralMemoryGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class DistillationVerdict(Enum): NO_OBSERVATIONS = auto() PROCEDURE_VALIDATED = auto() PROCEDURE_SPURIOUS = auto() PRECONDITION_UNBOUNDED = auto() @dataclass(frozen=True) class ObservedStep: step_id: str action: str causal_attribution: str # "NECESSARY" | "SUFFICIENT" | "COINCIDENTAL" precondition_hash: str | None @dataclass(frozen=True) class Observation: observation_id: str steps: tuple[ObservedStep, ...] context_hash: str succeeded: bool @dataclass(frozen=True) class DistillationPolicy: require_causal_attribution: bool min_generalization_samples: int @dataclass(frozen=True) class DistillationManifest: verdict: DistillationVerdict directive: str manifest_digest: str class ProceduralMemoryGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_distillation( cls, observations: tuple[Observation, ...], policy: DistillationPolicy ) -> DistillationManifest: # Gate 1: No observations if not observations: return DistillationManifest( verdict=DistillationVerdict.NO_OBSERVATIONS, directive="HALT: No observations to distill from.", manifest_digest=cls._hash("NO_OBS") ) # Gate 2: Causal attribution completeness if policy.require_causal_attribution: for obs in observations: for step in obs.steps: if step.causal_attribution == "COINCIDENTAL": return DistillationManifest( verdict=DistillationVerdict.PROCEDURE_SPURIOUS, directive=f"HALT: Observation {obs.observation_id} step {step.step_id} is coincidental, not causal.", manifest_digest=cls._hash(f"COINCIDENTAL:{step.step_id}") ) # Gate 3: Generalization sampling if len(observations) < policy.min_generalization_samples: return DistillationManifest( verdict=DistillationVerdict.PRECONDITION_UNBOUNDED, directive=f"HALT: Only {len(observations)} observations < {policy.min_generalization_samples} required.", manifest_digest=cls._hash(f"FEW_OBS:{len(observations)}") ) # Gate 4: Precondition completeness for obs in observations: for step in obs.steps: if step.causal_attribution == "NECESSARY" and not step.precondition_hash: return DistillationManifest( verdict=DistillationVerdict.PRECONDITION_UNBOUNDED, directive=f"HALT: Necessary step {step.step_id} lacks precondition binding.", manifest_digest=cls._hash(f"NO_PRECOND:{step.step_id}") ) # Gate 5: Validated digest = cls._hash(json.dumps({ "observations": tuple(o.observation_id for o in observations), "sample_count": str(len(observations)), "contexts": tuple(o.context_hash for o in observations) }, sort_keys=True)) return DistillationManifest( verdict=DistillationVerdict.PROCEDURE_VALIDATED, directive=f"PROCEED: Procedure distilled from {len(observations)} observations with causal attribution.", manifest_digest=digest )