f_Capture_Cascade.md

from dataclasses import dataclass
from typing import Optional
 
 
@dataclass
class CascadeEligibility:
    """Result of a cascade step eligibility evaluation."""
    eligible: bool
    k: int
    omega_k: float
    d_bind_k: float
    margin: float                   # omega_k - d_bind_k; positive = eligible
    termination_reason: Optional[str]  # None if eligible; T-NAT / T-DEPTH / T-CAP / T-INIT
 
 
def evaluate_cascade_eligibility(
    omega_k: float,
    d_bind_k: float,
    k: int,
    k_max: int,
    frame_count: int,
    capacity_MAX: int,
    gamma: float,
) -> CascadeEligibility:
    """
    Evaluate cascade step k for eligibility under conditions CAS-1 through CAS-4.
 
    This function is PURE — it does not mutate any external state.
    All four conditions are conjunctive; the first failure encountered terminates.
 
    Parameters
    ----------
    omega_k       : float  — Cascade transmission at depth k (already computed).
    d_bind_k      : float  — Binding threshold at depth k (field-perturbed).
    k             : int    — Current cascade depth (0-indexed, where k=0 is trigger).
    k_max         : int    — Maximum allowed cascade depth (hard ceiling).
    frame_count   : int    — Current number of registered elements in Frame.
    capacity_MAX  : int    — Maximum Frame capacity (from f_Frame.md §4.3).
    gamma         : float  — Cascade gain coefficient (checked for CAS-1 pre-flight).
 
    Returns
    -------
    CascadeEligibility
        eligible           : True iff all four conditions pass.
        k                  : Echo of depth parameter.
        omega_k            : Echo of cascade transmission.
        d_bind_k           : Echo of binding threshold.
        margin             : omega_k - d_bind_k (positive = eligible on CAS-3).
        termination_reason : None if eligible; one of T-INIT / T-DEPTH / T-NAT / T-CAP.
 
    Invariants enforced
    -------------------
    INV-001 : F_freq · F_fluid · F_force inseparability — all three nodes
              contributed to producing omega_k and d_bind_k upstream.
    INV-005 : Conditions are conjunctive; all must pass.
 
    Examples
    --------
    >>> evaluate_cascade_eligibility(
    ...     omega_k=0.48, d_bind_k=0.285, k=1, k_max=10,
    ...     frame_count=2, capacity_MAX=8, gamma=0.6
    ... )
    CascadeEligibility(eligible=True, k=1, omega_k=0.48, d_bind_k=0.285,
                       margin=0.195, termination_reason=None)
 
    >>> evaluate_cascade_eligibility(
    ...     omega_k=0.173, d_bind_k=0.255, k=3, k_max=10,
    ...     frame_count=2, capacity_MAX=8, gamma=0.6
    ... )
    CascadeEligibility(eligible=False, k=3, omega_k=0.173, d_bind_k=0.255,
                       margin=-0.082, termination_reason='T-NAT')
    """
    # CAS-1: Gain Positivity (pre-flight; caller should check before loop,
    # but guarded here defensively)
    if gamma <= 0.0:
        return CascadeEligibility(
            eligible=False, k=k, omega_k=omega_k, d_bind_k=d_bind_k,
            margin=omega_k - d_bind_k, termination_reason="T-INIT"
        )
 
    # CAS-2: Depth Bound
    if k >= k_max:
        return CascadeEligibility(
            eligible=False, k=k, omega_k=omega_k, d_bind_k=d_bind_k,
            margin=omega_k - d_bind_k, termination_reason="T-DEPTH"
        )
 
    # CAS-3: Binding Threshold
    margin = omega_k - d_bind_k
    if margin < 0.0:
        return CascadeEligibility(
            eligible=False, k=k, omega_k=omega_k, d_bind_k=d_bind_k,
            margin=margin, termination_reason="T-NAT"
        )
 
    # CAS-4: Frame Capacity Guard (FM-003-C)
    if frame_count + 1 > capacity_MAX:
        return CascadeEligibility(
            eligible=False, k=k, omega_k=omega_k, d_bind_k=d_bind_k,
            margin=margin, termination_reason="T-CAP"
        )
 
    return CascadeEligibility(
        eligible=True, k=k, omega_k=omega_k, d_bind_k=d_bind_k,
        margin=margin, termination_reason=None
    )

from dataclasses import dataclass, field
from typing import Any, Optional
 
 
@dataclass
class CascadeStepResult:
    """Result of executing a single cascade step."""
    status: str                       # CAPTURED / TERMINATED / FM-003-C
    k: int                            # cascade depth at which result was determined
    omega_k: float                    # cascade transmission at this step
    d_bind_k: float                   # binding threshold at this step
    phi_perturbed_k: float            # field value after perturbation at this step
    frame_count: int                  # frame count after this step (updated on CAPTURED)
    termination_reason: Optional[str] # None on CAPTURED; T-NAT / T-DEPTH / T-CAP / T-INIT
    margin: float                     # omega_k - d_bind_k
 
 
