controlai_data/generators/extended_v1.py

"""Safety, aerospace, stochastic, fault-tolerant, and learning-control families with Chain of Thought (CoT)."""

from __future__ import annotations

from collections.abc import Callable

import numpy as np

from controlai_data.schema import make_record


OPENERS = (
    "Provide a detailed step-by-step engineering calculation with intermediate derivations.",
    "Solve this numerical control problem step-by-step without omitting intermediate math.",
    "State the governing equations, compute the requested values, and verify the conclusion.",
    "Give a rigorous control engineering derivation with verified numbers.",
)


def number(value: float) -> str:
    if abs(value) < 5e-12:
        value = 0.0
    return f"{value:.10g}"


def record(
    *,
    index: int,
    prefix: str,
    domain: str,
    family: str,
    difficulty: str,
    prompt: str,
    answer: str,
    ground_truth: dict,
    source_refs: list[str],
    verifier: str,
) -> dict:
    return make_record(
        record_id=f"{prefix}_{index:05d}",
        domain=domain,
        family=family,
        task_type="numerical_reasoning",
        difficulty=difficulty,
        template_id=f"{prefix}_prompt_{index % 4}",
        prompt=prompt,
        answer=answer,
        ground_truth=ground_truth,
        source_refs=source_refs,
        verifier=verifier,
    )


def nonlinear_dynamic_inversion(_: np.random.Generator, index: int) -> dict:
    a = -0.7 + 0.018 * index
    b = 0.8 + 0.012 * index
    x = -1.2 + 0.041 * index
    reference = 0.35 - 0.006 * index
    gain = 1.1 + 0.017 * index
    virtual = -gain * (x - reference)
    control = (virtual - a * x) / b
    
    prompt = (
        f"{OPENERS[index % 4]} For scalar affine system x_dot={number(a)}x+{number(b)}u, "
        f"use exact nonlinear dynamic inversion so that x_dot=v=-{number(gain)}(x-r). "
        f"At x={number(x)}, r={number(reference)}, compute the virtual control v and the required control input u."
    )
    answer = (
        f"### 1. Virtual Control Law Definition\n"
        f"To enforce the target error dynamics $\\dot{{e}} = \\dot{{x}} - \\dot{{r}} = -k(x - r)$ with setpoint $r = {number(reference)}$:\n"
        f"$$v = -k (x - r) = -{number(gain)} \\times ({number(x)} - {number(reference)}) = -{number(gain)} \\times ({number(x - reference)}) = {number(virtual)}$$\n\n"
        f"### 2. Exact Dynamic Inversion Algebraic Solution\n"
        f"Equating the open-loop dynamics $\\dot{{x}} = a x + b u$ to the synthetic virtual input $v$:\n"
        f"$$a x + b u = v \\implies u = \\frac{{v - a x}}{{b}}$$\n"
        f"$$u = \\frac{{{number(virtual)} - ({number(a)})({number(x)})}}{{{number(b)}}} = \\frac{{{number(virtual)} - ({number(a * x)})}}{{{number(b)}}} = {number(control)}$$\n\n"
        f"### 3. Direct Closed-Loop Verification\n"
        f"$$\\dot{{x}} = ({number(a)})({number(x)}) + ({number(b)})({number(control)}) = {number(a * x)} + {number(b * control)} = {number(a * x + b * control)} = v$$\n"
        f"**Summary:** Virtual control v = {number(virtual)}, control input u = {number(control)}."
    )
    return record(
        index=index,
        prefix="ndi_scalar",
        domain="aerospace_control",
        family="scalar_affine_nonlinear_dynamic_inversion",
        difficulty="advanced",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "scalar_dynamic_inversion",
            "a": a,
            "b": b,
            "x": x,
            "reference": reference,
            "gain": gain,
            "virtual": virtual,
            "u": control,
        },
        source_refs=["slotine_li_applied_nonlinear_control", "ndi_control_literature"],
        verifier="analytic_scalar_dynamic_inversion",
    )


