ai-app/server/src/session/echo.rs

//! The phase-1 fake driver: no child process, just events. It exists to
//! prove the whole pipe -- spawn, transcript, SSE cursors, questions,
//! interrupts -- before any AI is involved, and stays useful afterwards as
//! a connectivity check that costs no tokens.
//!
//! Behavior: every message is echoed back as a few streamed text deltas. A
//! message starting with `/tool` also emits a fake tool run, and one
//! starting with `/question` asks one (exercising the answer path). This is
//! exactly the event vocabulary the real drivers produce, so a UI that
//! renders echo sessions correctly renders the real thing.

use std::sync::Mutex;
use std::time::Duration;

use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus};

/// Delay between streamed deltas -- long enough that streaming is visibly
/// streaming in the UI, short enough that tests waiting on a full turn
/// stay fast.
const DELTA_DELAY: Duration = Duration::from_millis(50);

pub struct EchoDriver {
    sink: EventSink,
    /// Id of the question currently awaiting an answer, if any. One at a
    /// time is all the echo behavior ever produces.
    pending_question: Mutex<Option<String>>,
}

impl EchoDriver {
    pub fn new(sink: EventSink) -> Self {
        let driver = Self {
            sink,
            pending_question: Mutex::new(None),
        };
        driver.emit(Event::Status {
            state: SessionStatus::Idle,
        });
        driver
    }

    /// Sends are infallible from the driver's point of view: a closed sink
    /// means the session is being torn down, and there is nobody left to
    /// report to.
    fn emit(&self, event: Event) {
        let _ = self.sink.send(event);
    }
}

impl Driver for EchoDriver {
    fn send_user_message(&self, text: String, _images: Vec<ImageRef>) {
        let sink = self.sink.clone();

        if let Some(rest) = text.strip_prefix("/question") {
            let id = format!("q-{}", super::random_hex());
            let prompt = if rest.trim().is_empty() {
                "Echo asks: proceed?".to_string()
            } else {
                format!("Echo asks: {}", rest.trim())
            };
            *self.pending_question.lock().unwrap() = Some(id.clone());
            self.emit(Event::Status {
                state: SessionStatus::Running,
            });
            self.emit(Event::Question {
                id,
                prompt,
                options: vec!["Yes".to_string(), "No".to_string()],
            });
            self.emit(Event::Status {
                state: SessionStatus::AwaitingInput,
            });
            return;
        }

        let run_tool = text
            .strip_prefix("/tool")
            .map(|rest| rest.trim().to_string());
        tokio::spawn(async move {
            let send = |event: Event| {
                let _ = sink.send(event);
            };
            send(Event::Status {
                state: SessionStatus::Running,
            });

            if let Some(input) = run_tool {
                let id = format!("t-{}", super::random_hex());
                send(Event::ToolStart {
                    id: id.clone(),
                    tool: "echo-tool".to_string(),
                    input: serde_json::json!({ "input": input }),
                });
                tokio::time::sleep(DELTA_DELAY).await;
                send(Event::ToolUpdate {
                    id: id.clone(),
                    output: "working...".to_string(),
                });
                tokio::time::sleep(DELTA_DELAY).await;
                send(Event::ToolEnd {
                    id,
                    output: format!("echoed: {input}"),
                });
            }

            // Word-at-a-time so streaming is visibly streaming.
            for word in format!("You said: {text}").split_inclusive(' ') {
                send(Event::AssistantText {
                    delta: word.to_string(),
                });
                tokio::time::sleep(DELTA_DELAY).await;
            }
            send(Event::UsageDelta {
                tokens: text.split_whitespace().count() as u64,
            });
            send(Event::Status {
                state: SessionStatus::Idle,
            });
        });
    }

    fn answer_question(&self, id: &str, answer: &str) {
        let mut pending = self.pending_question.lock().unwrap();
        match pending.as_deref() {
            Some(expected) if expected == id => {
                *pending = None;
                self.emit(Event::AssistantText {
                    delta: format!("You answered: {answer}"),
                });
                self.emit(Event::Status {
                    state: SessionStatus::Idle,
                });
            }
            _ => self.emit(Event::Error {
                message: format!("no question {id} is awaiting an answer"),
            }),
        }
    }

    fn interrupt(&self) {
        // Nothing real to stop; a pending question is abandoned so the
        // session isn't stuck awaiting input forever.
        *self.pending_question.lock().unwrap() = None;
        self.emit(Event::Status {
            state: SessionStatus::Idle,
        });
    }

    fn set_model(&self, model: &str) {
        self.emit(Event::Error {
            message: format!("echo sessions have no model to change to {model}"),
        });
    }

    fn compact(&self) {
        self.emit(Event::Error {
            message: "echo sessions have nothing to compact".to_string(),
        });
    }

    fn shutdown(&self) {
        self.emit(Event::Status {
            state: SessionStatus::Exited,
        });
    }
}