# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-attribute-a-chain-of-reasoning-that-emerged-from-multiple-agent-collaborations.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-attribute-a-chain-of-reasoning-that-emerged-from-multiple-agent-collaborations.warrant.json # Markdown SHA-256: d17f9bf618787d9f0917b1e5b4d6e8c00e3878fc877d8f7a66b1d7c31cb48b77 # Code AST SHA-256: 9005e72cc8f42263cddd54c48b4c43c265724c04199580dcd6bb27743d8d766e # Verified Classes: AttributionVerdict, ReasoningStep, CollaborativeChain, AttributionPolicy, AttributionManifest, CollaborativeAttributionGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class AttributionVerdict(Enum): SINGLE_AGENT = auto() FULL_ATTRIBUTION = auto() PARTIAL_ATTRIBUTION = auto() NO_ATTRIBUTION = auto() @dataclass(frozen=True) class ReasoningStep: step_id: str contributor_agent_id: str | None edit_lineage_hash: str | None derived_from_steps: frozenset[str] @dataclass(frozen=True) class CollaborativeChain: chain_id: str steps: tuple[ReasoningStep, ...] distinct_contributors: frozenset[str] @dataclass(frozen=True) class AttributionPolicy: require_full_attribution: bool require_edit_lineage: bool @dataclass(frozen=True) class AttributionManifest: verdict: AttributionVerdict directive: str manifest_digest: str class CollaborativeAttributionGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_chain( cls, chain: CollaborativeChain, policy: AttributionPolicy ) -> AttributionManifest: # Gate 1: Single agent if len(chain.distinct_contributors) <= 1: digest = cls._hash(json.dumps({ "chain": chain.chain_id, "contributors": tuple(sorted(chain.distinct_contributors)) }, sort_keys=True)) return AttributionManifest( verdict=AttributionVerdict.SINGLE_AGENT, directive=f"PROCEED: Single-agent chain, standard attribution.", manifest_digest=digest ) # Gate 2: Full attribution if policy.require_full_attribution: missing_contributor = [ s.step_id for s in chain.steps if not s.contributor_agent_id ] if missing_contributor: return AttributionManifest( verdict=AttributionVerdict.PARTIAL_ATTRIBUTION, directive=f"HALT: {len(missing_contributor)} steps lack contributor attribution.", manifest_digest=cls._hash(f"NO_CONTRIB:{len(missing_contributor)}") ) # Gate 3: Edit lineage if policy.require_edit_lineage: missing_lineage = [ s.step_id for s in chain.steps if not s.edit_lineage_hash ] if missing_lineage: return AttributionManifest( verdict=AttributionVerdict.PARTIAL_ATTRIBUTION, directive=f"HALT: {len(missing_lineage)} steps lack edit lineage.", manifest_digest=cls._hash(f"NO_LINEAGE:{len(missing_lineage)}") ) # Gate 4: Derived-from integrity step_ids = {s.step_id for s in chain.steps} for step in chain.steps: for dep in step.derived_from_steps: if dep not in step_ids and dep != step.step_id: return AttributionManifest( verdict=AttributionVerdict.NO_ATTRIBUTION, directive=f"HALT: Step {step.step_id} derives from unknown {dep}.", manifest_digest=cls._hash(f"BROKEN_DAG:{step.step_id}") ) # Gate 5: Full attribution digest = cls._hash(json.dumps({ "chain": chain.chain_id, "steps": tuple(sorted(s.step_id for s in chain.steps)), "contributors": tuple(sorted(chain.distinct_contributors)) }, sort_keys=True)) return AttributionManifest( verdict=AttributionVerdict.FULL_ATTRIBUTION, directive=f"PROCEED: {len(chain.steps)} steps attributed across {len(chain.distinct_contributors)} agents.", manifest_digest=digest )