def gain_schedule(_: np.random.Generator, index: int) -> dict:
    low = np.array([0.8 + 0.01 * index, 0.25 + 0.004 * index])
    high = np.array([2.0 + 0.014 * index, 0.75 + 0.006 * index])
    sigma_low = 10.0
    sigma_high = 30.0 + 0.1 * index
    sigma = 11.0 + (index % 17) * 0.9
    weight = (sigma - sigma_low) / (sigma_high - sigma_low)
    scheduled = (1 - weight) * low + weight * high
    
    prompt = (
        f"{OPENERS[index % 4]} Linearly schedule gain vector K between K_low={low.tolist()} at "
        f"sigma={number(sigma_low)} and K_high={high.tolist()} at sigma={number(sigma_high)}. "
        f"Find K at operating point sigma={number(sigma)}; do not extrapolate or claim global stability."
    )
    answer = (
        f"### 1. Interpolation Weight Calculation\n"
        f"$$\\mu = \\frac{{\\sigma - \\sigma_{{low}}}}{{\\sigma_{{high}} - \\sigma_{{low}}}} = \\frac{{{number(sigma)} - {number(sigma_low)}}}{{{number(sigma_high)} - {number(sigma_low)}}} = {number(weight)}$$\n\n"
        f"### 2. Gain Vector Linear Interpolation\n"
        f"$$K(\\sigma) = (1 - \\mu) K_{{low}} + \\mu K_{{high}} = (1 - {number(weight)}) {low.tolist()} + {number(weight)} {high.tolist()} = {scheduled.tolist()}$$\n\n"
        f"### 3. Engineering Context\n"
        f"The scheduled gain is $K = {scheduled.tolist()}$. This gain scheduling is valid for slow parameter variation; it does not constitute a global nonlinear stability certificate."
    )
    return record(
        index=index,
        prefix="gain_schedule",
        domain="aerospace_control",
        family="linear_gain_schedule_interpolation",
        difficulty="intermediate",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "gain_schedule_interpolation",
            "low": low.tolist(),
            "high": high.tolist(),
            "sigma_low": sigma_low,
            "sigma_high": sigma_high,
            "sigma": sigma,
            "weight": weight,
            "scheduled": scheduled.tolist(),
        },
        source_refs=["gain_scheduling_control_literature"],
        verifier="analytic_linear_interpolation",
    )


def cbf_filter(_: np.random.Generator, index: int) -> dict:
    x_min = -0.4 + 0.005 * index
    x = x_min + 0.08 + 0.012 * (index % 9)
    alpha = 0.8 + 0.025 * index
    u_nom = -1.1 + 0.033 * index
    h = x - x_min
    lower = -alpha * h
    safe = max(u_nom, lower)
    active = u_nom < lower
    
    prompt = (
        f"{OPENERS[index % 4]} A single integrator x_dot=u must satisfy safety set x>={number(x_min)}. "
        f"With barrier function h(x)=x-x_min, enforce the forward-invariance CBF condition h_dot+{number(alpha)}h>=0 "
        f"while minimally modifying nominal command u_nom={number(u_nom)}. At current state x={number(x)}, compute the filtered control input u."
    )
    answer = (
        f"### 1. Control Barrier Function (CBF) Condition\n"
        f"For safety barrier $h(x) = x - x_{{min}} = {number(x)} - ({number(x_min)}) = {number(h)}$:\n"
        f"$$\\dot{{h}}(x, u) + \\alpha h(x) \\ge 0 \\implies u + {number(alpha)} ({number(h)}) \\ge 0 \\implies u \\ge -{number(alpha)} \\times {number(h)} = {number(lower)}$$\n\n"
        f"### 2. Quadratic Program Safety Filter Projection\n"
        f"$$\\min_u \\frac{{1}}{{2}} (u - u_{{nom}})^2 \\quad \\text{{s.t.}} \\quad u \\ge {number(lower)}$$\n"
        f"The analytical scalar projection is:\n"
        f"$$u^* = \\max(u_{{nom}}, -\\alpha h) = \\max({number(u_nom)}, {number(lower)}) = {number(safe)}$$\n\n"
        f"### 3. Verification\n"
        f"- Constraint status: **{'Active (intervened to enforce safety)' if active else 'Inactive (nominal command is already safe)'}**\n"
        f"- Residual safety derivative: $\\dot{{h}} + \\alpha h = {number(safe)} + {number(alpha * h)} = {number(safe + alpha * h)} \\ge 0$\n\n"
        f"**Summary:** Filtered control u = {number(safe)}."
    )
    return record(
        index=index,
        prefix="cbf_scalar",
        domain="safety_critical_control",
        family="scalar_control_barrier_safety_filter",
        difficulty="advanced",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "scalar_cbf_filter",
            "x_min": x_min,
            "x": x,
            "alpha": alpha,
            "u_nom": u_nom,
            "h": h,
            "lower_bound": lower,
            "u_safe": safe,
            "active": active,
        },
        source_refs=["ames_control_barrier_functions"],
        verifier="analytic_scalar_cbf_projection",
    )


