intervention

Scheduling and monitoring. Answers: "When should humans step in?" and "What kind of problem is this?"

Paper reference: Section 1 (Autonomy time, Prop. 3); Section 4 (Demo 3, diagnostic differential; Algorithm 1, Step 6)

compute_autonomy_time

`minimal_oversight.intervention.compute_autonomy_time(c_op, p_min, mu_eff, governance_gap=0.02, process_entropy=0.0)`

Expected time before human intervention is needed.

T*_auto = B_eff / μ_eff

Ref: Equation 17.

\(T_\text{auto}^* = B_\text{eff} / \mu_\text{eff}\) — the expected time before quality drops below \(p_\text{min}\).

compute_pipeline_intervention_schedule

`minimal_oversight.intervention.compute_pipeline_intervention_schedule(pipeline, node_capacities, p_min=0.8, governance_gap=0.02, process_entropy=0.0)`

Compute intervention schedule across all nodes.

Nodes with short T_auto need frequent review. Nodes with long T_auto can operate autonomously.

Returns per-node intervention timing, sorted by urgency (shortest \(T_\text{auto}^*\) first).

from minimal_oversight.intervention import compute_pipeline_intervention_schedule

schedule = compute_pipeline_intervention_schedule(pipeline, capacities, p_min=0.80)
for s in schedule:
    print(f"{s.node_name}: review every {s.t_auto:.0f} steps (rank {s.priority_rank})")

diagnose_failure_mode

`minimal_oversight.intervention.diagnose_failure_mode(delta_sigma_raw, delta_m_star, threshold=0.01)`

Classify the failure mode from the diagnostic differential.

Based on joint movement of (Δσ_raw, ΔM*): stable/stable → healthy rising/falling → agent improving falling/rising → masking degradation (URGENT) falling/falling → correlated drift stable/rising → corrector coasting

Ref: Section 4, Demonstration 3.

Classifies the failure mode from \((\Delta\sigma_\text{raw}, \Delta M^*)\):

\(\Delta\sigma_\text{raw}\)	\(\Delta M^*\)	Mode	Action
Stable	Stable	`HEALTHY`	None
Rising	Falling	`AGENT_IMPROVING`	Reduce corrector budget
Falling	Rising	`MASKING_DEGRADATION`	Retrain agent
Falling	Falling	`CORRELATED_DRIFT`	Retrain both
Stable	Rising	`CORRECTOR_COASTING`	Check agent

from minimal_oversight.intervention import diagnose_failure_mode

mode = diagnose_failure_mode(delta_sigma_raw=-0.04, delta_m_star=0.05)
print(mode)  # FailureMode.MASKING_DEGRADATION

check_alerts

`minimal_oversight.intervention.check_alerts(pipeline, node_capacities, p_min=0.8, masking_threshold=1.5, buffer_threshold=0.05, governance_gap=0.02, process_entropy=0.0)`

Check all monitoring conditions and generate alerts.

Three alert categories

Masking rising (M* > threshold)
Buffer collapsing (B_eff near zero)
K/N approaching capacity threshold

Three alert categories: masking (\(M^* > \tau\)), buffer (\(B_\text{eff} < 0.05\)), capacity (\(K/N < \text{threshold}\)). Returns alerts sorted by severity.

explain_failure_surface

`minimal_oversight.intervention.explain_failure_surface(pipeline, node_capacities, p_min=0.8, governance_gap=0.02, process_entropy=0.0)`

Human-readable explanation of the pipeline's failure surface.

Covers all five failure modes

Infeasible target
Masking-driven false confidence
Process overload
Conditional fragility
Upstream bottleneck

Human-readable analysis covering five failure modes: infeasible target, masking-driven false confidence, process overload, conditional fragility, upstream bottleneck.

Data classes

AlertLevel (Enum)

INFO, WARNING, CRITICAL

FailureMode (Enum)

HEALTHY, AGENT_IMPROVING, MASKING_DEGRADATION, CORRELATED_DRIFT, CORRECTOR_COASTING

InterventionSchedule

Field	Type	Description
`node_name`	`str`	Node
`t_auto`	`float`	\(T_\text{auto}^*\)
`intervention_frequency`	`float`	\(1/T_\text{auto}^*\)
`review_cost`	`float`	Cost weight (uniform in v1)
`priority_rank`	`int`	1 = most urgent

MonitoringAlert

Field	Type	Description
`level`	`AlertLevel`	Severity
`node_name`	`str`	Which node
`failure_mode`	`FailureMode`	Classified mode
`message`	`str`	What's wrong
`recommended_action`	`str`	What to do