Supervisor

Supervisor

The mission orchestration layer: assembles a validated MissionPlan from the configuration, runs the emitter and the receiver as supervised tasks (abort / continue / restart policies, heartbeat watchdog, the single-writer component_events.csv), keeps the lifecycle sentinels consistent on every exit path, and drives the failure-isolated post-processing stages. The headless entry point scripts/run_full_sim.jl is argument parsing plus one call to run_mission.

CleanFileLogger(path::String, rotate_bytes::Int)

Component log sink: one line per record — [Level] message, with structured keyword values on indented lines and exceptions rendered compactly — without the └ @ Module file:line source-location suffix, which is noise when the file is followed in a dashboard terminal. ANSI escape sequences are stripped so the file reads cleanly after the run. Records are appended per write (log rates are a few lines per batch), which allows size-capped rotation to name#k.log (retention.log_rotate_mb) without a held-open stream across the rotation boundary. Formatting never throws: a throwing logger would terminate the task it logs for.

strip_ansi(text::AbstractString) -> String

Removes ANSI escape sequences (colors, cursor motion) from a log message.

render_log_value(v) -> String

Text of one structured log value for CleanFileLogger: exceptions (also the exception of an (exception, backtrace) tuple) through showerror, everything else through string; a value whose rendering throws yields a <unprintable T> placeholder, so the logger itself never throws.

RESTART_SEED_OFFSET

Offset of the RNG seed of a restarted emitter: attempt k ≥ 1 seeds its fresh instrument with rng_seed + RESTART_SEED_OFFSET + k, so the noise realization after a restart is reproducible from simulation.rng_seed yet distinct from the primary stream (rng_seed) and from the loss channel (rng_seed + 1).

MissionPlan

Everything derived from a validated configuration before the mission clock starts: the run ID, the [simulation] scalars, the typed [telemetry], [physics], and [supervision] settings, the composite link model, the loss channel (seeded with rng_seed + 1), the retry limit, and the retention policy. Built by mission_plan; nothing on disk depends on it until run_mission creates the run directory.

mission_plan(cfg::Dict{String,Any}; run_id = "") -> MissionPlan

Validates cfg (safe intervals, then the storage budget), builds the channel models — the physics stream is seeded with simulation.rng_seed, the loss channel with rng_seed + 1, so both are independently reproducible — stamps external-input provenance when physics.data_source = "external", and fixes the run ID (generated when empty). Writes nothing to disk and leaves cfg unmodified: the plan carries a shallow copy of it, which is what receives the provenance stamp and becomes the run's configuration snapshot.

stamp_external_provenance!(cfg, physics, needed_days)

External-input coverage report and provenance stamp: row count and SHA-256 of the input file are computed up front — before any directory exists — so exhaustion is predicted at startup rather than discovered mid-mission, and the run snapshot pins the exact input consumed (no sidecar or generator metadata is assumed). Writes cfg["provenance"].

record_generation_gap!(run_dir, clock)

Bounds an emitter outage in events_tx.csv: a gap_start row at the last recorded generation instant and a gap_end row at the current mission time (Batch = STREAM). Called before a replacement emitter is spawned — no live events_tx writer exists at that instant.

log_component_event!(run_dir, clock, component, event)

Appends one row to the single-writer component_events.csv (SimTime,Component,Event; events down, restart, stalled, recovered).

supervise!(spawners, run_dir, clock, stop_flag, heartbeats, policy;
           on_restart = (name, attempt) -> nothing,
           poll_sec = TelemetryCore.RECEIVER_POLL_INTERVAL_SEC) -> Dict{Symbol,Int}

Runs the components until all of them have finished. spawners[name](attempt) returns the component's Task (attempt 0 is the primary launch, attempt ≥ 1 a supervised restart). On a failed task the policy (policy.on_component_failure) decides: "abort" raises stop_flag so the partner stops cooperatively, "continue" leaves the partner running one-sided, "restart" relaunches up to policy.max_restarts times after calling on_restart(name, attempt). A component whose heartbeat file in heartbeats stays untouched for longer than policy.watchdog_sec is recorded as stalled (and recovered when it resumes); the watchdog only records, it never intervenes. Every lifecycle transition is appended to component_events.csv, and failures, policy decisions, restarts, and watchdog trips are reported as [SUPERVISOR] log records through the logger active in the calling task (the global logger under run_mission; the component tasks log to their own files). Returns the restart count per component.

component_spawners(plan, run_dir, clock, deadline, stop_flag, heartbeats,
                   instrument, pending_segments, emitter_logger,
                   receiver_logger, orig_stdout) -> Dict{Symbol,Function}

The emitter and receiver launchers consumed by supervise!. Attempt 0 continues the pre-populated instrument and partial batch; a restarted emitter (attempt ≥ 1) takes a fresh instrument anchored at the current mission time — a genuine generation gap with a new noise realization on the seed rng_seed + RESTART_SEED_OFFSET + attempt — and no carried-over partial batch.

post_process!(plan, run_dir; orig_stdout = stdout)

The derived products after both components have finished: the 2D batch-state timeline (post_processing.generate_mask_timeline), the alert-latency metric (alert_latency, look-back alert_lookback_hours), the delivery-delay metric (delivery_delay, requirement delivery_requirement_hours), the point-wise 0/1 expansions (expand_to_pointwise_masks, rows from target_event_rows), the HDF5 product export (hdf5_export), and the publication figure export ([post_processing.publication]). Each stage is failure-isolated: the simulation data is already on disk, so an error is reported loudly but never aborts the remaining stages (every product can be regenerated by the standalone post-processing scripts).

expand_pointwise_masks!(plan::MissionPlan, run_dir::String, orig_stdout::IO)

The point-wise expansion stage of post_process!: every row of post_processing.target_event_rows ("all", integers, -1 = last, range strings) expanded by Receiver.expand_pointwise_mask, each row failure-isolated.

contact_summary(plan::MissionPlan) -> String

One-line description of the contact schedule for the mission banner: the daily window with its seasonal extension and exception count, or the explicit pass count, plus the low-latency periods and whether they are enabled.

print_banner(io::IO, plan::MissionPlan, run_dir::String)

Writes the mission-start banner to io: run ID, mission span and speed-up, link capacity and the daily capacity balance, the contact schedule (contact_summary), markers, generation gaps, the on-board recorder ceiling, and the log location.

warm_up_components!(plan, orig_stdout)

Compiles both component loops before the mission clock starts: each is entered once with a deadline already in the past against a scratch run directory (<run_id>__warmup, removed afterwards), with the same argument types the mission uses. Without this, the first-call compilation of the loops — several wall-clock seconds in a fresh process — would elapse as mission time after the anchor (hours at high speed_up; a short mission could expire before either component ran). Returns the warm-up duration [s].

execute_mission!(plan, run_dir, orig_stdout)

The mission proper inside an existing run directory: fresh component logs, sequential pre-population of the onboard buffer, the component warm-up (warm_up_components!), the mission clock with its persisted anchor and absolute deadline, the supervised component tasks, and the post-processing stages. Called by run_mission, which owns the lifecycle sentinels.

run_mission(cfg::Dict{String,Any}; run_id = "", orig_stdout = stdout) -> String

The headless pipeline end to end: single-thread advisory, mission_plan (validation, storage gate, models, provenance), the run directory with its configuration snapshot, the lifecycle sentinels — RUN_ACTIVE while the pipeline may still write, then RUN_COMPLETE at lifecycle end (not success: a failed component still reaches it after the failure-isolated post-processing) or RUN_ABORTED on any escaping exception — and execute_mission!. Returns the run directory.