def scalar_stationary_covariance(_: np.random.Generator, index: int) -> dict:
    a = 0.2 + 0.009 * index
    q = 0.05 + 0.003 * index
    p = q / (1 - a * a)
    prompt = (
        f"{OPENERS[index % 4]} For discrete stochastic system x_(k+1)={number(a)}x_k+w_k with E[w_k^2]="
        f"{number(q)}, compute the stationary variance P if it exists and verify the Lyapunov equation."
    )
    answer = (
        f"### 1. Existence of Stationary Second Moment\n"
        f"For $x_{{k+1}} = a x_k + w_k$, a stationary variance exists if and only if $|a| < 1$.\n"
        f"Here $|a| = {number(abs(a))} < 1$, so the state process is stationary.\n\n"
        f"### 2. Discrete Lyapunov Equation for Covariance\n"
        f"$$P = a^2 P + Q \\implies (1 - a^2) P = Q \\implies P = \\frac{{Q}}{{1 - a^2}}$$\n"
        f"$$P = \\frac{{{number(q)}}}{{1 - ({number(a)})^2}} = \\frac{{{number(q)}}}{{1 - {number(a * a)}}} = \\frac{{{number(q)}}}{{{number(1 - a * a)}}} = {number(p)}$$\n\n"
        f"### 3. Verification\n"
        f"$$a^2 P + Q = ({number(a * a)})({number(p)}) + {number(q)} = {number(a * a * p + q)} = P$$\n"
        f"**Summary:** Stationary variance P = {number(p)}."
    )
    return record(
        index=index,
        prefix="stationary_covariance",
        domain="stochastic_control",
        family="scalar_stationary_process_covariance",
        difficulty="intermediate",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "scalar_stationary_covariance",
            "a": a,
            "q": q,
            "P": p,
            "exists": abs(a) < 1,
        },
        source_refs=["anderson_moore_optimal_filtering"],
        verifier="scalar_discrete_lyapunov",
    )


