openquiver-runtime 0.1.0

//! What each node type actually does.
//!
//! A match over type ids rather than a trait with six implementations: there are
//! six of them and they fit on one screen. The trait arrives with the first node
//! that comes from outside this crate — a plugin — and inventing it now would
//! only guess at what that boundary needs.

use std::collections::BTreeMap;

use openquiver_acp::{Agents, Session, SessionHandle};
use openquiver_compile::CompiledNode;
use openquiver_graph::NodeName;

use crate::journal::{Printed, Record};
use crate::{RunValue, RuntimeError};

/// The world a node is allowed to touch.
pub(crate) struct Effects<'a> {
    pub agents: &'a dyn Agents,
    /// Live sessions, by agent *and* session id. Empty after a restart, which is
    /// exactly when a handle from the journal has to be turned back into a
    /// conversation.
    ///
    /// A session id is unique within one agent and says nothing across agents —
    /// two of them numbering their sessions from one is the ordinary case. Keyed
    /// on the id alone, a second agent's first session displaces the first
    /// agent's, and a prompt wired to one conversation is served by the other:
    /// a fresh context that answers just as happily, having forgotten
    /// everything, while the run reports success.
    pub sessions: BTreeMap<(String, String), Box<dyn Session>>,
}

/// How a session is found again in a process that already has it.
fn key(handle: &SessionHandle) -> (String, String) {
    (handle.agent.clone(), handle.session_id.clone())
}

impl Effects<'_> {
    /// The live session for a handle, picking it up again if this process does
    /// not have it.
    async fn session(&mut self, handle: &SessionHandle) -> Result<&mut dyn Session, RuntimeError> {
        let key = key(handle);
        if !self.sessions.contains_key(&key) {
            let session = self.agents.resume_session(handle).await?;
            self.sessions.insert(key.clone(), session);
        }
        Ok(self.sessions.get_mut(&key).expect("just inserted").as_mut())
    }
}

/// What running a node produced.
pub(crate) struct Outcome {
    pub outputs: BTreeMap<String, RunValue>,
    /// Which execution output continues. `None` ends this line of execution.
    pub exec: Option<&'static str>,
    /// Anything else worth recording, in the order it happened.
    pub records: Vec<Record>,
}

impl Outcome {
    fn goes_to(pin: &'static str) -> Self {
        Self {
            outputs: BTreeMap::new(),
            exec: Some(pin),
            records: Vec::new(),
        }
    }

    fn with_output(mut self, pin: &str, value: RunValue) -> Self {
        self.outputs.insert(pin.to_owned(), value);
        self
    }

    fn with_record(mut self, record: Record) -> Self {
        self.records.push(record);
        self
    }
}

pub(crate) async fn perform(
    effects: &mut Effects<'_>,
    name: &NodeName,
    node: &CompiledNode,
    inputs: &BTreeMap<String, RunValue>,
) -> Result<Outcome, RuntimeError> {
    let at = Pins { name, inputs };

    Ok(match node.type_id.as_str() {
        "event.manual" => Outcome::goes_to("out"),

        // Connecting is deferred to the session: an ACP agent process is started
        // when a conversation needs it, and pooled by command. This node names
        // the agent and checks it exists in the manifest; nothing is spawned by
        // holding a name.
        "acp.connect_agent" => {
            Outcome::goes_to("out").with_output("agent", RunValue::Agent(at.text("agent")?))
        }

        "acp.new_session" => {
            let agent = at.agent("agent")?;
            let cwd = at.optional_text("cwd")?.unwrap_or_else(|| ".".to_owned());
            let session = effects.agents.new_session(&agent, &cwd).await?;
            let handle = session.handle();
            effects.sessions.insert(key(&handle), session);
            Outcome::goes_to("out").with_output("session", RunValue::Session(handle))
        }

        "acp.prompt" => {
            let handle = at.session("session")?;
            let text = at.text("text")?;
            let turn = effects.session(&handle).await?.prompt(&text).await?;
            // The stop reason chooses the pin, which is the entire reason there
            // are four of them: a refused turn must not continue as an answered
            // one.
            Outcome::goes_to(turn.stop_reason.pin()).with_output("text", RunValue::Text(turn.text))
        }

        "io.print" => Outcome::goes_to("out").with_record(Record::Printed(Printed {
            text: at.text("value")?,
        })),

        "flow.branch" => Outcome::goes_to(if at.bool("condition")? {
            "true"
        } else {
            "false"
        }),

        other => return Err(RuntimeError::NoSuchNodeType(other.to_owned())),
    })
}

/// One node's resolved inputs, with the node's name for the error messages.
struct Pins<'a> {
    name: &'a NodeName,
    inputs: &'a BTreeMap<String, RunValue>,
}

impl Pins<'_> {
    fn get(&self, pin: &str) -> Result<&RunValue, RuntimeError> {
        self.inputs.get(pin).ok_or_else(|| RuntimeError::NoInput {
            node: self.name.as_str().to_owned(),
            pin: pin.to_owned(),
        })
    }

    fn wrong(&self, pin: &str, expected: &'static str) -> RuntimeError {
        RuntimeError::WrongValue {
            node: self.name.as_str().to_owned(),
            pin: pin.to_owned(),
            expected,
        }
    }

    fn text(&self, pin: &str) -> Result<String, RuntimeError> {
        self.get(pin)?
            .as_text()
            .ok_or_else(|| self.wrong(pin, "text"))
    }

    fn optional_text(&self, pin: &str) -> Result<Option<String>, RuntimeError> {
        match self.inputs.get(pin) {
            None => Ok(None),
            Some(value) => value
                .as_text()
                .map(Some)
                .ok_or_else(|| self.wrong(pin, "text")),
        }
    }

    fn bool(&self, pin: &str) -> Result<bool, RuntimeError> {
        self.get(pin)?
            .as_bool()
            .ok_or_else(|| self.wrong(pin, "true or false"))
    }

    fn agent(&self, pin: &str) -> Result<String, RuntimeError> {
        self.get(pin)?
            .as_agent()
            .map(str::to_owned)
            .ok_or_else(|| self.wrong(pin, "a connected agent"))
    }

    fn session(&self, pin: &str) -> Result<SessionHandle, RuntimeError> {
        self.get(pin)?
            .as_session()
            .cloned()
            .ok_or_else(|| self.wrong(pin, "an open session"))
    }
}