def execute_cascade_step(
    omega_prev: float,
    gamma: float,
    phi_perturbed_prev: float,
    k_perturb: float,
    d_bind_prev: float,
    e_prev: float,
    beta: float,
    e_k: float,
    k: int,
    k_max: int,
    frame_count: int,
    capacity_MAX: int,
    registry: Any,          # GravityGraph instance (f_Frame.md §4.5)
    element_id: str,
) -> CascadeStepResult:
    """
    Execute a single cascade step k, integrating field perturbation,
    transmission propagation, eligibility evaluation, and registry update.
 
    This function is IMPURE — it calls register_capture (PRIM:003) on
    the registry when the step is eligible, mutating Frame state.
 
    Parameters
    ----------
    omega_prev          : float  — Ω_cascade(k−1); cascade transmission from prior step.
    gamma               : float  — Cascade gain coefficient γ. Must be > 0 (CAS-1).
    phi_perturbed_prev  : float  — Φ_perturbed(k−1); field value after prior perturbation.
    k_perturb           : float  — Perturbation rate coefficient (system constant).
                                   From f_Capture_Multi.md §4.4.
    d_bind_prev         : float  — d_bind(k−1); prior step's binding threshold.
    e_prev              : float  — eccentricity at step k−1 (for δ_perturb computation).
    beta                : float  — F_fluid binding capacity (invariant across chain).
    e_k                 : float  — eccentricity at step k (for d_bind(k) computation).
    k                   : int    — Current cascade depth (k ≥ 1).
    k_max               : int    — Maximum cascade depth (CAS-2 ceiling).
    frame_count         : int    — Current Frame registration count before this step.
    capacity_MAX        : int    — Frame capacity ceiling (CAS-4).
    registry            : Any    — GravityGraph instance; mutated on CAPTURED.
    element_id          : str    — Identifier of the downstream candidate at depth k.
 
    Returns
    -------
    CascadeStepResult with fields described in class definition above.
 
    Side effects
    ------------
    On CAPTURED: calls registry.register_capture(element_id, ...) — mutates Frame.
    On TERMINATED / FM-003-C: no registry mutation.
 
    On FM-003-C: emits GravityGraph notification CASCADE_INTERRUPTED.
 
    Formulas applied (in order)
    ---------------------------
    1. δ_perturb(k) = d_bind_prev × (1 − e_prev) × k_perturb
    2. Φ_perturbed(k) = phi_perturbed_prev − δ_perturb(k)
       Φ_perturbed(k) = max(Φ_perturbed(k), 0.0)      # floor at 0
    3. ρ(Φ_perturbed(k)) = Φ_perturbed(k)              # simplified linear mapping
    4. d_bind(k) = beta × ρ(Φ_perturbed(k)) × (1 − e_k)
    5. Ω_cascade(k) = omega_prev × gamma
    6. eligibility = evaluate_cascade_eligibility(...)  # PRIM:027
 
    Invariants enforced
    -------------------
    INV-001 : Triadic product intact — ρ(Φ_perturbed(k)) · β · Ω_cascade(k).
    INV-003 : Frame capacity ceiling via CAS-4 / FM-003-C.
    INV-005 : CAS-1..CAS-4 conjunctive.
 
    Examples
    --------
    >>> result = execute_cascade_step(
    ...     omega_prev=0.8, gamma=0.6,
    ...     phi_perturbed_prev=0.9, k_perturb=0.05,
    ...     d_bind_prev=0.3, e_prev=0.1,
    ...     beta=0.5, e_k=0.1,
    ...     k=1, k_max=10, frame_count=2, capacity_MAX=8,
    ...     registry=graph, element_id="E-beta-1"
    ... )
    >>> result.status
    'CAPTURED'
    >>> result.omega_k
    0.48
    """
    # Step 1: Field perturbation
    delta_perturb_k = d_bind_prev * (1.0 - e_prev) * k_perturb
    phi_perturbed_k = max(phi_perturbed_prev - delta_perturb_k, 0.0)
 
    # Step 2: Field coherence at depth k (linear mapping)
    rho_phi_k = phi_perturbed_k
 
    # Step 3: Binding threshold at depth k
    d_bind_k = beta * rho_phi_k * (1.0 - e_k)
 
    # Step 4: Cascade transmission at depth k
    omega_k = omega_prev * gamma
 
    # Step 5: Eligibility evaluation (PRIM:027 — pure)
    eligibility = evaluate_cascade_eligibility(
        omega_k=omega_k,
        d_bind_k=d_bind_k,
        k=k,
        k_max=k_max,
        frame_count=frame_count,
        capacity_MAX=capacity_MAX,
        gamma=gamma,
    )
 
    if not eligibility.eligible:
        # FM-003-C: special notification for capacity exhaustion mid-chain
        if eligibility.termination_reason == "T-CAP":
            registry.notify("CASCADE_INTERRUPTED", {
                "element_id": element_id,
                "k": k,
                "frame_count": frame_count,
                "capacity_MAX": capacity_MAX,
                "fm": "FM-003-C",
            })
            return CascadeStepResult(
                status="FM-003-C", k=k, omega_k=omega_k, d_bind_k=d_bind_k,
                phi_perturbed_k=phi_perturbed_k, frame_count=frame_count,
                termination_reason="T-CAP", margin=eligibility.margin
            )
 
        return CascadeStepResult(
            status="TERMINATED", k=k, omega_k=omega_k, d_bind_k=d_bind_k,
            phi_perturbed_k=phi_perturbed_k, frame_count=frame_count,
            termination_reason=eligibility.termination_reason,
            margin=eligibility.margin
        )
 
    # Step 6: Register capture — IMPURE (mutates registry)
    registry.register_capture(element_id, d_bind=d_bind_k, depth=k)
    new_frame_count = frame_count + 1
 
    return CascadeStepResult(
        status="CAPTURED", k=k, omega_k=omega_k, d_bind_k=d_bind_k,
        phi_perturbed_k=phi_perturbed_k, frame_count=new_frame_count,
        termination_reason=None, margin=eligibility.margin
    )