def fault_allocation(_: np.random.Generator, index: int) -> dict:
    effectiveness = np.array([1.0, 0.75 + 0.003 * index, 0.45 + 0.002 * index])
    desired = -1.0 + 0.047 * index
    b_norm_sq = float(effectiveness @ effectiveness)
    allocation = effectiveness * desired / b_norm_sq
    achieved = float(effectiveness @ allocation)
    
    prompt = (
        f"{OPENERS[index % 4]} Three redundant actuators have effectiveness row "
        f"B={effectiveness.tolist()}. Find the unconstrained minimum-2-norm control command u "
        f"that produces desired virtual control tau={number(desired)}, and verify Bu=tau."
    )
    answer = (
        f"### 1. Minimum 2-Norm Control Allocation Formulation\n"
        f"$$\\min_u \\frac{{1}}{{2}} \\|u\\|_2^2 \\quad \\text{{s.t.}} \\quad B u = \\tau$$\n"
        f"For full-row-rank matrix $B$, the unique minimum-norm solution via Moore-Penrose pseudoinverse is:\n"
        f"$$u^* = B^T (B B^T)^{{-1}} \\tau = \\frac{{\\tau}}{{B B^T}} B^T$$\n\n"
        f"### 2. Evaluation\n"
        f"- Denominator: $B B^T = \\|B\\|_2^2 = 1.0^2 + {number(effectiveness[1])}^2 + {number(effectiveness[2])}^2 = {number(b_norm_sq)}$\n"
        f"- Multiplier: $\\frac{{\\tau}}{{\\|B\\|_2^2}} = \\frac{{{number(desired)}}}{{{number(b_norm_sq)}}} = {number(desired / b_norm_sq)}$\n"
        f"- Allocated actuator commands:\n"
        f"  $$u^* = {allocation.tolist()}$$\n\n"
        f"### 3. Virtual Control Verification\n"
        f"$$B u^* = {effectiveness.tolist()} \\cdot {allocation.tolist()} = {number(achieved)}$$\n"
        f"Residual: $|B u^* - \\tau| = {number(abs(achieved - desired))} \\approx 0$.\n\n"
        f"**Summary:** Allocated command u = {allocation.tolist()}."
    )
    return record(
        index=index,
        prefix="fault_allocation",
        domain="fault_tolerant_control",
        family="minimum_norm_redundant_control_allocation",
        difficulty="advanced",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "minimum_norm_allocation",
            "B": effectiveness.tolist(),
            "desired": desired,
            "u": allocation.tolist(),
            "achieved": achieved,
        },
        source_refs=["fault_tolerant_control_allocation_literature"],
        verifier="pseudoinverse_allocation",
    )


def jump_scalar(_: np.random.Generator, index: int) -> dict:
    probability = 0.2 + 0.006 * (index % 30)
    a1 = 0.45 + 0.003 * index
    a2 = 1.02 + 0.002 * index
    factor = probability * a1 * a1 + (1 - probability) * a2 * a2
    stable = factor < 1
    
    prompt = (
        f"{OPENERS[index % 4]} An i.i.d. scalar jump linear system x_(k+1)=a_k x_k uses multiplier a1={number(a1)} "
        f"with probability p={number(probability)} and a2={number(a2)} with probability 1-p={number(1 - probability)}. "
        "Compute the expected squared multiplier E[a^2] and decide mean-square stability."
    )
    answer = (
        f"### 1. Second-Moment Dynamics\n"
        f"For the i.i.d. scalar jump system $x_{{k+1}} = a_k x_k$:\n"
        f"$$\\mathbb{{E}}[x_{{k+1}}^2] = \\mathbb{{E}}[a_k^2] \\mathbb{{E}}[x_k^2]$$\n"
        f"The system is mean-square stable if and only if $\\mathbb{{E}}[a^2] < 1$.\n\n"
        f"### 2. Expected Squared Multiplier Calculation\n"
        f"$$\\mathbb{{E}}[a^2] = p a_1^2 + (1 - p) a_2^2$$\n"
        f"$$\\mathbb{{E}}[a^2] = ({number(probability)}) ({number(a1)})^2 + ({number(1 - probability)}) ({number(a2)})^2$$\n"
        f"$$\\mathbb{{E}}[a^2] = ({number(probability)}) ({number(a1 * a1)}) + ({number(1 - probability)}) ({number(a2 * a2)}) = {number(factor)}$$\n\n"
        f"### 3. Stability Conclusion\n"
        f"Since $\\mathbb{{E}}[a^2] = {number(factor)}$ {'< 1' if stable else '>= 1'}, the system is **{'mean-square stable' if stable else 'not certified mean-square stable'}**."
    )
    return record(
        index=index,
        prefix="jump_scalar",
        domain="stochastic_control",
        family="iid_scalar_jump_mean_square_test",
        difficulty="advanced",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "iid_jump_mean_square",
            "p": probability,
            "a1": a1,
            "a2": a2,
            "factor": factor,
            "stable": stable,
        },
        source_refs=["markov_jump_linear_systems_literature"],
        verifier="expected_square_multiplier",
    )


