openquiver-runtime 0.1.0

//! Killing a run and picking it up again.
//!
//! The central tests of this crate, not extra ones. A resumed run that finishes
//! correctly proves very little on its own — a run that simply started over
//! would also finish correctly. The invariant worth proving is that the agent
//! was never asked the same thing twice, because a re-prompt silently pays for
//! the same turn again and is invisible in the result.

mod support;

use std::collections::BTreeMap;

use openquiver_acp::mock::{MockAgents, Script, SessionStart};
use openquiver_runtime::{InDoubt, Progress, Runtime, RuntimeError};
use openquiver_store::{RunStatus, Store};
use serde_json::json;

use openquiver_nodes::{NodeType, Pin, Purity, Registry};

use support::{plan, plan_with, printed};

/// Two prompt turns, so a kill can land between them.
const TWO_TURNS: &str = r#"
schema 1

graph "two-turns" {
  node start = event.manual
  node agent = acp.connect_agent { agent = "mock" }
  node sess = acp.new_session
  node first = acp.prompt { text = "First question." }
  node say_first = io.print
  node second = acp.prompt { text = "Second question." }
  node say_second = io.print

  start.out -> agent.in
  agent.out -> sess.in
  sess.out -> first.in
  first.end_turn -> say_first.in
  say_first.out -> second.in
  second.end_turn -> say_second.in

  agent.agent => sess.agent
  first.text => say_first.value
  second.text => say_second.value
  sess.session => first.session
  sess.session => second.session
}
"#;

fn script() -> Script {
    Script::new().says("one").says("two")
}

#[tokio::test]
async fn a_run_killed_after_any_node_resumes_without_asking_the_agent_twice() {
    let plan = plan(TWO_TURNS);

    // Every kill point, discovered rather than assumed: a graph that grows a
    // node must not quietly stop being covered at its far end.
    let mut steps = 0;
    {
        let store = Store::open_in_memory().await.expect("store");
        let agents = MockAgents::new(script());
        let mut run = Runtime::start(&store, &plan, &agents, "two-turns")
            .await
            .expect("start");
        while let Progress::Ran(_) = run.step().await.expect("step") {
            steps += 1;
        }
    }
    assert_eq!(steps, 7, "the fixture has seven nodes");

    for killed_after in 0..=steps {
        let store = Store::open_in_memory().await.expect("store");
        let agents = MockAgents::new(script());

        let id = {
            let mut run = Runtime::start(&store, &plan, &agents, "two-turns")
                .await
                .expect("start");
            for _ in 0..killed_after {
                run.step().await.expect("step");
            }
            run.id().clone()
            // Dropped here: the process is gone, and every live session with it.
        };

        let mut resumed = Runtime::resume(&store, &plan, &agents, id.clone(), InDoubt::Ask)
            .await
            .unwrap_or_else(|e| panic!("resume after {killed_after} nodes: {e}"));
        resumed.run().await.expect("finish");

        assert_eq!(
            store.status(&id).await.expect("status"),
            RunStatus::Finished,
            "killed after {killed_after} nodes"
        );
        assert_eq!(
            printed(&store, &id).await,
            ["one", "two"],
            "killed after {killed_after} nodes"
        );
        assert!(
            !agents.transcript().has_repeats(),
            "killed after {killed_after} nodes, the agent saw: {:?}",
            agents.transcript().prompts()
        );

        // Said directly rather than inferred from the agent's transcript: three
        // of the seven nodes never touch an agent, and a run that redid one of
        // those would look innocent in the prompts.
        let mut entered: BTreeMap<String, usize> = BTreeMap::new();
        for event in support::events(&store, &id).await {
            if event.kind == "node.entered" {
                *entered
                    .entry(event.payload["node"].as_str().expect("node").to_owned())
                    .or_default() += 1;
            }
        }
        assert!(
            entered.values().all(|count| *count == 1),
            "killed after {killed_after} nodes, a node ran twice: {entered:?}"
        );
    }
}

#[tokio::test]
async fn a_journal_that_does_not_begin_with_the_start_of_a_run_is_refused() {
    let plan = plan(TWO_TURNS);
    let store = Store::open_in_memory().await.expect("store");
    let agents = MockAgents::new(script());

    // `Runtime::start` records the run and then records that it started: two
    // writes, and a process killed between them leaves this.
    let id = store.start_run("two-turns").await.expect("start");

    let refused = Runtime::resume(&store, &plan, &agents, id, InDoubt::Ask)
        .await
        .err();

    // Read leniently, an empty journal is a run that finished having done
    // nothing — and would be recorded as a success.
    assert!(
        matches!(refused, Some(RuntimeError::BadJournal(_))),
        "{refused:?}"
    );
}

