//! The T0 runtime: per-owner state, memo invalidation, and deterministic replay.
//!
//! Replaces `side_channel_generation.rs`. The registries a widget reads besides its graph
//! (files, models, completions, captures, board latches) were process-wide statics with a
//! process-wide generation counter. They now belong to a `Runtime`, so these tests need no
//! lock: each builds its own.

use flowrs_core::graph::GraphExecutor;
use flowrs_core::graph::llm_runtime::{Ask, Completion};
use flowrs_core::graph::runtime::{Event, Runtime, replay};
use flowrs_core::graph::twin_runtime::Capture;
use flowrs_core::graph::widget::WidgetRegistry;
use flowrs_core::graph::{Graph, WireMode};

fn capture() -> Capture {
    Capture {
        samples: Some(vec![1.0, 0.0]),
        sample_rate_hz: 1000.0,
        aliased: false,
        source: "test".into(),
        serial: String::new(),
    }
}

/// Every mutator must invalidate the T0 memo, which folds in the generation; a read must
/// not.
#[test]
fn every_mutator_bumps_the_generation_and_reads_do_not() {
    let rt = Runtime::new();
    let mut last = rt.generation();
    let mut check = |label: &str, rt: &Runtime| {
        let g = rt.generation();
        assert!(g > last, "{label} did not bump the generation");
        last = g;
    };
    rt.put_file("f", vec![1]);
    check("put_file", &rt);
    rt.put_model("m", vec![1]);
    check("put_model", &rt);
    rt.put_capture("c", capture());
    check("put_capture", &rt);
    rt.put_completion(
        &Ask::new("p", "m", 0.0),
        Completion {
            text: "t".into(),
            model: "m".into(),
        },
    );
    check("put_completion", &rt);
    for kind in ["file", "model", "capture", "completion"] {
        rt.clear(kind);
        check(kind, &rt);
    }
    let g = rt.generation();
    let _ = rt.files.get("f");
    let _ = rt.models.names();
    let _ = rt.captures.get("c");
    let _ = rt.completions.is_empty();
    assert_eq!(rt.generation(), g, "a read bumped the generation");
}

/// Regression: two runtimes in one process no longer share state. A file one owner
/// published, or a pin one board latched, was visible to every other.
#[test]
fn two_runtimes_share_nothing() {
    let a = Runtime::new();
    let b = Runtime::new();
    a.put_file("run.csv", b"x".to_vec());
    a.board.digital_write(13, true);
    assert_eq!(b.files.get("run.csv"), None);
    assert!(!b.board.digital_read(13));
    assert!(a.board.digital_read(13));
}

fn adc_graph() -> Graph {
    let mut g = Graph::new();
    let src = g.add_node("embedded.adc", 0.0, 0.0);
    g.get_node_mut(src)
        .unwrap()
        .properties
        .insert("channel".into(), "3".into());
    let abs = g.add_node("math.abs", 200.0, 0.0);
    let reg = WidgetRegistry::with_builtins();
    let out = reg.get("embedded.adc").unwrap().outputs()[0].name.clone();
    let _ = g.add_wire(src, &out, abs, "value", WireMode::Move);
    g
}

/// A run is a function of its graph and the logged external changes, so a log replays:
/// every step and in-node change reproduces on a fresh runtime.
#[test]
fn a_logged_run_replays_exactly() {
    let rt = Runtime::new();
    let mut exec = GraphExecutor::with_runtime(WidgetRegistry::with_builtins(), rt.clone());
    let g = adc_graph();
    rt.put_file("notes.txt", b"bench 4".to_vec());
    exec.run(&g).expect("first run");
    rt.put_model("m.onnx", vec![1, 2, 3]);
    exec.run(&g).expect("second run");

    let log = rt.log();
    let steps = log
        .events
        .iter()
        .filter(|e| matches!(e, Event::Step { .. }))
        .count();
    assert_eq!(steps, 4, "two nodes, two runs: {:?}", log.events);
    assert!(
        log.events.iter().any(|e| matches!(e, Event::External(_))),
        "external changes are logged"
    );
    replay(&log, &g, WidgetRegistry::with_builtins()).expect("the log reproduces");
}

/// Replay is a check, not an echo: a log whose recorded outputs were tampered with fails.
#[test]
fn a_tampered_log_does_not_replay() {
    let rt = Runtime::new();
    let mut exec = GraphExecutor::with_runtime(WidgetRegistry::with_builtins(), rt.clone());
    let g = adc_graph();
    exec.run(&g).expect("run");
    let mut log = rt.log();
    for e in &mut log.events {
        if let Event::Step { outputs, .. } = e {
            for v in outputs.values_mut() {
                v.push_str(" (edited)");
            }
        }
    }
    let err = replay(&log, &g, WidgetRegistry::with_builtins()).unwrap_err();
    assert!(err[0].contains("diverges"), "{err:?}");
}