@dataclass
class CascadeStepResult:
    status:        str    # "BOUND", "T-NAT", "T-DEPTH", "T-CAP"
    element_id:    str    # candidate processed
    k:             int    # step index
    omega_k:       float  # Ω_cascade(k) used
    phi_k:         float  # Φ_perturbed at this step
    d_bind_k:      float  # binding demand at this step
    bound:         bool   # True iff element was captured
    fm_triggered:  str | None  # "FM-003-C" or None

def execute_cascade_step(
    omega_prev: float,
    gamma: float,
    phi_perturbed_prev: float,
    k_perturb: float,
    d_bind_prev: float,
    e_prev: float,
    beta: float,
    e_k: float,
    k: int,
    k_max: int,
    frame_count: int,
    capacity_MAX: int,
    registry: list,
    element_id: str,
) -> CascadeStepResult:
    """
    Execute one step of a cascade chain.
 
    Guards checked in order:
      1. k_max ceiling  → T-DEPTH
      2. FM-003-C capacity → T-CAP
      3. CAS-3 binding condition → T-NAT (if fails)
      4. Success → BOUND
    """
 
    # ── Step 1: depth ceiling ──────────────────────────────────────────
    if k > k_max:
        return CascadeStepResult(
            status="T-DEPTH",
            element_id=element_id,
            k=k,
            omega_k=omega_prev * gamma,   # still compute for audit
            phi_k=phi_perturbed_prev,
            d_bind_k=None,
            bound=False,
            fm_triggered=None,
        )
 
    # ── Step 2: advance Ω and Φ ───────────────────────────────────────
    omega_k       = omega_prev * gamma
    delta_perturb = d_bind_prev * (1.0 - e_prev) * k_perturb
    phi_k         = max(0.0, phi_perturbed_prev - delta_perturb)
 
    # ── Step 3: compute d_bind(k) ─────────────────────────────────────
    # d_bind(k) = beta * (1 − e_k) * phi_k   (from f_Capture.md §3)
    d_bind_k = beta * (1.0 - e_k) * phi_k
 
    # ── Step 4: FM-003-C capacity guard ───────────────────────────────
    if frame_count >= capacity_MAX:
        return CascadeStepResult(
            status="T-CAP",
            element_id=element_id,
            k=k,
            omega_k=omega_k,
            phi_k=phi_k,
            d_bind_k=d_bind_k,
            bound=False,
            fm_triggered="FM-003-C",
        )
 
    # ── Step 5: CAS-3 binding condition ───────────────────────────────
    if omega_k < d_bind_k:
        return CascadeStepResult(
            status="T-NAT",
            element_id=element_id,
            k=k,
            omega_k=omega_k,
            phi_k=phi_k,
            d_bind_k=d_bind_k,
            bound=False,
            fm_triggered=None,
        )
 
    # ── Step 6: capture ───────────────────────────────────────────────
    registry.append(element_id)
    return CascadeStepResult(
        status="BOUND",
        element_id=element_id,
        k=k,
        omega_k=omega_k,
        phi_k=phi_k,
        d_bind_k=d_bind_k,
        bound=True,
        fm_triggered=None,
    )