#[tokio::test]
async fn a_run_whose_end_was_recorded_but_never_marked_is_closed_on_resume() {
    let plan = plan(TWO_TURNS);
    let store = Store::open_in_memory().await.expect("store");
    let agents = MockAgents::new(script());

    // The other half of the same two-write problem: the journal says the run
    // ended, the run row still says it is going.
    let id = store.start_run("two-turns").await.expect("start");
    for (kind, payload) in [
        ("run.started", json!({ "plan": plan.fingerprint() })),
        ("run.finished", json!({})),
    ] {
        store.append(&id, kind, &payload).await.expect("append");
    }

    let mut resumed = Runtime::resume(&store, &plan, &agents, id.clone(), InDoubt::Ask)
        .await
        .expect("resume");
    resumed.run().await.expect("finish");

    // Otherwise it stays in the list of runs to pick up for ever, and every
    // startup resumes it into the same nothing.
    assert_eq!(
        store.status(&id).await.expect("status"),
        RunStatus::Finished
    );
    assert!(store.resumable().await.expect("list").is_empty());
}

/// The same workflow with one execution wire taken out.
const TWO_TURNS_EDITED: &str = r#"
schema 1

graph "two-turns" {
  node start = event.manual
  node agent = acp.connect_agent { agent = "mock" }
  node sess = acp.new_session
  node first = acp.prompt { text = "First question." }
  node say_first = io.print
  node second = acp.prompt { text = "Second question." }
  node say_second = io.print

  start.out -> agent.in
  agent.out -> sess.in
  sess.out -> first.in
  say_first.out -> second.in
  second.end_turn -> say_second.in

  agent.agent => sess.agent
  first.text => say_first.value
  second.text => say_second.value
  sess.session => first.session
  sess.session => second.session
}
"#;

#[tokio::test]
async fn a_run_resumed_against_an_edited_graph_is_refused() {
    let original = plan(TWO_TURNS);
    let store = Store::open_in_memory().await.expect("store");
    let agents = MockAgents::new(script());

    let id = {
        let mut run = Runtime::start(&store, &original, &agents, "two-turns")
            .await
            .expect("start");
        for _ in 0..3 {
            run.step().await.expect("step");
        }
        run.id().clone()
    };

    // Somebody deleted `first.end_turn -> say_first.in` while the run was
    // suspended. Replayed against the new graph the run would stop at `first`,
    // print nothing, and be recorded as having finished successfully.
    let edited = plan(TWO_TURNS_EDITED);
    let refused = Runtime::resume(&store, &edited, &agents, id, InDoubt::Ask)
        .await
        .err();

    assert!(
        matches!(refused, Some(RuntimeError::DifferentPlan)),
        "{refused:?}"
    );
}

/// A node type the registry knows and this build cannot run.
///
/// What a plugin node looks like when its library is missing: the graph
/// compiles, the palette drew it, and there is nothing behind it.
fn unimplemented() -> Registry {
    let mut registry = Registry::builtin();
    registry
        .register(
            NodeType::new("plugin.thing", Purity::Impure)
                .with_input(Pin::exec("in", "Run this step."))
                .with_output(Pin::exec("out", "Continue.")),
        )
        .expect("a fresh type id");
    registry
}

const UNRUNNABLE: &str = r#"
schema 1

graph "unrunnable" {
  node start = event.manual
  node thing = plugin.thing

  start.out -> thing.in
}
"#;

#[tokio::test]
async fn a_run_that_can_never_get_further_is_failed_rather_than_left_to_be_picked_up() {
    let plan = plan_with(UNRUNNABLE, unimplemented());
    let store = Store::open_in_memory().await.expect("store");
    let agents = MockAgents::new(script());

    let mut run = Runtime::start(&store, &plan, &agents, "unrunnable")
        .await
        .expect("start");
    let error = run
        .run()
        .await
        .expect_err("nothing implements `plugin.thing`");
    assert!(matches!(error, RuntimeError::NoSuchNodeType(_)), "{error}");

    // Recording this as merely suspended would put it in the list of runs to
    // pick up, where every attempt appends another `node.entered` to a journal
    // that will never reach an end.
    assert_eq!(
        store.status(run.id()).await.expect("status"),
        RunStatus::Failed
    );
    assert!(store.resumable().await.expect("list").is_empty());
}