/// After an edit, the runs made with the old graph diverge by construction; `replay_from`
/// checks only the runs since, starting from the state they saw (the file published
/// before run 0 is applied, not lost with the run that is skipped).
#[test]
fn replay_from_checks_only_the_runs_since_an_edit() {
    use flowrs_core::graph::runtime::replay_from;
    let rt = Runtime::new();
    let mut exec = GraphExecutor::with_runtime(WidgetRegistry::with_builtins(), rt.clone());
    let mut g = Graph::new();
    let path = g.add_node("StringLiteral", 0.0, 0.0);
    g.get_node_mut(path)
        .unwrap()
        .properties
        .insert("value".into(), "in.txt".into());
    let read = g.add_node("ReadFile", 0.0, 0.0);
    g.add_wire(path, "value", read, "path", WireMode::Copy)
        .unwrap();
    rt.put_file("in.txt", b"abc".to_vec());
    exec.run(&g).expect("run 0");
    // The edit: the read is now checked.
    let ok = g.add_node("IsOk", 0.0, 0.0);
    g.add_wire(read, "contents", ok, "input", WireMode::Copy)
        .unwrap();
    exec.run(&g).expect("run 1");
    assert!(exec.unexecuted().is_empty(), "{:?}", exec.unexecuted());

    let log = rt.log();
    let whole = replay(&log, &g, WidgetRegistry::with_builtins()).unwrap_err();
    assert!(
        whole.len() == 1 && whole[0].starts_with("run 0 "),
        "{whole:?}"
    );
    replay_from(&log, &g, WidgetRegistry::with_builtins(), 1).expect("run 1 reproduces");
}

/// A change a NODE made during a skipped run is state the next run starts from too: a
/// file written in run 0 is there when run 1 reads it. (No built-in node writes a store
/// while it runs today, so the writer here is the runtime's own step API.)
#[test]
fn replay_from_applies_what_skipped_runs_wrote() {
    use flowrs_core::graph::runtime::replay_from;
    use std::collections::HashMap;
    let rt = Runtime::new();
    let none = HashMap::new();
    rt.begin_run();
    rt.begin_step(99);
    rt.put_file("in.txt", b"written in run 0".to_vec());
    rt.end_step(99, "test.writer", &HashMap::new(), &none, &none);

    let mut g = Graph::new();
    let path = g.add_node("StringLiteral", 0.0, 0.0);
    g.get_node_mut(path)
        .unwrap()
        .properties
        .insert("value".into(), "in.txt".into());
    let read = g.add_node("ReadFile", 0.0, 0.0);
    g.add_wire(path, "value", read, "path", WireMode::Copy)
        .unwrap();
    let mut exec = GraphExecutor::with_runtime(WidgetRegistry::with_builtins(), rt.clone());
    exec.run(&g).expect("run 1");
    assert!(exec.unexecuted().is_empty(), "{:?}", exec.unexecuted());
    let log = rt.log();
    assert!(
        log.events
            .iter()
            .any(|e| matches!(e, Event::Internal { node: 99, .. })),
        "the write is logged as the node's own"
    );
    replay_from(&log, &g, WidgetRegistry::with_builtins(), 1).expect("run 1 reproduces");
}

/// A twin capture lives in the runtime of whoever ran the twin; the exported program
/// carries it only when codegen is given that runtime.
#[test]
fn the_exported_program_carries_the_runtimes_capture() {
    let mut g = Graph::new();
    let n = g.add_node("twin.capture", 0.0, 0.0);
    g.get_node_mut(n)
        .unwrap()
        .properties
        .insert("capture".into(), "bench".into());
    let rt = Runtime::new();
    rt.put_capture(
        "bench",
        Capture {
            samples: Some(vec![1.0, 0.0, 1.0]),
            source: "uno D13".into(),
            ..capture()
        },
    );
    let with = g.to_rust_with_runtime(&rt).expect("codegen");
    assert!(with.contains("uno D13"), "{with}");
    let without = g.to_rust().expect("codegen");
    assert!(without.contains("no capture"), "{without}");
}

#[test]
fn the_replay_log_is_a_ring_buffer_of_whole_runs() {
    use flowrs_core::graph::runtime::{Event, Runtime};
    use std::collections::HashMap;
    let rt = Runtime::new();
    rt.set_log_cap(Some(7));
    let params: HashMap<String, String> = HashMap::new();
    let none = HashMap::new();
    for run in 0..100u64 {
        rt.begin_run();
        for n in 0..2 {
            rt.begin_step(n);
            rt.end_step(n, "t", &params, &none, &none);
        }
        let log = rt.log();
        assert!(log.events.len() <= 7, "run {run}: {}", log.events.len());
        assert!(
            matches!(log.events[0], Event::RunStart),
            "starts at a run boundary"
        );
    }
    let log = rt.log();
    // Three events per run; 7 keeps the last two whole runs.
    assert_eq!(log.events.len(), 6);
    assert_eq!(log.dropped, 300 - 6);
    // A single run longer than the cap is cut from its front.
    rt.set_log_cap(Some(2));
    assert_eq!(rt.log().events.len(), 2);
    // Unbounded by default.
    let free = Runtime::new();
    for _ in 0..1000 {
        free.begin_run();
    }
    assert_eq!(free.log().events.len(), 1000);
    assert_eq!(free.log().dropped, 0);
}