def event_trigger(_: np.random.Generator, index: int) -> dict:
    state = -1.4 + 0.05 * index
    held = state + (-0.3 + 0.011 * index)
    sigma = 0.12 + 0.002 * index
    error = held - state
    threshold = sigma * abs(state)
    trigger = abs(error) >= threshold
    prompt = (
        f"{OPENERS[index % 4]} An event-triggered control mechanism broadcasts an update when |x_hat-x|>=sigma|x|. "
        f"Given current state x={number(state)}, last transmitted state x_hat={number(held)}, and threshold parameter sigma={number(sigma)}, "
        "compute both sides of the condition and decide whether an event is triggered."
    )
    answer = (
        f"### 1. State Error and Threshold Evaluation\n"
        f"- State discrepancy: $|\\hat{{x}} - x| = |{number(held)} - ({number(state)})| = |{number(error)}| = {number(abs(error))}$\n"
        f"- State-dependent threshold: $\\sigma |x| = {number(sigma)} \\times |{number(state)}| = {number(threshold)}$\n\n"
        f"### 2. Trigger Condition Check\n"
        f"Condition $|\\hat{{x}} - x| \\ge \\sigma |x|$ evaluates to ${number(abs(error))} \\ge {number(threshold)}$, which is **{trigger}**.\n\n"
        f"**Summary:** Trigger is {str(trigger).lower()}."
    )
    return record(
        index=index,
        prefix="event_trigger",
        domain="networked_control",
        family="relative_state_error_event_trigger",
        difficulty="intermediate",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "relative_event_trigger",
            "x": state,
            "x_hat": held,
            "sigma": sigma,
            "error_abs": abs(error),
            "threshold": threshold,
            "trigger": trigger,
        },
        source_refs=["event_triggered_control_literature"],
        verifier="analytic_event_threshold",
    )


def tube_tightening(_: np.random.Generator, index: int) -> dict:
    bound = 2.0 + 0.025 * index
    error = 0.15 + 0.004 * index
    tightened = bound - error
    prompt = (
        f"{OPENERS[index % 4]} In scalar tube MPC with state decomposition x=z+e, the true state constraint is "
        f"|x|<={number(bound)}, and the robustly invariant error tube is |e|<={number(error)}. "
        "Compute the Pontryagin-tightened nominal state constraint interval for z."
    )
    answer = (
        f"### 1. Pontryagin Set Difference $\\mathbb{{Z}} = \\mathbb{{X}} \\ominus \\mathbb{{E}}$\n"
        f"For state constraint set $\\mathbb{{X}} = [-{number(bound)}, {number(bound)}]$ and error tube $\\mathbb{{E}} = [-{number(error)}, {number(error)}]$:\n"
        f"$$z \\in \\mathbb{{X}} \\ominus \\mathbb{{E}} \\iff |z| \\le X_{{\\max}} - E_{{\\max}}$$\n\n"
        f"### 2. Tightened Bound Calculation\n"
        f"$$z_{{\\max}} = {number(bound)} - {number(error)} = {number(tightened)}$$\n"
        f"The nominal state $z$ must satisfy $|z| \\le {number(tightened)}$, i.e., $z \\in [{number(-tightened)}, {number(tightened)}]$.\n\n"
        f"**Summary:** Tightened interval is [{number(-tightened)}, {number(tightened)}]."
    )
    return record(
        index=index,
        prefix="tube_tightening",
        domain="mpc",
        family="symmetric_tube_mpc_constraint_tightening",
        difficulty="advanced",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "symmetric_tube_tightening",
            "bound": bound,
            "error": error,
            "tightened": tightened,
            "interval": [-tightened, tightened],
            "nonempty": tightened >= 0,
        },
        source_refs=["rawlings_mayne_diehl_mpc_2e"],
        verifier="interval_pontryagin_difference",
    )


