frame-conv 0.2.0

//! F-3b R3 — the real crash/restart/resume proof: a REAL host subprocess
//! embeds the aion engine over PERSISTENT disk storage, reaches a named
//! window, and dies by real SIGKILL; a fresh process reopens the same
//! store, reconstructs the same workflow from held identity, and
//! subscribes from committed sequence. The F-2a house harness spine
//! (frame-state `src/tests.rs`): parent waits for a durable READY
//! witness before `child.kill()` (SIGKILL on unix), then asserts the
//! specific reconciliation result for each crash window.
//!
//! THE CONTRACT, STATED BEFORE THE KILL (never inferred from whichever
//! run happened):
//! - Committed history is REPLAYED on restart, never executed as new
//!   work: step A's external effect occurs exactly once across BOTH
//!   processes, and the full-history observation carries exactly one
//!   `StepCompleted` for it.
//! - An uncommitted step MAY be re-executed by the engine after the
//!   kill: step B's `begin` count may exceed one in the mid-step window
//!   (disclosed, asserted), while its EFFECT — keyed by the declared
//!   idempotency key, the deterministic activity id the dispatch
//!   carries — commits exactly once.
//! - The post-restart observation from cursor 1 is gap-free and
//!   duplicate-free, and reaches the Completed terminal.
//! - frame-conv wrote NOTHING durable: every file in the store directory
//!   is engine storage or harness-owned (asserted by name).
//!
//! Windows:
//! - **between-steps**: after step A's completion commit, before step B
//!   admission — the workflow parks at a durable signal gate, so the
//!   window is quiescent by construction, not by timing.
//! - **mid-step**: step B's external effect has BEGUN (its `begin` hits
//!   the ledger — the READY witness) but its completion commit never
//!   lands (the child's dispatcher parks forever).

#![allow(clippy::expect_used, clippy::unwrap_used)]

mod support;

use std::error::Error;
use std::fs;
use std::io::{BufRead, BufReader, Write};
use std::num::NonZeroU64;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::sync::mpsc;
use std::time::Duration;

use aion::{ActivityDispatch, ActivityDispatcher};
use frame_conv::workflow::{
    WorkflowClient, WorkflowConversationId, WorkflowIdentity, WorkflowItem, WorkflowProgressKind,
    WorkflowRunId, WorkflowTerminal,
};
use serde::Deserialize;
use support::embedded::EmbeddedAion;

/// The two-step workflow: an effectful step, a durable signal gate, a
/// second effectful step that routes the outcome.
const TWO_STEP_AWL: &str = r"//! Two effectful steps split by a durable gate.
workflow two_steps
  input job: String
  signal proceed: Go
  outcome done: type Done, route success

type Go   { note: String }
type Ack  { receipt: String }
type Done { receipt: String }

worker effects
  action step_a(job: String) -> Ack
  action step_b(job: String) -> Done

step first
  step_a(job: job) -> a

step gate
  wait proceed -> go

step second
  step_b(job: job) -> b

  b |> route done
";

#[derive(Debug, Deserialize, PartialEq)]
struct Done {
    receipt: String,
}

#[derive(Debug, Deserialize, PartialEq)]
struct RoutedOutcome<P> {
    outcome: String,
    payload: P,
}

/// Positive budget for one observation step or ledger poll window.
const OBSERVE: Duration = Duration::from_secs(20);

/// The append-only effect ledger: `begin <name> <activity> attempt=<n>`
/// on every dispatch; `effect <name> <activity>` exactly once per
/// activity id — the DECLARED idempotency mechanism (key = the
/// deterministic activity id the dispatch carries).
struct Ledger {
    path: PathBuf,
}

impl Ledger {
    fn begin(&self, dispatch: &ActivityDispatch) {
        self.append(&format!(
            "begin {} {} attempt={}\n",
            dispatch.name, dispatch.activity_id, dispatch.attempt
        ));
    }

    /// Commits the effect if this activity id has not already committed
    /// it — a re-executed attempt observes the first attempt's effect.
    fn effect_once(&self, dispatch: &ActivityDispatch) {
        let line = format!("effect {} {}\n", dispatch.name, dispatch.activity_id);
        let existing = fs::read_to_string(&self.path).unwrap_or_default();
        if !existing.contains(&line) {
            self.append(&line);
        }
    }