#[tokio::test]
async fn a_resumed_run_picks_the_session_up_rather_than_opening_another() {
    let plan = plan(TWO_TURNS);
    let store = Store::open_in_memory().await.expect("store");
    let agents = MockAgents::new(script());

    let id = {
        let mut run = Runtime::start(&store, &plan, &agents, "two-turns")
            .await
            .expect("start");
        // start, agent, sess: the session exists and nothing has used it yet.
        for _ in 0..3 {
            run.step().await.expect("step");
        }
        run.id().clone()
    };

    let mut resumed = Runtime::resume(&store, &plan, &agents, id, InDoubt::Ask)
        .await
        .expect("resume");
    resumed.run().await.expect("finish");

    // The whole reason a session handle rather than a session object is written
    // to the journal: the conversation continues instead of starting over.
    assert_eq!(
        agents.transcript().sessions(),
        [
            SessionStart::New {
                agent: "mock".to_owned()
            },
            SessionStart::Resumed {
                session_id: "session-1".to_owned()
            },
        ]
    );
}

#[tokio::test]
async fn an_agent_that_cannot_resume_a_session_says_so_instead_of_starting_over() {
    let plan = plan(TWO_TURNS);
    let store = Store::open_in_memory().await.expect("store");
    let agents = MockAgents::new(script()).without_resume();

    let id = {
        let mut run = Runtime::start(&store, &plan, &agents, "two-turns")
            .await
            .expect("start");
        for _ in 0..3 {
            run.step().await.expect("step");
        }
        run.id().clone()
    };

    let mut resumed = Runtime::resume(&store, &plan, &agents, id, InDoubt::Ask)
        .await
        .expect("resume");
    let error = resumed.run().await.expect_err("the agent cannot resume");

    // Quietly opening a fresh session would lose the conversation and look like
    // it worked, which is the worst of both.
    assert!(matches!(error, RuntimeError::Agent(_)), "{error}");
}

#[tokio::test]
async fn a_node_that_died_between_entering_and_leaving_is_not_run_again_unasked() {
    let plan = plan(TWO_TURNS);
    let store = Store::open_in_memory().await.expect("store");

    // A script with no replies: the prompt node fails after it was entered, so
    // the journal cannot say whether the agent did the work or not.
    let starved = MockAgents::new(Script::new());
    let id = {
        let mut run = Runtime::start(&store, &plan, &starved, "two-turns")
            .await
            .expect("start");
        run.run().await.expect_err("the script has no reply");
        run.id().clone()
    };
    assert_eq!(
        store.status(&id).await.expect("status"),
        RunStatus::Suspended
    );

    let agents = MockAgents::new(script());
    let refused = Runtime::resume(&store, &plan, &agents, id.clone(), InDoubt::Ask).await;
    assert!(
        matches!(refused, Err(RuntimeError::InDoubt { .. })),
        "a prompt turn may have been paid for already, so nobody but a human can decide"
    );

    let mut resumed = Runtime::resume(&store, &plan, &agents, id.clone(), InDoubt::Retry)
        .await
        .expect("told to retry");
    resumed.run().await.expect("finish");
    assert_eq!(printed(&store, &id).await, ["one", "two"]);
}

#[tokio::test]
async fn a_node_that_is_safe_to_repeat_is_repeated_without_asking_and_does_happen_twice() {
    let plan = plan(TWO_TURNS);
    let store = Store::open_in_memory().await.expect("store");
    let agents = MockAgents::new(script());

    let id = {
        let mut run = Runtime::start(&store, &plan, &agents, "two-turns")
            .await
            .expect("start");
        // start, agent, sess, first: `say_first` is next.
        for _ in 0..4 {
            run.step().await.expect("step");
        }
        run.id().clone()
    };

    // Hand-written because no node can be made to die inside on demand. The
    // journal is the interface a resumed run reads, so writing the state
    // directly is the honest way to reach it: entered, the effect happened, and
    // then nothing — which is exactly what a kill mid-node leaves.
    for (kind, payload) in [
        ("node.entered", json!({ "node": "say_first" })),
        ("print", json!({ "text": "one" })),
    ] {
        store.append(&id, kind, &payload).await.expect("append");
    }

    let mut resumed = Runtime::resume(&store, &plan, &agents, id.clone(), InDoubt::Ask)
        .await
        .expect("printing is safe to repeat, so nobody is asked");
    resumed.run().await.expect("finish");

    // And this is what `Recovery::Retry` costs, said out loud: the effect really
    // does happen a second time. It is chosen for nodes where that is cheap, not
    // for nodes where it is invisible.
    assert_eq!(printed(&store, &id).await, ["one", "one", "two"]);
}