def discounted_return(_: np.random.Generator, index: int) -> dict:
    rewards = np.array([1.0 + 0.01 * index, -0.25 + 0.004 * index, 0.6 - 0.003 * index, 0.2 + 0.002 * index])
    gamma = 0.82 + 0.0015 * index
    terms = rewards * gamma ** np.arange(len(rewards))
    total = float(terms.sum())
    prompt = (
        f"{OPENERS[index % 4]} For reward sequence r={rewards.tolist()} and discount factor gamma="
        f"{number(gamma)}, compute the finite discounted return G_0 = sum_(k=0)^3 gamma^k r_k."
    )
    answer = (
        f"### 1. Discounted Return Formula\n"
        f"$$G_0 = r_0 + \\gamma r_1 + \\gamma^2 r_2 + \\gamma^3 r_3$$\n\n"
        f"### 2. Step-by-Step Discounted Terms\n"
        f"- $k=0$: $r_0 = {number(terms[0])}$\n"
        f"- $k=1$: $\\gamma r_1 = ({number(gamma)})({number(rewards[1])}) = {number(terms[1])}$\n"
        f"- $k=2$: $\\gamma^2 r_2 = ({number(gamma**2)})({number(rewards[2])}) = {number(terms[2])}$\n"
        f"- $k=3$: $\\gamma^3 r_3 = ({number(gamma**3)})({number(rewards[3])}) = {number(terms[3])}$\n\n"
        f"### 3. Total Return\n"
        f"$$G_0 = {number(terms[0])} + ({number(terms[1])}) + {number(terms[2])} + {number(terms[3])} = {number(total)}$$\n\n"
        f"**Summary:** G_0 = {number(total)}."
    )
    return record(
        index=index,
        prefix="rl_return",
        domain="reinforcement_learning_control",
        family="finite_discounted_control_return",
        difficulty="foundation",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "finite_discounted_return",
            "rewards": rewards.tolist(),
            "gamma": gamma,
            "terms": terms.tolist(),
            "return": total,
        },
        source_refs=["sutton_barto_reinforcement_learning_2e"],
        verifier="finite_discounted_sum",
    )


def nonlinear_feedforward(_: np.random.Generator, index: int) -> dict:
    a = 0.5 + 0.012 * index
    b = 0.9 + 0.008 * index
    reference = -0.8 + 0.026 * index
    equilibrium_input = a * reference ** 3 / b
    residual = -a * reference ** 3 + b * equilibrium_input
    prompt = (
        f"{OPENERS[index % 4]} For nonlinear system x_dot=-{number(a)}x^3+{number(b)}u, find the "
        f"constant feedforward control input u_eq that makes state x=r={number(reference)} an equilibrium. "
        "Verify the residual; do not claim stability."
    )
    answer = (
        f"### 1. Equilibrium Condition\n"
        f"At equilibrium $x = r = {number(reference)}$, $\\dot{{x}} = 0$:\n"
        f"$$\\dot{{x}} = -{number(a)} r^3 + {number(b)} u_{{eq}} = 0 \\implies u_{{eq}} = \\frac{{{number(a)} r^3}}{{{number(b)}}}$$\n\n"
        f"### 2. Numerical Computation\n"
        f"$$u_{{eq}} = \\frac{{{number(a)} \\times ({number(reference)})^3}}{{{number(b)}}} = \\frac{{{number(a * reference**3)}}}{{{number(b)}}} = {number(equilibrium_input)}$$\n\n"
        f"### 3. Residual Verification\n"
        f"$$\\dot{{x}} = -{number(a)}({number(reference)})^3 + {number(b)}({number(equilibrium_input)}) = {number(residual)}$$\n"
        f"**Summary:** u_eq = {number(equilibrium_input)}."
    )
    return record(
        index=index,
        prefix="nonlinear_feedforward",
        domain="nonlinear_control",
        family="nonlinear_equilibrium_feedforward_input",
        difficulty="intermediate",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "nonlinear_equilibrium_feedforward",
            "a": a,
            "b": b,
            "reference": reference,
            "u_eq": equilibrium_input,
            "residual": residual,
        },
        source_refs=["khalil_nonlinear_systems"],
        verifier="analytic_equilibrium_substitution",
    )