    fn append(&self, line: &str) {
        let mut file = fs::OpenOptions::new()
            .create(true)
            .append(true)
            .open(&self.path)
            .expect("ledger opens");
        file.write_all(line.as_bytes()).expect("ledger appends");
        file.sync_all().expect("ledger reaches disk");
    }
}

/// The child-side dispatcher: records every begin, commits effects
/// idempotently, and — when told to wedge a step — parks that dispatch
/// forever, holding the mid-step window open until SIGKILL.
struct LedgerDispatcher {
    ledger: Ledger,
    wedge: Option<String>,
}

impl ActivityDispatcher for LedgerDispatcher {
    fn dispatch(&self, request: ActivityDispatch) -> Result<String, String> {
        self.ledger.begin(&request);
        if self.wedge.as_deref() == Some(request.name.as_str()) {
            loop {
                std::thread::park();
            }
        }
        self.ledger.effect_once(&request);
        match request.name.as_str() {
            "step_a" => Ok(r#"{"receipt":"a-done"}"#.to_owned()),
            "step_b" => Ok(r#"{"receipt":"b-done"}"#.to_owned()),
            other => Err(format!("unknown activity {other}")),
        }
    }
}

fn ledger_lines(dir: &Path) -> Vec<String> {
    fs::read_to_string(dir.join("effects.log"))
        .unwrap_or_default()
        .lines()
        .map(ToOwned::to_owned)
        .collect()
}

fn dispatcher(dir: &Path, wedge: Option<&str>) -> LedgerDispatcher {
    LedgerDispatcher {
        ledger: Ledger {
            path: dir.join("effects.log"),
        },
        wedge: wedge.map(ToOwned::to_owned),
    }
}

/// Child body: builds the embedded engine on the parent-named store dir,
/// opens the workflow, drives it to the window, persists the identity,
/// prints READY, and parks until SIGKILL.
fn run_child(window: &str, dir: &Path) -> Result<(), Box<dyn Error>> {
    let wedge = (window == "mid-step").then_some("step_b");
    let substrate = EmbeddedAion::start_at(dir.to_path_buf(), &[TWO_STEP_AWL], |builder| {
        builder.activity_dispatcher(std::sync::Arc::new(dispatcher(dir, wedge)))
    })?;
    let client = WorkflowClient::from_substrate_client(substrate.client())?;
    let conversation = client.open("two_steps", &serde_json::json!({"job": "j1"}), "crash-open")?;
    let identity = conversation.identity();
    fs::write(
        dir.join("identity"),
        [
            identity.conversation.to_bytes().as_slice(),
            identity.run.to_bytes().as_slice(),
        ]
        .concat(),
    )?;

    match window {
        // Between steps: step A commits, the workflow parks at the gate.
        // READY only after the completion is OBSERVED in history.
        "between-steps" => {
            let mut observation = conversation.observe_from(NonZeroU64::new(1).expect("nonzero"));
            loop {
                if let WorkflowItem::Progress(progress) = observation.next_item()?
                    && matches!(progress.kind, WorkflowProgressKind::StepCompleted { .. })
                {
                    break;
                }
            }
        }
        // Mid-step: contribute past the gate; step B's begin reaching
        // the ledger is the READY condition (its dispatch is wedged, so
        // the completion commit can never land).
        "mid-step" => {
            conversation.contribute("proceed", &serde_json::json!({"note": "go"}))?;
            let (sender, receiver) = mpsc::channel();
            let dir_owned = dir.to_path_buf();
            std::thread::spawn(move || {
                loop {
                    if ledger_lines(&dir_owned)
                        .iter()
                        .any(|line| line.starts_with("begin step_b"))
                    {
                        let _ = sender.send(());
                        break;
                    }
                    std::thread::yield_now();
                }
            });
            receiver.recv_timeout(OBSERVE)?;
        }
        other => return Err(format!("unknown window {other}").into()),
    }

    println!("READY {window}");
    std::io::stdout().flush()?;
    loop {
        std::thread::park();
    }
}

/// Parent side: spawn the child (this same test binary, env-gated), wait
/// for the durable READY witness, then real SIGKILL.
fn run_and_kill_child(window: &str, dir: &Path) -> Result<(), Box<dyn Error>> {
    let executable = std::env::current_exe()?;
    let mut child = Command::new(executable)
        .args([
            "--exact",
            "kill_nine_between_steps_and_mid_step",
            "--nocapture",
        ])
        .env("F3B_CRASH_WINDOW", window)
        .env("F3B_CRASH_DIR", dir)
        .stdout(Stdio::piped())
        .spawn()?;
    let stdout = child.stdout.take().expect("child stdout was piped");
    let mut ready = false;
    for line in BufReader::new(stdout).lines() {
        if line?.contains(&format!("READY {window}")) {
            ready = true;
            break;
        }
    }
    if !ready {
        let _ = child.kill();
        return Err(format!("{window}: child exited before the READY witness").into());
    }
    child.kill()?;
    let status = child.wait()?;
    if status.success() {
        return Err(format!("{window}: SIGKILLed child reported success").into());
    }
    Ok(())
}

fn held_identity(dir: &Path) -> Result<WorkflowIdentity, Box<dyn Error>> {
    let bytes = fs::read(dir.join("identity"))?;
    if bytes.len() != 32 {
        return Err(format!("identity file holds {} bytes, want 32", bytes.len()).into());
    }
    let mut conversation = [0_u8; 16];
    let mut run = [0_u8; 16];
    conversation.copy_from_slice(&bytes[..16]);
    run.copy_from_slice(&bytes[16..]);
    Ok(WorkflowIdentity {
        conversation: WorkflowConversationId::from_bytes(conversation),
        run: WorkflowRunId::from_bytes(run),
    })
}

/// Collects one observation through its FRONTIER terminal within the
/// budget (a replayed superseded terminal would not end this run's
/// history — see the workflow module's consumer-side law — but this
/// workflow is never reopened, so the first terminal is the frontier).
fn collect_through_terminal(
    client: &WorkflowClient,
    identity: WorkflowIdentity,
) -> Result<Vec<WorkflowItem>, Box<dyn Error>> {
    let mut observation = client
        .conversation(identity)
        .observe_from(NonZeroU64::new(1).expect("nonzero"));
    let (sender, receiver) = mpsc::channel();
    std::thread::spawn(move || {
        loop {
            let item = observation.next_item();
            let last = matches!(&item, Ok(WorkflowItem::Terminal(_)) | Err(_));
            if sender.send(item).is_err() || last {
                break;
            }
        }
    });
    let mut collected = Vec::new();
    loop {
        let item = receiver.recv_timeout(OBSERVE)??;
        let done = matches!(item, WorkflowItem::Terminal(_));
        collected.push(item);
        if done {
            return Ok(collected);
        }
    }
}

fn item_seq(item: &WorkflowItem) -> u64 {
    match item {
        WorkflowItem::Progress(progress) => progress.seq,
        WorkflowItem::Terminal(
            WorkflowTerminal::Completed { seq, .. }
            | WorkflowTerminal::Failed { seq, .. }
            | WorkflowTerminal::Cancelled { seq, .. }
            | WorkflowTerminal::TimedOut { seq, .. }
            | WorkflowTerminal::ContinuedAsNew { seq, .. },
        ) => *seq,
    }
}

/// Every name the store directory may contain after the proof: engine
/// storage, compiled archives, and harness-owned files. Anything else is
/// durable state someone wrote that the contract bans.
fn assert_no_foreign_durable_state(dir: &Path) -> Result<(), Box<dyn Error>> {
    for entry in fs::read_dir(dir)? {
        let name = entry?.file_name().to_string_lossy().into_owned();
        let sanctioned = name.starts_with("aion.db")
            || name.starts_with("workflow-")
            || name == "effects.log"
            || name == "identity";
        if !sanctioned {
            return Err(format!("unsanctioned durable file in the store dir: {name}").into());
        }
    }
    Ok(())
}

/// R3 — both kill windows, end to end. Child body runs when env-gated
/// (the F-2a in-binary re-entry shape); the parent otherwise drives both
/// windows sequentially against separate stores.
#[test]
fn kill_nine_between_steps_and_mid_step() -> Result<(), Box<dyn Error>> {
    if let Ok(window) = std::env::var("F3B_CRASH_WINDOW") {
        let dir = PathBuf::from(std::env::var("F3B_CRASH_DIR")?);
        return run_child(&window, &dir);
    }

    for window in ["between-steps", "mid-step"] {
        let dir = support::store_dir(&format!("workflow-crash-{window}"))?;
        run_and_kill_child(window, &dir)?;

        let killed_lines = ledger_lines(&dir);
        assert!(
            killed_lines
                .iter()
                .any(|line| line.starts_with("effect step_a")),
            "{window}: step A's effect committed before the kill: {killed_lines:?}"
        );

        // Restart: a fresh engine over the SAME disk store; startup
        // recovery re-drives from committed truth; the dispatcher now
        // completes every step.
        let substrate = EmbeddedAion::start_at(dir.clone(), &[TWO_STEP_AWL], |builder| {
            builder.activity_dispatcher(std::sync::Arc::new(dispatcher(&dir, None)))
        })?;
        let client = WorkflowClient::from_substrate_client(substrate.client())?;
        let identity = held_identity(&dir)?;

        // Between-steps: the gate is still waiting — contribute past it.
        if window == "between-steps" {
            client
                .conversation(identity)
                .contribute("proceed", &serde_json::json!({"note": "resume"}))?;
        }

        let items = collect_through_terminal(&client, identity)?;

        // The stated contract, asserted per clause.
        let seqs: Vec<u64> = items.iter().map(item_seq).collect();
        let expected: Vec<u64> = (seqs[0]..seqs[0] + seqs.len() as u64).collect();
        assert_eq!(seqs, expected, "{window}: gap-free, duplicate-free");

        let step_completions = items
            .iter()
            .filter(|item| {
                matches!(
                    item,
                    WorkflowItem::Progress(progress)
                        if matches!(progress.kind, WorkflowProgressKind::StepCompleted { .. })
                )
            })
            .count();
        assert_eq!(
            step_completions, 2,
            "{window}: exactly one committed completion per step — replay, not re-execution"
        );

        let WorkflowItem::Terminal(WorkflowTerminal::Completed { result, .. }) =
            items.last().expect("nonempty")
        else {
            return Err(format!("{window}: the run must complete: {items:?}").into());
        };
        let routed: RoutedOutcome<Done> = result.decode()?;
        assert_eq!(routed.outcome, "done");
        assert_eq!(routed.payload.receipt, "b-done");

        let final_lines = ledger_lines(&dir);
        let effect_a = final_lines
            .iter()
            .filter(|line| line.starts_with("effect step_a"))
            .count();
        let effect_b = final_lines
            .iter()
            .filter(|line| line.starts_with("effect step_b"))
            .count();
        let begin_b = final_lines
            .iter()
            .filter(|line| line.starts_with("begin step_b"))
            .count();
        assert_eq!(
            effect_a, 1,
            "{window}: committed step A occurred exactly once across both processes"
        );
        assert_eq!(
            effect_b, 1,
            "{window}: step B's effect committed exactly once"
        );
        match window {
            // The uncommitted step was BEGUN in the killed process and
            // re-executed after restart — the disclosed duplicate risk,
            // absorbed by the declared idempotent effect.
            "mid-step" => assert!(
                begin_b >= 2,
                "mid-step: the wedged begin plus the re-execution: {final_lines:?}"
            ),
            _ => assert_eq!(
                begin_b, 1,
                "between-steps: step B was never admitted before the kill"
            ),
        }

        assert_no_foreign_durable_state(&dir)?;
    }
    Ok(())
}

/// R4 — the completion/last-disconnect race: the terminal commit and the
/// final observer's departure release together; whichever linearization
/// wins, the terminal lives in committed history and a LATER
/// cursor subscription observes it.
#[test]
fn completion_and_last_disconnect_race_leaves_terminal_observable() -> Result<(), Box<dyn Error>> {
    let dir = support::store_dir("workflow-race")?;
    let substrate = EmbeddedAion::start_at(dir.clone(), &[TWO_STEP_AWL], |builder| {
        builder.activity_dispatcher(std::sync::Arc::new(dispatcher(&dir, None)))
    })?;
    let client = WorkflowClient::from_substrate_client(substrate.client())?;
    let conversation = client.open(
        "two_steps",
        &serde_json::json!({"job": "race"}),
        "race-open",
    )?;
    let identity = conversation.identity();

    // The racing observer: parked in next_item when the terminal commits.
    let racer = client
        .conversation(identity)
        .observe_from(NonZeroU64::new(1).expect("nonzero"));

    // Release the terminal and drop the last observation together.
    conversation.contribute("proceed", &serde_json::json!({"note": "go"}))?;
    drop(racer);

    // One linearization won; the terminal is committed truth either way:
    // a LATER subscription from cursor 1 observes the full history
    // through Completed.
    let items = collect_through_terminal(&client, identity)?;
    let WorkflowItem::Terminal(WorkflowTerminal::Completed { .. }) =
        items.last().expect("nonempty")
    else {
        return Err(format!("the terminal survives the race: {items:?}").into());
    };
    Ok(())
}