lightspeed

//// 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
      })
  }
}