def robust_safe_interval(_: np.random.Generator, index: int) -> dict:
    a = 0.82 + 0.002 * index
    b = 0.65 + 0.004 * index
    state = -1.0 + 0.031 * index
    lower_state = -2.2 - 0.005 * index
    upper_state = 2.0 + 0.006 * index
    disturbance = 0.08 + 0.0015 * index
    nominal = -1.1 + 0.037 * index
    lower_u = (lower_state + disturbance - a * state) / b
    upper_u = (upper_state - disturbance - a * state) / b
    safe_u = float(np.clip(nominal, lower_u, upper_u))
    prompt = (
        f"{OPENERS[index % 4]} For uncertain discrete system x_next={number(a)}x+{number(b)}u+w with bounded disturbance "
        f"|w|<={number(disturbance)}, require x_next in [{number(lower_state)}, {number(upper_state)}] for all possible w. "
        f"At current state x={number(state)}, derive the robust admissible u interval and project nominal command u_nom={number(nominal)} onto it."
    )
    answer = (
        f"### 1. Robust Constraint Inversion under Worst-Case Disturbance\n"
        f"Since $x_{{next}} = a x + b u + w$, the safety condition $x_{{min}} \\le x_{{next}} \\le x_{{max}}$ requires:\n"
        f"- Worst-case lower bound ($w = -w_{{max}}$): $a x + b u - w_{{max}} \\ge x_{{min}} \\implies u \\ge \\frac{{x_{{min}} + w_{{max}} - a x}}{{b}} = {number(lower_u)}$\n"
        f"- Worst-case upper bound ($w = +w_{{max}}$): $a x + b u + w_{{max}} \\le x_{{max}} \\implies u \\le \\frac{{x_{{max}} - w_{{max}} - a x}}{{b}} = {number(upper_u)}$\n\n"
        f"### 2. Admissible Control Interval and Projection\n"
        f"Admissible interval: $u \\in [{number(lower_u)}, {number(upper_u)}]$.\n"
        f"Projecting $u_{{nom}} = {number(nominal)}$ onto this interval yields:\n"
        f"$$u^* = \\text{{clip}}({number(nominal)}, {number(lower_u)}, {number(upper_u)}) = {number(safe_u)}$$\n\n"
        f"**Summary:** Admissible interval = [{number(lower_u)}, {number(upper_u)}], safe control u = {number(safe_u)}."
    )
    return record(
        index=index,
        prefix="robust_safe_interval",
        domain="safety_critical_control",
        family="robust_one_step_safe_input_interval",
        difficulty="advanced",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "robust_safe_input_interval",
            "a": a,
            "b": b,
            "x": state,
            "x_min": lower_state,
            "x_max": upper_state,
            "w_max": disturbance,
            "u_nom": nominal,
            "u_lower": lower_u,
            "u_upper": upper_u,
            "u_safe": safe_u,
        },
        source_refs=["ames_control_barrier_functions", "rawlings_mayne_diehl_mpc_2e"],
        verifier="robust_interval_projection",
    )


