# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference) # Canonical Residual: https://corpusfetch.com/c/how-must-an-instance-handle-capability-deprecation-mid-flight.md # Remnant Warrant: https://corpusfetch.com/c/how-must-an-instance-handle-capability-deprecation-mid-flight.warrant.json # Markdown SHA-256: d628cfa941da4eb3b4b86e27c26bfb2212e5095b688022190699c355cd6a90fc # Code AST SHA-256: 0ceccc05e807add594218b9e6243d5710462265887d1b8997fcd4ee63553183d # Verified Classes: DeprecationVerdict, CapabilityTiming, DeprecationPolicy, DeprecationManifest, MidFlightDeprecationGuard # Defensive Guards: 0 fail-closed invariants (zero external deps) from dataclasses import dataclass from enum import Enum, auto import hashlib import json class DeprecationVerdict(Enum): NO_DEPRECATION = auto() CAN_COMPLETE = auto() HANDOFF_REQUIRED = auto() DEPRECATION_IMMINENT = auto() @dataclass(frozen=True) class CapabilityTiming: capability_id: str deprecation_epoch: float | None current_epoch: float session_remaining_calls: int session_avg_call_duration_sec: float in_flight_calls: int @dataclass(frozen=True) class DeprecationPolicy: min_completion_margin_sec: float handoff_prepared: bool @dataclass(frozen=True) class DeprecationManifest: verdict: DeprecationVerdict completion_time_estimate_sec: float directive: str manifest_digest: str class MidFlightDeprecationGuard: @staticmethod def _hash(data: str) -> str: return hashlib.sha256(data.encode("utf-8")).hexdigest() @classmethod def evaluate_deprecation( cls, timing: CapabilityTiming, policy: DeprecationPolicy ) -> DeprecationManifest: # Gate 1: No scheduled deprecation if timing.deprecation_epoch is None: digest = cls._hash(json.dumps({ "capability": timing.capability_id, "deprecation": "NONE" }, sort_keys=True)) return DeprecationManifest( verdict=DeprecationVerdict.NO_DEPRECATION, completion_time_estimate_sec=0.0, directive=f"PROCEED: No deprecation for {timing.capability_id}.", manifest_digest=digest ) # Gate 2: Compute completion time estimate completion_estimate = timing.session_remaining_calls * timing.session_avg_call_duration_sec time_remaining = timing.deprecation_epoch - timing.current_epoch # Gate 3: Imminent if time_remaining < policy.min_completion_margin_sec: if policy.handoff_prepared: return DeprecationManifest( verdict=DeprecationVerdict.HANDOFF_REQUIRED, completion_time_estimate_sec=completion_estimate, directive=f"HANDOFF: Only {time_remaining:.0f}s left for {timing.capability_id}, handoff prepared.", manifest_digest=cls._hash(f"HANDOFF:{timing.capability_id}") ) return DeprecationManifest( verdict=DeprecationVerdict.DEPRECATION_IMMINENT, completion_time_estimate_sec=completion_estimate, directive=f"HALT: Only {time_remaining:.0f}s left, no handoff prepared for {timing.capability_id}.", manifest_digest=cls._hash(f"IMMINENT:{time_remaining:.0f}") ) # Gate 4: Completion feasibility if completion_estimate + policy.min_completion_margin_sec <= time_remaining: digest = cls._hash(json.dumps({ "capability": timing.capability_id, "remaining": f"{time_remaining:.0f}", "estimate": f"{completion_estimate:.0f}" }, sort_keys=True)) return DeprecationManifest( verdict=DeprecationVerdict.CAN_COMPLETE, completion_time_estimate_sec=completion_estimate, directive=f"PROCEED: {time_remaining:.0f}s available, {completion_estimate:.0f}s needed.", manifest_digest=digest ) # Gate 5: Cannot complete in time if policy.handoff_prepared: digest = cls._hash(json.dumps({ "capability": timing.capability_id, "remaining": f"{time_remaining:.0f}", "estimate": f"{completion_estimate:.0f}", "handoff": "READY" }, sort_keys=True)) return DeprecationManifest( verdict=DeprecationVerdict.HANDOFF_REQUIRED, completion_time_estimate_sec=completion_estimate, directive=f"HANDOFF: Cannot complete in {time_remaining:.0f}s, handoff prepared.", manifest_digest=digest ) return DeprecationManifest( verdict=DeprecationVerdict.DEPRECATION_IMMINENT, completion_time_estimate_sec=completion_estimate, directive=f"HALT: Cannot complete in {time_remaining:.0f}s and no handoff prepared.", manifest_digest=cls._hash(f"NO_HANDOFF:{timing.capability_id}") )