//// Cross-surface operational autopilot and SLO-aware controls (M54). import gleam/int import gleam/list import lightspeed/ops/slo_autopilot import lightspeed/ops/telemetry import lightspeed/pipeline/slo as pipeline_slo import lightspeed/tenant/policy as tenant_policy /// Runtime surface controlled by the operational autopilot. pub type Surface { RuntimeSurface TransportSurface TenantSurface EtlSurface } /// Bounded automated control action. pub type Control { ShedRuntimeEvents(percent: Int) ThrottleTransport(max_events_per_sec: Int) IsolateTenant(tenant_id: String) PauseEtlPipelines } /// Operator override and automation policy. pub type OverridePolicy { OverridePolicy( name: String, allow_runtime: Bool, allow_transport: Bool, allow_tenant: Bool, allow_etl: Bool, require_operator_override: Bool, max_actions_per_incident: Int, ) } /// Autopilot decision for one control action. pub type Decision { Applied RequiresOverride(override_id: String) BlockedByPolicy(reason: String) RolledBack } /// Deterministic M54 drill scenario. pub type DrillScenario { RuntimeLatencySpike TransportAckStorm TenantHotspotIsolation EtlReplayLag } /// Auditable control action evidence. pub type ControlAction { ControlAction( surface: Surface, control: Control, decision: Decision, reason: String, detected_at_ms: Int, applied_at_ms: Int, recovered_at_ms: Int, reversible: Bool, ) } /// Cross-surface incident correlation evidence. pub type Correlation { Correlation( incident_id: String, surface: Surface, trace_ref: String, detected_at_ms: Int, mitigated_at_ms: Int, recovered_at_ms: Int, ) } /// Deterministic autopilot drill result. pub type DrillResult { DrillResult( scenario: DrillScenario, surface: Surface, status: slo_autopilot.Status, baseline_detection_ms: Int, autopilot_detection_ms: Int, baseline_mitigation_ms: Int, autopilot_mitigation_ms: Int, actions: List(ControlAction), correlation: Correlation, ) } /// Default policy allows bounded automation on every M54 surface. pub fn default_policy() -> OverridePolicy { OverridePolicy( name: "default", allow_runtime: True, allow_transport: True, allow_tenant: True, allow_etl: True, require_operator_override: False, max_actions_per_incident: 2, ) } /// Strict policy requires explicit override semantics for high-impact controls. pub fn strict_policy() -> OverridePolicy { OverridePolicy( name: "strict", allow_runtime: True, allow_transport: True, allow_tenant: True, allow_etl: True, require_operator_override: True, max_actions_per_incident: 2, ) } /// Blocked policy used by tests and operators to verify policy bounds. pub fn blocked_policy() -> OverridePolicy { OverridePolicy( name: "blocked", allow_runtime: True, allow_transport: True, allow_tenant: False, allow_etl: False, require_operator_override: True, max_actions_per_incident: 1, ) } /// Run one deterministic M54 drill. pub fn run_drill( scenario: DrillScenario, policy: OverridePolicy, ) -> DrillResult { case scenario { RuntimeLatencySpike -> run_runtime_latency_spike(policy) TransportAckStorm -> run_transport_ack_storm(policy) TenantHotspotIsolation -> run_tenant_hotspot_isolation(policy) EtlReplayLag -> run_etl_replay_lag(policy) } } /// Run all default M54 drills. pub fn run_default_drills() -> List(DrillResult) { let policy = default_policy() [ run_drill(RuntimeLatencySpike, policy), run_drill(TransportAckStorm, policy), run_drill(TenantHotspotIsolation, policy), run_drill(EtlReplayLag, policy), ] } /// Drill scenario label. pub fn drill_scenario_label(scenario: DrillScenario) -> String { case scenario { RuntimeLatencySpike -> "runtime_latency_spike" TransportAckStorm -> "transport_ack_storm" TenantHotspotIsolation -> "tenant_hotspot_isolation" EtlReplayLag -> "etl_replay_lag" } } /// Surface label. pub fn surface_label(surface: Surface) -> String { case surface { RuntimeSurface -> "runtime" TransportSurface -> "transport" TenantSurface -> "tenant" EtlSurface -> "etl" } } /// Control label. pub fn control_label(control: Control) -> String { case control { ShedRuntimeEvents(percent) -> "shed_runtime_events:" <> int.to_string(percent) ThrottleTransport(max_events_per_sec) -> "throttle_transport:" <> int.to_string(max_events_per_sec) IsolateTenant(tenant_id) -> "isolate_tenant:" <> tenant_id PauseEtlPipelines -> "pause_etl_pipelines" } } /// Decision label. pub fn decision_label(decision: Decision) -> String { case decision { Applied -> "applied" RequiresOverride(override_id) -> "requires_override:" <> override_id BlockedByPolicy(reason) -> "blocked_by_policy:" <> reason RolledBack -> "rolled_back" } } /// Override policy label. pub fn policy_label(policy: OverridePolicy) -> String { "policy=" <> policy.name <> ":runtime=" <> bool_label(policy.allow_runtime) <> ":transport=" <> bool_label(policy.allow_transport) <> ":tenant=" <> bool_label(policy.allow_tenant) <> ":etl=" <> bool_label(policy.allow_etl) <> ":override=" <> bool_label(policy.require_operator_override) <> ":max_actions=" <> int.to_string(policy.max_actions_per_incident) } /// Control action label. pub fn action_label(action: ControlAction) -> String { "surface=" <> surface_label(action.surface) <> ":control=" <> control_label(action.control) <> ":decision=" <> decision_label(action.decision) <> ":reason=" <> action.reason <> ":detected=" <> int.to_string(action.detected_at_ms) <> ":applied=" <> int.to_string(action.applied_at_ms) <> ":recovered=" <> int.to_string(action.recovered_at_ms) <> ":reversible=" <> bool_label(action.reversible) } /// Correlation label for incident triage. pub fn correlation_label(correlation: Correlation) -> String { "incident=" <> correlation.incident_id <> ":surface=" <> surface_label(correlation.surface) <> ":trace=" <> correlation.trace_ref <> ":detected=" <> int.to_string(correlation.detected_at_ms) <> ":mitigated=" <> int.to_string(correlation.mitigated_at_ms) <> ":recovered=" <> int.to_string(correlation.recovered_at_ms) } /// Stable drill result signature. pub fn drill_result_signature(result: DrillResult) -> String { "drill=" <> drill_scenario_label(result.scenario) <> "|surface=" <> surface_label(result.surface) <> "|status=" <> slo_autopilot.status_label(result.status) <> "|latency_reduced=" <> bool_label(latency_reduced(result)) <> "|baseline_detection_ms=" <> int.to_string(result.baseline_detection_ms) <> "|autopilot_detection_ms=" <> int.to_string(result.autopilot_detection_ms) <> "|baseline_mitigation_ms=" <> int.to_string(result.baseline_mitigation_ms) <> "|autopilot_mitigation_ms=" <> int.to_string(result.autopilot_mitigation_ms) <> "|actions=" <> join_with(",", list.map(result.actions, action_label)) <> "|" <> correlation_label(result.correlation) } /// Drill actions accessor. pub fn actions(result: DrillResult) -> List(ControlAction) { result.actions } /// Drill correlation accessor. pub fn correlation(result: DrillResult) -> Correlation { result.correlation } /// Count actions in a drill result. pub fn action_count(result: DrillResult) -> Int { list.length(result.actions) } /// True when autopilot detection and mitigation beat the manual baseline. pub fn latency_reduced(result: DrillResult) -> Bool { result.autopilot_detection_ms < result.baseline_detection_ms && result.autopilot_mitigation_ms < result.baseline_mitigation_ms } /// Recovery latency measured from detection to recovery. pub fn recovery_latency_ms(result: DrillResult) -> Int { result.correlation.recovered_at_ms - result.correlation.detected_at_ms } /// True when actions are bounded by policy and not silently unbounded. pub fn actions_bounded(result: DrillResult, policy: OverridePolicy) -> Bool { action_count(result) <= policy.max_actions_per_incident } /// True when every emitted action is reversible. pub fn actions_reversible(result: DrillResult) -> Bool { list.all(result.actions, fn(action) { action.reversible }) } /// True when every action is auditable and bounded by explicit policy outcomes. pub fn actions_auditable(result: DrillResult) -> Bool { list.all(result.actions, fn(action) { action.reason != "" && decision_label(action.decision) != "" }) } /// True when a drill has acceptable certification evidence. pub fn drill_result_valid(result: DrillResult, policy: OverridePolicy) -> Bool { latency_reduced(result) && actions_bounded(result, policy) && actions_reversible(result) && actions_auditable(result) && recovery_latency_ms(result) > 0 } /// True when every drill passes the same certification criteria. pub fn all_drills_valid( results: List(DrillResult), policy: OverridePolicy, ) -> Bool { list.all(results, fn(result) { drill_result_valid(result, policy) }) } /// True when a control action requested explicit operator override. pub fn requires_override(action: ControlAction) -> Bool { case action.decision { RequiresOverride(_) -> True Applied | BlockedByPolicy(_) | RolledBack -> False } } /// True when a control action was blocked by policy. pub fn blocked_by_policy(action: ControlAction) -> Bool { case action.decision { BlockedByPolicy(_) -> True Applied | RequiresOverride(_) | RolledBack -> False } } /// Telemetry metric for one autopilot action. pub fn audit_metric(action: ControlAction) -> telemetry.Metric { telemetry.Counter( name: "lightspeed.operational_autopilot.action_total", value: 1, tags: [ telemetry.Tag("surface", surface_label(action.surface)), telemetry.Tag("control", control_label(action.control)), telemetry.Tag("decision", decision_label(action.decision)), telemetry.Tag("reason", action.reason), ], ) } fn run_runtime_latency_spike(policy: OverridePolicy) -> DrillResult { let sample = slo_autopilot.sample("runtime", "availability", 2500, 1700, "") let status = slo_autopilot.classify(sample, slo_autopilot.default_thresholds()) let action = control_action( RuntimeSurface, ShedRuntimeEvents(40), decide(policy, RuntimeSurface), "slo_burn_rate:runtime_availability", 15_000, 42_000, 72_000, ) drill_result( RuntimeLatencySpike, RuntimeSurface, status, 120_000, 15_000, 180_000, 42_000, [action], "m54-runtime-latency", "runtime:slo:burn-rate", ) } fn run_transport_ack_storm(policy: OverridePolicy) -> DrillResult { let sample = slo_autopilot.sample("transport", "ack_latency", 2200, 1600, "") let status = slo_autopilot.classify(sample, slo_autopilot.default_thresholds()) let action = control_action( TransportSurface, ThrottleTransport(220), decide(policy, TransportSurface), "transport_ack_storm:cadence_guard", 12_000, 34_000, 68_000, ) drill_result( TransportAckStorm, TransportSurface, status, 100_000, 12_000, 150_000, 34_000, [action], "m54-transport-ack", "transport:ack:storm", ) } fn run_tenant_hotspot_isolation(policy: OverridePolicy) -> DrillResult { let sample = slo_autopilot.sample("tenant", "hotspot_budget", 2600, 1900, "tenant-a") let status = slo_autopilot.classify(sample, slo_autopilot.default_thresholds()) let policy_reason = tenant_policy_reason() let action = control_action( TenantSurface, IsolateTenant("tenant-a"), decide(policy, TenantSurface), policy_reason, 18_000, 49_000, 84_000, ) drill_result( TenantHotspotIsolation, TenantSurface, status, 130_000, 18_000, 190_000, 49_000, [action], "m54-tenant-hotspot", "tenant:policy:isolation", ) } fn run_etl_replay_lag(policy: OverridePolicy) -> DrillResult { let observation = pipeline_slo.observation("orders_replay", 380, 180, 11, 100, 5200) let budget = pipeline_slo.budget("orders_replay", 320, 220, 8, 4500) let results = pipeline_slo.evaluate([observation], [budget]) let result = case results { [result, ..] -> result [] -> pipeline_slo.BudgetResult("orders_replay", False, "missing_budget") } let status = case result.passed { True -> slo_autopilot.Healthy False -> slo_autopilot.Critical } let action = control_action( EtlSurface, PauseEtlPipelines, decide(policy, EtlSurface), "etl_slo:" <> pipeline_slo.budget_result_label(result), 22_000, 58_000, 96_000, ) drill_result( EtlReplayLag, EtlSurface, status, 150_000, 22_000, 210_000, 58_000, [action], "m54-etl-replay", "etl:replay:lag", ) } fn tenant_policy_reason() -> String { let context = tenant_policy.tenant_context( "operator-a", "tenant-a", tenant_policy.TenantAdmin, ) let runtime = tenant_policy.new( context, tenant_policy.expanded_budget(10, 10, 10, 10, 10, 10), ) let #(_, outcome) = tenant_policy.evaluate( runtime, tenant_policy.ApplyMitigation(tenant_policy.IsolateTenant, 1), ) "tenant_policy:" <> tenant_policy.outcome_label(outcome) } fn decide(policy: OverridePolicy, surface: Surface) -> Decision { let allowed = case surface { RuntimeSurface -> policy.allow_runtime TransportSurface -> policy.allow_transport TenantSurface -> policy.allow_tenant EtlSurface -> policy.allow_etl } case allowed { False -> BlockedByPolicy("surface_disabled:" <> surface_label(surface)) True -> { case policy.require_operator_override && high_impact_surface(surface) { True -> RequiresOverride( "override:" <> policy.name <> ":" <> surface_label(surface), ) False -> Applied } } } } fn high_impact_surface(surface: Surface) -> Bool { case surface { RuntimeSurface | TransportSurface -> False TenantSurface | EtlSurface -> True } } fn control_action( surface: Surface, control: Control, decision: Decision, reason: String, detected_at_ms: Int, applied_at_ms: Int, recovered_at_ms: Int, ) -> ControlAction { ControlAction( surface: surface, control: control, decision: decision, reason: reason, detected_at_ms: detected_at_ms, applied_at_ms: applied_at_ms, recovered_at_ms: recovered_at_ms, reversible: True, ) } fn drill_result( scenario: DrillScenario, surface: Surface, status: slo_autopilot.Status, baseline_detection_ms: Int, autopilot_detection_ms: Int, baseline_mitigation_ms: Int, autopilot_mitigation_ms: Int, actions: List(ControlAction), incident_id: String, trace_ref: String, ) -> DrillResult { let correlation = Correlation( incident_id: incident_id, surface: surface, trace_ref: trace_ref, detected_at_ms: autopilot_detection_ms, mitigated_at_ms: autopilot_mitigation_ms, recovered_at_ms: first_recovery(actions), ) DrillResult( scenario: scenario, surface: surface, status: status, baseline_detection_ms: baseline_detection_ms, autopilot_detection_ms: autopilot_detection_ms, baseline_mitigation_ms: baseline_mitigation_ms, autopilot_mitigation_ms: autopilot_mitigation_ms, actions: actions, correlation: correlation, ) } fn first_recovery(actions: List(ControlAction)) -> Int { case actions { [action, ..] -> action.recovered_at_ms [] -> 0 } } fn bool_label(value: Bool) -> String { case value { True -> "true" False -> "false" } } fn join_with(separator: String, values: List(String)) -> String { case values { [] -> "" [first, ..rest] -> list.fold(rest, first, fn(accumulator, value) { accumulator <> separator <> value }) } }