def actuator_loss_isolation(_: np.random.Generator, index: int) -> dict:
    effectiveness = np.array([1.0, 0.8 + 0.002 * index, 0.55 + 0.0015 * index])
    command = np.array([0.4 + 0.011 * index, -0.7 + 0.008 * index, 0.9 - 0.006 * index])
    loss_factor = 0.35 + 0.004 * (index % 20)
    failed = (index - 1) % 3
    candidate_outputs = []
    for candidate in range(3):
        candidate_effectiveness = effectiveness.copy()
        candidate_effectiveness[candidate] *= loss_factor
        candidate_outputs.append(float(candidate_effectiveness @ command))
    measured = candidate_outputs[failed]
    residuals = np.abs(np.asarray(candidate_outputs) - measured)
    identified = int(np.argmin(residuals))
    
    prompt = (
        f"{OPENERS[index % 4]} Nominal actuator effectiveness is B={effectiveness.tolist()} "
        f"and commanded input is u={command.tolist()}. Exactly one actuator retains effectiveness factor "
        f"lambda={number(loss_factor)} while the others remain nominal. The measured virtual control is tau={number(measured)}. "
        "Compute each single-fault hypothesis prediction/residual and isolate the failed actuator."
    )
    answer = (
        f"### 1. Single-Fault Hypothesis Testing\n"
        f"For each actuator $i \\in \\{{1, 2, 3\\}}$, compute predicted virtual control $\\tau_i = B^{{(i)}} u$ with $B^{{(i)}}_i = \\lambda B_i$:\n"
        f"- Hypothesis 1 (Actuator 1 degraded): $\\tau_1 = {number(candidate_outputs[0])}$, residual $r_1 = |\\tau_1 - \\tau| = {number(residuals[0])}$\n"
        f"- Hypothesis 2 (Actuator 2 degraded): $\\tau_2 = {number(candidate_outputs[1])}$, residual $r_2 = |\\tau_2 - \\tau| = {number(residuals[1])}$\n"
        f"- Hypothesis 3 (Actuator 3 degraded): $\\tau_3 = {number(candidate_outputs[2])}$, residual $r_3 = |\\tau_3 - \\tau| = {number(residuals[2])}$\n\n"
        f"### 2. Fault Isolation Decision\n"
        f"The minimum residual is $r_{{{identified + 1}}} = {number(residuals[identified])} \\approx 0$.\n\n"
        f"**Summary:** Actuator {identified + 1} is isolated as the degraded unit."
    )
    return record(
        index=index,
        prefix="actuator_loss",
        domain="fault_tolerant_control",
        family="single_actuator_loss_residual_isolation",
        difficulty="advanced",
        prompt=prompt,
        answer=answer,
        ground_truth={
            "kind": "single_actuator_loss_isolation",
            "B": effectiveness.tolist(),
            "u": command.tolist(),
            "loss_factor": loss_factor,
            "measured": measured,
            "candidate_outputs": candidate_outputs,
            "residuals": residuals.tolist(),
            "failed_index": failed,
            "identified_index": identified,
        },
        source_refs=["fault_tolerant_control_allocation_literature"],
        verifier="single_fault_hypothesis_residuals",
    )


FAMILIES: tuple[tuple[str, Callable[[np.random.Generator, int], dict]], ...] = (
    ("scalar_affine_nonlinear_dynamic_inversion", nonlinear_dynamic_inversion),
    ("linear_gain_schedule_interpolation", gain_schedule),
    ("scalar_control_barrier_safety_filter", cbf_filter),
    ("scalar_stationary_process_covariance", scalar_stationary_covariance),
    ("minimum_norm_redundant_control_allocation", fault_allocation),
    ("iid_scalar_jump_mean_square_test", jump_scalar),
    ("relative_state_error_event_trigger", event_trigger),
    ("symmetric_tube_mpc_constraint_tightening", tube_tightening),
    ("finite_discounted_control_return", discounted_return),
    ("nonlinear_equilibrium_feedforward_input", nonlinear_feedforward),
    ("robust_one_step_safe_input_interval", robust_safe_interval),
    ("single_actuator_loss_residual_isolation", actuator_loss_isolation),
)


def generate_extended_v1(count_per_family: int, seed: int) -> list[dict]:
    rng = np.random.default_rng(seed)
    return [
        generator(rng, index)
        for _, generator in FAMILIES
        for index in range(1, count_per_family + 1)
    ]