diff --git a/services/flysim/Cargo.lock b/services/flysim/Cargo.lock index 322d886..25f88d6 100644 --- a/services/flysim/Cargo.lock +++ b/services/flysim/Cargo.lock @@ -508,6 +508,7 @@ dependencies = [ "jsonschema", "serde", "serde_json", + "sha2", "tempfile", "tokio", "tokio-tungstenite", diff --git a/services/flysim/crates/flysim/Cargo.toml b/services/flysim/crates/flysim/Cargo.toml index 9f6c3db..b3184ba 100644 --- a/services/flysim/crates/flysim/Cargo.toml +++ b/services/flysim/crates/flysim/Cargo.toml @@ -37,6 +37,7 @@ axum = { version = "0.8", features = ["ws"] } clap = { version = "4.5", features = ["derive"] } serde = { workspace = true } serde_json = { workspace = true } +sha2 = { workspace = true } tokio = { version = "1", features = [ "rt-multi-thread", "net", diff --git a/services/flysim/crates/flysim/src/lib.rs b/services/flysim/crates/flysim/src/lib.rs index fe47337..56b7523 100644 --- a/services/flysim/crates/flysim/src/lib.rs +++ b/services/flysim/crates/flysim/src/lib.rs @@ -37,6 +37,7 @@ pub mod sdnotify; pub mod simloop; pub mod snapshot; pub mod store; +pub mod trace; use std::sync::Arc; diff --git a/services/flysim/crates/flysim/src/simloop.rs b/services/flysim/crates/flysim/src/simloop.rs index 0b24bcb..36c9c55 100644 --- a/services/flysim/crates/flysim/src/simloop.rs +++ b/services/flysim/crates/flysim/src/simloop.rs @@ -56,6 +56,7 @@ use crate::snapshot::{ RewardCounts, RewardKind, Snapshot, f32_bytes, finite, spike_bitset, }; use crate::store::{self, RuntimeState, Store}; +use crate::trace::FrameTrace; /// Commands the control API queues for the sim thread. #[derive(Debug)] @@ -397,6 +398,8 @@ pub struct Sim { restored: bool, /// Per-phase timing, off unless `FLY_PROFILE_SECONDS` is set (`crate::profile`). profiler: Profiler, + /// The per-frame trace, off unless `FLY_TRACE` names a file (`crate::trace`). + trace: Option, } /// The checkpoint compatibility string, from the pieces that make it up. @@ -603,6 +606,8 @@ impl Sim { semantic_rewards, restored: false, profiler: Profiler::from_env(now), + trace: FrameTrace::from_env() + .with_context(|| format!("creating the {} file", crate::trace::ENV))?, agent, emulator, adapter, @@ -929,11 +934,17 @@ impl Sim { /// One frame, in the prototype's order. fn step_frame(&mut self) -> Result<()> { + if let Some(trace) = self.trace.as_mut() { + trace.begin(self.frame_counter, self.agent.network.ms); + } // 2. Step the brain: 16 or 17 integer ticks, the remainder carried and checkpointed. self.remainder += self.agent.ms_per_frame; let steps = self.remainder.floor(); self.remainder -= steps; self.agent.network.step(steps as u64); + if let Some(trace) = self.trace.as_mut() { + trace.ticked(steps as u64, self.remainder, &self.agent.network); + } self.profiler.lap(Phase::Step); if self.profiler.enabled() { self.profiler.absorb_brain(self.agent.network.timings()); @@ -969,6 +980,9 @@ impl Sim { blocked.as_deref(), bound.as_deref(), ); + if let Some(trace) = self.trace.as_mut() { + trace.decided(&active); + } // A new winner starts its own window: it has not had a hold to move in yet. let held = self.agent.decoder.current().map(str::to_string); if held != self.held_channel { @@ -1001,6 +1015,9 @@ impl Sim { } None => Vec::new(), }; + if let Some(trace) = self.trace.as_mut() { + trace.executed(self.buttons, &started_or_finished); + } self.emit_macro_events(&started_or_finished); self.emulator.set_buttons(self.buttons as u8); self.profiler.lap(Phase::Decode); @@ -1017,6 +1034,9 @@ impl Sim { self.agent .network .set_visual_frame(&self.frame_buffer, width, height); + if let Some(trace) = self.trace.as_mut() { + trace.advanced(&self.frame_buffer, &self.emulator); + } let raw = self.emulator.take_audio_u8(); self.dc_blocker.process_into(&raw, &mut self.pending_audio); self.profiler.lap(Phase::Retina); @@ -1074,6 +1094,9 @@ impl Sim { // 10. Ratchet: observe, and recover if it says so. let progress = self.adapter.progress(); + if let Some(trace) = self.trace.as_mut() { + trace.evaluated(&events, &abandoned, progress.rank); + } self.track_rank(&progress, ms); let safe = self.adapter.safe_for_snapshot(); let capture_due = safe && u64::from(progress.rank) > self.ratchet.state.best; @@ -1095,6 +1118,7 @@ impl Sim { // (`docs/design/macros.md` section 12.15); in raw mode there is no layer and no // objective, and the answer is false. let nearer = self.macros.as_ref().is_some_and(MacroLayer::nearer_the_objective); + let trace = &mut self.trace; let recover = self.ratchet.observe_with_progress( safe, u64::from(progress.rank), @@ -1102,7 +1126,14 @@ impl Sim { ms as u64, self.adapter.game_over(), nearer, - || captured.expect("the ratchet only captures when a snapshot was prepared"), + || { + let snapshot = + captured.expect("the ratchet only captures when a snapshot was prepared"); + if let Some(trace) = trace.as_mut() { + trace.slot_saved(&snapshot.game); + } + snapshot + }, ); if recover { // Two triggers, two stories on the ticker: a game over ended the run, a stall did not. @@ -1197,6 +1228,9 @@ impl Sim { } None => Vec::new(), }; + if let Some(trace) = self.trace.as_mut() { + trace.rolled_back(&abandoned); + } self.emit_macro_events(&abandoned); self.pending_recovery = true; if let Err(error) = self.checkpoint(true, None) { @@ -1292,6 +1326,9 @@ impl Sim { .clamp(1.0, self.shared.config.control.sugar_max_ms) .min(self.shared.config.control.sugar_max_ms); self.agent.network.stimulate(duration); + if let Some(trace) = self.trace.as_mut() { + trace.sugar(duration); + } let day = utc_day(now_ms); if day != self.sugar_day { @@ -1313,6 +1350,9 @@ impl Sim { fn reward(&mut self, value: f64, by: &str, source: &str) -> u64 { let ms = self.agent.network.ms; self.agent.network.plasticity.reinforce(value, ms); + if let Some(trace) = self.trace.as_mut() { + trace.reward_pulse(value); + } tracing::info!(by, source, value, "reward pulse applied"); self.emit( NewEvent::new(FeedEventKind::Reward, format!("{by} sent a reward pulse ({value})")) @@ -1390,6 +1430,9 @@ impl Sim { if let Err(error) = self.checkpoint_blocking(true, None) { tracing::error!(%error, "the final checkpoint failed"); } + if let Some(trace) = self.trace.as_mut() { + trace.finish(); + } if let Err(error) = self.log.flush() { tracing::error!(%error, "the final event log flush failed"); } @@ -1499,6 +1542,9 @@ impl Sim { durable && self.best_archived_rank.is_none_or(|best| *rank > best) }); let (generation, agent, runtime) = self.snapshot_state()?; + if let Some(trace) = self.trace.as_mut() { + trace.capture(generation, self.frame_counter); + } if let Some(rank) = archive_rank { self.best_archived_rank = Some(rank); } diff --git a/services/flysim/crates/flysim/src/trace.rs b/services/flysim/crates/flysim/src/trace.rs new file mode 100644 index 0000000..c7274b0 --- /dev/null +++ b/services/flysim/crates/flysim/src/trace.rs @@ -0,0 +1,362 @@ +//! `FLY_TRACE=`: a per-frame record of the legacy loop, for parity and for the port. +//! +//! One JSON object per line. The first line names the format; every other line is either one +//! transition `k -> k+1` of the legacy frame order (`crate::frame`) together with what happened at +//! the boundary it reached, or -- once, before the first transition -- the captures taken at the +//! boundary the run started on. +//! +//! The field names follow the step trace of the session framework +//! (`docs/design/session-framework/step-v1.md` section 8 and its 2026-09-23 amendment, Rust +//! `fly_session_types::trace`) wherever a legacy field is the same thing: +//! +//! - `behaviour` is what two runs of one build pair must agree on, byte for byte: +//! `step`, `ticksAdvanced`, `brainTicks` and `remainder` (a `RationalNs`, exact, because every +//! legacy remainder is a multiple of 2^-15 ms: legacy-gameboy-v1 section 3), the decision, the +//! controller mask, the macro and reward events in the order they happened, the digests of the +//! rates, of the spikes of this transition, of the frame and of work RAM, the rank, and +//! `boundaryActions` -- a ratchet capture is a `save-slot` of slot `best` with the digest of the +//! saved emulator state, a ratchet recovery is a `rollback` to `best` -- in the shape of +//! `TraceBehaviour.boundaryActions`; +//! - `operational.captures` is every checkpoint taken at the boundary, in order, each with the +//! number of boundary actions applied before it, in the shape of `TraceOperational.captures`. +//! The legacy loop archives a milestone *before* the ratchet captures (legacy-gameboy-v1 section +//! 4), so a climb records `afterActions: 0` ahead of a `save-slot`: the declared difference, as +//! it happens, which the ported loop's validator refuses on purpose. +//! +//! `admissions` is sugar and operator reward pulses applied at the top of the frame, before the +//! brain ticks: the admission cut of legacy-gameboy-v1 section 15. +//! +//! Wall time is never written, so a run with the periodic checkpoint intervals pushed out of the +//! way produces the same file twice. Off by default; when off, nothing here is constructed and the +//! loop pays one `Option` test per hook. + +use std::fs::File; +use std::io::{BufWriter, Write}; +use std::path::Path; + +use flybrain_core::lif::LifNetwork; +use flybrain_gb::RewardEvent; +use flybrain_gb::emulator::Emulator; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; + +use crate::macros::MacroEvent; + +/// The format name on the first line. +pub const FORMAT: &str = "flysim-legacy-frame-trace-v1"; + +/// The environment variable that turns the trace on. +pub const ENV: &str = "FLY_TRACE"; + +/// The ratchet's one slot, as the legacy composition declares it (legacy-gameboy-v1 section 9). +pub const SLOT: &str = "best"; + +/// Work RAM, `$C000..=$DFFF`, as the CPU sees it. +const WRAM: std::ops::RangeInclusive = 0xc000..=0xdfff; + +/// Lowercase hex SHA-256. +pub fn sha256_hex(bytes: &[u8]) -> String { + hex(&Sha256::digest(bytes)) +} + +fn hex(bytes: &[u8]) -> String { + use std::fmt::Write as _; + let mut out = String::with_capacity(bytes.len() * 2); + for byte in bytes { + let _ = write!(out, "{byte:02x}"); + } + out +} + +/// A legacy remainder in milliseconds as exact nanoseconds, `{numerator, denominator}` in lowest +/// terms. The remainder is always `m / 32768` ms for an integer `m` (legacy-gameboy-v1 section +/// 3), so this never rounds; a value that is not is written with its bits instead. +pub fn remainder_ns(remainder_ms: f64) -> Value { + let scaled = remainder_ms * 32_768.0; + if scaled.fract() != 0.0 || !(0.0..32_768.0 * 1_000.0).contains(&scaled) { + return json!({ "inexactBits": format!("{:016x}", remainder_ms.to_bits()) }); + } + // ns = m * 1e6 / 32768 = m * 15625 / 512. + let mut numerator = scaled as u64 * 15_625; + let mut denominator = 512u64; + let (mut a, mut b) = (numerator, denominator); + while b != 0 { + (a, b) = (b, a % b); + } + if a > 1 { + numerator /= a; + denominator /= a; + } + if numerator == 0 { + denominator = 1; + } + json!({ "numerator": numerator.to_string(), "denominator": denominator.to_string() }) +} + +fn macro_json(phase: &str, event: &MacroEvent) -> Value { + json!({ + "phase": phase, + "slot": event.slot, + "name": event.name, + "outcome": event.outcome.map(|outcome| outcome.as_str()), + }) +} + +/// One transition and the boundary it reached, while it is being recorded. +#[derive(Default)] +struct Record { + step: u64, + admissions: Vec, + ticks: u64, + brain_ticks: u64, + remainder: Value, + rates: String, + spikes: String, + spike_count: u64, + decision: Vec, + mask: u32, + macros: Vec, + framebuffer: String, + wram: String, + rewards: Vec, + rank: u32, + boundary_actions: Vec, + captures: Vec, +} + +/// The recorder. The loop calls it at fixed points of the frame order; it holds the transition +/// open until the next one starts, so the captures and admissions a host takes between two frames +/// land on the boundary they belong to. +pub struct FrameTrace { + out: BufWriter, + /// Captures at the boundary the run started on, before any transition. + initial: Vec, + initial_step: Option, + open: Option, + /// Admissions since the last transition started: they belong to the next one. + admissions: Vec, + /// Brain clock before this transition's ticks, the lower edge of its spike window. + ms_before: f64, +} + +impl FrameTrace { + /// `FLY_TRACE`, if it is set and non-empty. A path that cannot be created is an error rather + /// than a silent run without the trace that was asked for. + pub fn from_env() -> std::io::Result> { + match std::env::var_os(ENV) { + Some(path) if !path.is_empty() => Self::create(Path::new(&path)).map(Some), + _ => Ok(None), + } + } + + pub fn create(path: &Path) -> std::io::Result { + let mut out = BufWriter::with_capacity(1 << 20, File::create(path)?); + writeln!(out, "{}", json!({ "format": FORMAT }))?; + Ok(Self { + out, + initial: Vec::new(), + initial_step: None, + open: None, + admissions: Vec::new(), + ms_before: 0.0, + }) + } + + fn write(&mut self, value: &Value) { + if let Err(error) = writeln!(self.out, "{value}") { + tracing::warn!(%error, "could not write the frame trace"); + } + } + + /// Sugar admitted at the top of the frame, before the brain ticks. + pub fn sugar(&mut self, duration_ms: f64) { + self.admissions.push(json!({ "kind": "sugar", "durationMs": duration_ms })); + } + + /// An operator reward pulse, applied at the top of the frame. + pub fn reward_pulse(&mut self, value: f64) { + self.admissions.push(json!({ "kind": "reward", "value": value })); + } + + /// A checkpoint capture at the current boundary: the open transition's, or the start's. + pub fn capture(&mut self, generation: u64, step: u64) { + let (captures, after) = match self.open.as_mut() { + Some(record) => { + let after = record.boundary_actions.len(); + (&mut record.captures, after) + } + None => { + self.initial_step.get_or_insert(step); + (&mut self.initial, 0) + } + }; + captures.push(json!({ "checkpointId": format!("g{generation}"), "afterActions": after })); + } + + /// Transition `step -> step + 1` begins: the previous one is complete. + pub fn begin(&mut self, step: u64, ms_before: f64) { + self.flush_open(); + if let Some(start) = self.initial_step.take() { + let initial = std::mem::take(&mut self.initial); + self.write(&json!({ + "boundary": start.to_string(), + "operational": { "captures": initial }, + })); + } + self.ms_before = ms_before; + self.open = Some(Record { + step, + admissions: std::mem::take(&mut self.admissions), + ..Record::default() + }); + } + + /// Phase A's ticks, and the network they left behind. + pub fn ticked(&mut self, ticks: u64, remainder_ms: f64, network: &LifNetwork) { + let Some(record) = self.open.as_mut() else { return }; + record.ticks = ticks; + record.brain_ticks = network.ms as u64; + record.remainder = remainder_ns(remainder_ms); + let mut hasher = Sha256::new(); + for (role, rate) in network.rates.iter() { + hasher.update((role.len() as u32).to_le_bytes()); + hasher.update(role.as_bytes()); + hasher.update(rate.to_bits().to_le_bytes()); + } + record.rates = hex(&hasher.finalize()); + let (bits, count) = + crate::snapshot::spike_bitset(&network.last_spike_ms, self.ms_before, network.ms); + record.spikes = sha256_hex(&bits); + record.spike_count = count; + } + + /// The readout's decision, as decoded. + pub fn decided(&mut self, active: &[String]) { + if let Some(record) = self.open.as_mut() { + record.decision = active.to_vec(); + } + } + + /// Phase B: the mask the emulator is given, and the macro events deciding it produced. + pub fn executed(&mut self, mask: u32, events: &[MacroEvent]) { + if let Some(record) = self.open.as_mut() { + record.mask = mask; + record.macros.extend(events.iter().map(|event| macro_json("execute", event))); + } + } + + /// The frame the emulator produced, and work RAM after it. Read uncached, so the trace never + /// fills the per-frame read cache the adapter and the macros share. + pub fn advanced(&mut self, framebuffer: &[u8], emulator: &Emulator) { + if let Some(record) = self.open.as_mut() { + record.framebuffer = sha256_hex(framebuffer); + let wram: Vec = WRAM.map(|address| emulator.read_uncached(address)).collect(); + record.wram = sha256_hex(&wram); + } + } + + /// Phase C: the reward events in adapter order, the scene's own macro events, and the rank. + pub fn evaluated(&mut self, rewards: &[RewardEvent], abandoned: &[MacroEvent], rank: u32) { + if let Some(record) = self.open.as_mut() { + record.rewards.extend(rewards.iter().map(|event| { + json!({ + "kind": event.kind, + "value": event.value, + "stimulationMs": event.stimulation_ms, + }) + })); + record.macros.extend(abandoned.iter().map(|event| macro_json("evaluate", event))); + record.rank = rank; + } + } + + /// The ratchet captured: a slot save at the boundary. + pub fn slot_saved(&mut self, state: &[u8]) { + if let Some(record) = self.open.as_mut() { + record.boundary_actions.push(json!({ + "kind": "save-slot", + "slotId": SLOT, + "stateDigest": sha256_hex(state), + })); + } + } + + /// The ratchet rolled back, and the macro events the rollback produced. + pub fn rolled_back(&mut self, events: &[MacroEvent]) { + if let Some(record) = self.open.as_mut() { + record.boundary_actions.push(json!({ + "kind": "rollback", + "slotId": SLOT, + "stateDigest": null, + })); + record.macros.extend(events.iter().map(|event| macro_json("rollback", event))); + } + } + + fn flush_open(&mut self) { + let Some(record) = self.open.take() else { return }; + let line = json!({ + "behaviour": { + "step": record.step.to_string(), + "admissions": record.admissions, + "ticksAdvanced": record.ticks.to_string(), + "brainTicks": record.brain_ticks.to_string(), + "remainder": record.remainder, + "ratesDigest": record.rates, + "spikesDigest": record.spikes, + "spikeCount": record.spike_count, + "decision": record.decision, + "mask": record.mask, + "macroEvents": record.macros, + "framebufferDigest": record.framebuffer, + "wramDigest": record.wram, + "rewards": record.rewards, + "rank": record.rank, + "acknowledgedBoundary": (record.step + 1).to_string(), + "boundaryActions": record.boundary_actions, + }, + "operational": { "captures": record.captures }, + }); + self.write(&line); + } + + /// Write the open transition and flush the file: on shutdown, and on drop. + pub fn finish(&mut self) { + self.flush_open(); + if let Err(error) = self.out.flush() { + tracing::warn!(%error, "could not flush the frame trace"); + } + } +} + +impl Drop for FrameTrace { + fn drop(&mut self) { + self.finish(); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn every_legacy_remainder_is_exact_nanoseconds() { + // The first twelve legacy frames' remainders, from the clock the fixture records. + let per_frame = 1000.0 / (4_194_304.0 / 70_224.0); + let mut remainder = 0.0f64; + for _ in 0..100_000 { + remainder += per_frame; + remainder -= remainder.floor(); + let value = remainder_ns(remainder); + assert!(value.get("numerator").is_some(), "{remainder} was not exact: {value}"); + } + assert_eq!(remainder_ns(0.0), json!({ "numerator": "0", "denominator": "1" })); + // 0.5 ms is 500000 ns. + assert_eq!(remainder_ns(0.5), json!({ "numerator": "500000", "denominator": "1" })); + // One 2^-15 ms step is 15625/512 ns. + assert_eq!( + remainder_ns(1.0 / 32_768.0), + json!({ "numerator": "15625", "denominator": "512" }) + ); + } +}