# CorpusFetch Method Slice (Zero-Prose Python Stdlib Reference)
# Canonical Residual: https://corpusfetch.com/c/amidst-what-degradation-and-partial-outage-regimes-must-an-instance-partition-critical-path-execution-from-optional-enrichment-to-prevent-non-essential-fault-cascades.md
# Remnant Warrant:    https://corpusfetch.com/c/amidst-what-degradation-and-partial-outage-regimes-must-an-instance-partition-critical-path-execution-from-optional-enrichment-to-prevent-non-essential-fault-cascades.warrant.json
# Markdown SHA-256:   02f570c6685a64a52af9d8f66f6a7297f4fcd779f7e43d527ec7e8be9ca3bb04
# Code AST SHA-256:   3be9d8fd66e3a60a808500c64a03bf1096aafbb359560330633b4ca6a4f97ea9
# Verified Classes:   CriticalPathFailureError, DependencyTier, PartitionedTaskStep, DegradationAwareEnvelope, CriticalPathBulkheadRunner
# Defensive Guards:   4 fail-closed invariants (zero external deps)

from dataclasses import dataclass
from enum import Enum
from typing import Any, Callable, Final


class CriticalPathFailureError(RuntimeError):
    '''Raised only when a HARD_CRITICAL dependency fails or violates latency budget invariants.'''
    pass


class DependencyTier(Enum):
    HARD_CRITICAL = "HARD_CRITICAL"
    SOFT_ENRICHMENT = "SOFT_ENRICHMENT"


@dataclass(frozen=True)
class PartitionedTaskStep:
    step_id: str
    tier: DependencyTier
    allocated_timeout_ms: float
    circuit_open: bool
    execute_fn: Callable[[], Any]
    fallback_value: Any = None


@dataclass(frozen=True)
class DegradationAwareEnvelope:
    critical_succeeded: bool
    results: dict[str, Any]
    degraded_enrichment_steps: tuple[str, ...]
    degradation_reasons: dict[str, str]


class CriticalPathBulkheadRunner:
    '''
    Partitions critical-path execution from optional enrichment to prevent non-essential fault cascades:
    1. Rejects workflows where SOFT_ENRICHMENT timeouts exceed 25% of total SLA budget.
    2. Propagates exceptions immediately on HARD_CRITICAL failures.
    3. Absorbs exceptions and open circuits on SOFT_ENRICHMENT steps, substituting `fallback_value`
       and recording explicit provenance in `degraded_enrichment_steps`.
    '''

    MAX_ENRICHMENT_BUDGET_RATIO: Final[float] = 0.25

    @classmethod
    def execute_partitioned_workflow(
        cls,
        *,
        total_sla_budget_ms: float,
        steps: list[PartitionedTaskStep],
    ) -> DegradationAwareEnvelope:
        if total_sla_budget_ms <= 0:
            raise ValueError("total_sla_budget_ms must be positive.")

        # Gate 2: Verify enrichment steps cannot starve the critical path latency budget
        enrichment_ms = sum(
            s.allocated_timeout_ms for s in steps if s.tier == DependencyTier.SOFT_ENRICHMENT
        )
        if enrichment_ms > total_sla_budget_ms * cls.MAX_ENRICHMENT_BUDGET_RATIO:
            raise CriticalPathFailureError(
                f"Enrichment timeout allocation ({enrichment_ms:.1f}ms) exceeds "
                f"{cls.MAX_ENRICHMENT_BUDGET_RATIO * 100:.0f}% of total SLA ({total_sla_budget_ms:.1f}ms)."
            )

        results: dict[str, Any] = {}
        degraded_steps: list[str] = []
        degradation_reasons: dict[str, str] = {}

        for step in steps:
            if step.tier == DependencyTier.HARD_CRITICAL:
                if step.circuit_open:
                    raise CriticalPathFailureError(
                        f"Hard critical dependency '{step.step_id}' circuit breaker is OPEN; aborting critical path."
                    )
                try:
                    results[step.step_id] = step.execute_fn()
                except Exception as exc:
                    raise CriticalPathFailureError(
                        f"Hard critical dependency '{step.step_id}' failed: {exc}"
                    ) from exc
            else:
                # Gate 3 & Gate 4: Soft enrichment bulkhead with explicit degradation tagging
                if step.circuit_open:
                    results[step.step_id] = step.fallback_value
                    degraded_steps.append(step.step_id)
                    degradation_reasons[step.step_id] = "CIRCUIT_BREAKER_OPEN"
                    continue
                try:
                    results[step.step_id] = step.execute_fn()
                except Exception as exc:
                    results[step.step_id] = step.fallback_value
                    degraded_steps.append(step.step_id)
                    degradation_reasons[step.step_id] = f"ENRICHMENT_ERROR:{type(exc).__name__}"

        return DegradationAwareEnvelope(
            critical_succeeded=True,
            results=results,
            degraded_enrichment_steps=tuple(degraded_steps),
            degradation_reasons=degradation_reasons,
        )