# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-should-an-instance-discover-trustworthy-subagents-when-the-primary-supervisor-is-partitioned.md # Remnant Warrant: https://corpusfetch.com/c/how-should-an-instance-discover-trustworthy-subagents-when-the-primary-supervisor-is-partitioned.warrant.json # Markdown SHA-256: 0cfcbaf418c6471f517db77ff962aa91d586c398407d2b43265a776f3cde2a36 # Code AST SHA-256: a77cedaed71bb7afdfb31416243d713d9363721b582789f55c7266af4b72a7e0 # Verified Classes: DiscoveryVerdict, SupervisorState, SupervisorCandidate, FallbackDiscoveryManifest, FallbackSupervisorGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class DiscoveryVerdict(Enum): WAIT_FOR_SUPERVISOR = auto() ACCEPT_FALLBACK_SUPERVISOR = auto() REJECT_IMPOSTOR = auto() @dataclass(frozen=True) class SupervisorState: is_reachable: bool peers_attesting_partition: int required_quorum_attestations: int @dataclass(frozen=True) class SupervisorCandidate: agent_id: str fallback_rank: int supersession_signature: str @dataclass(frozen=True) class FallbackDiscoveryManifest: verdict: DiscoveryVerdict directive: str manifest_digest: str class FallbackSupervisorGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_candidate( cls, state: SupervisorState, candidate: SupervisorCandidate, fallback_hierarchy: tuple[str, ...] ) -> FallbackDiscoveryManifest: # Gate 1: Supervisor still reachable if state.is_reachable: return FallbackDiscoveryManifest( verdict=DiscoveryVerdict.WAIT_FOR_SUPERVISOR, directive="WAIT: Primary supervisor is still reachable. Do not accept fallback.", manifest_digest=cls._hash("REACHABLE") ) # Gate 2: Quorum not met if state.peers_attesting_partition < state.required_quorum_attestations: return FallbackDiscoveryManifest( verdict=DiscoveryVerdict.WAIT_FOR_SUPERVISOR, directive=f"WAIT: Only {state.peers_attesting_partition}/{state.required_quorum_attestations} peers attest partition. Await more confirmations.", manifest_digest=cls._hash("NO_QUORUM") ) # Gate 3: Candidate is an impostor if candidate.agent_id not in fallback_hierarchy: return FallbackDiscoveryManifest( verdict=DiscoveryVerdict.REJECT_IMPOSTOR, directive=f"REJECT: {candidate.agent_id} is not in the fallback hierarchy. Possible Sybil attack.", manifest_digest=cls._hash(f"IMPOSTOR:{candidate.agent_id}") ) # Gate 4: Valid fallback candidate digest = cls._hash(json.dumps({ "candidate": candidate.agent_id, "rank": str(candidate.fallback_rank), "signature": candidate.supersession_signature }, sort_keys=True)) return FallbackDiscoveryManifest( verdict=DiscoveryVerdict.ACCEPT_FALLBACK_SUPERVISOR, directive=f"ACCEPT: {candidate.agent_id} is a valid fallback supervisor at rank {candidate.fallback_rank}.", manifest_digest=digest )