def run_cascade(
    d_bind_0: float,
    gamma: float,
    phi_0: float,
    k_perturb: float,
    beta: float,
    candidates: list[dict],   # each: {"id": str, "e": float}
    k_max: int,
    capacity_MAX: int,
    registry: list,
) -> dict:
    """
    Full cascade orchestration.
 
    candidates: ordered list of dicts with keys 'id' (str) and 'e' (float).
    Returns summary dict with termination code, chain depth, and step log.
    """
 
    # ── PRIM:027: validate eligibility before any steps ───────────────
    elig = evaluate_cascade_eligibility(
        gamma=gamma,
        phi_0=phi_0,
        d_bind_0=d_bind_0,
        k_max=k_max,
        capacity_MAX=capacity_MAX,
        frame_count=len(registry),
    )
    if not elig["eligible"]:
        return {
            "termination": "T-INIT",
            "reason": elig["reason"],
            "chain_depth": 0,
            "steps": [],
        }
 
    omega_prev         = d_bind_0
    phi_prev           = phi_0
    d_bind_prev        = d_bind_0
    e_prev             = candidates[0]["e"] if candidates else 0.0
    steps              = []
 
    for k, candidate in enumerate(candidates, start=1):
        result = execute_cascade_step(
            omega_prev        = omega_prev,
            gamma             = gamma,
            phi_perturbed_prev= phi_prev,
            k_perturb         = k_perturb,
            d_bind_prev       = d_bind_prev,
            e_prev            = e_prev,
            beta              = beta,
            e_k               = candidate["e"],
            k                 = k,
            k_max             = k_max,
            frame_count       = len(registry),
            capacity_MAX      = capacity_MAX,
            registry          = registry,
            element_id        = candidate["id"],
        )
        steps.append(result)
 
        if result.status != "BOUND":
            return {
                "termination": result.status,
                "chain_depth": k - 1,
                "steps": steps,
                "fm_triggered": result.fm_triggered,
            }
 
        # advance state for next step
        omega_prev  = result.omega_k
        phi_prev    = result.phi_k
        d_bind_prev = result.d_bind_k
        e_prev      = candidate["e"]
 
    # exhausted candidate list without a stopping condition
    return {
        "termination": "T-NAT",
        "chain_depth": len(steps),
        "steps": steps,
        "fm_triggered": None,
    }

### Wave 4 Operators — f_Capture_Cascade.md
 
| Symbol         | Name              | Type   | Domain      | Defined In              |
|----------------|-------------------|--------|-------------|-------------------------|
| cascade_depth  | Cascade Depth     | int    | ≥ 0         | f_Capture_Cascade §4.1  |
| k_max          | Max Chain Depth   | int    | ≥ 1         | f_Capture_Cascade §4.1  |
| γ (gamma)      | Cascade Gain      | float  | > 0         | f_Capture_Cascade §4.2  |
| Ω_cascade(k)   | Cascade Binding   | float  | ≥ 0         | f_Capture_Cascade §4.3  |
| k_perturb      | Perturb Coeff     | float  | [0, 1)      | f_Capture_Multi §4      |
| δ_perturb(k)   | Field Perturbation| float  | ≥ 0         | f_Capture_Multi §4      |
| Φ_perturbed(k) | Perturbed Field   | float  | [0, 1]      | f_Capture_Multi §4      |

## [Wave 4] f_Capture_Cascade.md — Initial Release
 
### Added
- Cascade capture variant: sequential chain mechanics with geometric gain γ
- Operators: cascade_depth, k_max, γ, Ω_cascade(k)
- Conditions: CAS-1 (γ validity), CAS-2 (depth ceiling), CAS-3 (binding),
  CAS-4 (field floor)
- Failure mode: FM-003-C (Cascade Frame Saturation), sub-mode of FM-003
- Primitives: PRIM:027 evaluate_cascade_eligibility (Pure),
  PRIM:028 execute_cascade_step (Impure)
- Orchestration wrapper: run_cascade (convenience, not a registered PRIM)
- Termination codes: T-NAT, T-DEPTH, T-CAP, T-INIT
- Canonical examples: 4 (attenuating, mid-chain FM-003-C,
  neutral deep chain, amplifying)
- OPERATORS.md registration block (§9.3)
 
### Cross-references
- Upstream: f_Capture.md, f_Capture_Multi.md, f_Field.md, f_Force.md,
  f_Frame.md, f_Orbit.md, f_Amplify.md, f_Dampen.md
- Downstream: f_Capture_Soft.md, f_Capture_Resonant.md,
  f_Collapse.md, f_Emit.md, f_Decay.md