Implement SimpleBot native client (#86)

programs/src/bin/simple_bot.rs

@@ -1,3 +1,3 @@
 fn main() -> std::process::ExitCode {
-    libremetaverse_programs::pending_program("SimpleBot")
+    libremetaverse_programs::simple_bot::main_entry()
 }

programs/src/lib.rs

@@ -2,5 +2,6 @@
 
 pub mod commands;
 pub mod osd_inspector;
+pub mod simple_bot;
 
 pub use libremetaverse::shim::pending_program;

programs/src/simple_bot.rs

@@ -0,0 +1,964 @@
+//! Async, cancellation-safe native implementation of the `SimpleBot` example.
+
+use clap::Parser;
+use libremetaverse::types::compat::{CancellationToken, CancellationTokenSource, Subscription};
+use libremetaverse::types::{UUID, Vector3};
+use libremetaverse::{
+    AgentManager, Animations, ChatEventArgs, ChatType, DisconnectedEventArgs, GridClient,
+    InstantMessageEventArgs, LoginProgressEventArgs, NetworkManager,
+};
+use std::fmt;
+use std::fs::File;
+use std::io::{self, BufRead, BufReader, Read, Write};
+use std::path::{Path, PathBuf};
+use std::process::ExitCode;
+use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
+use std::sync::{Arc, Mutex};
+use std::time::Duration;
+use tokio::sync::mpsc;
+use tokio::task::JoinSet;
+
+pub const EXIT_SUCCESS: u8 = 0;
+pub const EXIT_USAGE: u8 = 2;
+pub const EXIT_INPUT: u8 = 3;
+pub const EXIT_CLIENT: u8 = 4;
+
+const MAX_SCRIPT_BYTES: u64 = 1024 * 1024;
+const MAX_SCRIPT_EVENTS: usize = 1024;
+const MAX_NAME_BYTES: usize = 256;
+const MAX_MESSAGE_BYTES: usize = 4096;
+const EVENT_QUEUE_CAPACITY: usize = 256;
+const JUMP_DURATION: Duration = Duration::from_millis(500);
+const GREETING_DELAY: Duration = Duration::from_millis(750);
+const BOT_ID_TEXT: &str = "11111111-2222-3333-4444-555555555555";
+const FAKE_REGION: &str = "Scripted Test Region";
+
+#[derive(Parser)]
+#[command(
+    name = "simple-bot",
+    version,
+    about = "Run the native LibreMetaverse SimpleBot",
+    long_about = None,
+    arg_required_else_help = true,
+    after_help = "Commands via IM:\n  help, ?             Show available commands\n  where, location     Report current location\n  sit                  Sit on the ground\n  stand                Stand up\n  dance                Start the DANCE1 animation\n  fly                  Start flying\n  walk                 Stop flying\n  jump                 Jump briefly\n  hello, hi, hey       Return a greeting"
+)]
+struct Cli {
+    /// Avatar first name. May also be supplied as `GRID_FIRST_NAME`.
+    #[arg(value_name = "FIRSTNAME")]
+    first_name: Option<String>,
+
+    /// Avatar last name. May also be supplied as `GRID_LAST_NAME`.
+    #[arg(value_name = "LASTNAME")]
+    last_name: Option<String>,
+
+    /// Avatar password. May also be supplied as `GRID_PASSWORD`.
+    #[arg(value_name = "PASSWORD")]
+    password: Option<String>,
+
+    /// Run a bounded tab-separated fake-grid conversation instead of logging in.
+    #[arg(long, value_name = "FILE")]
+    fake_script: Option<PathBuf>,
+
+    /// Override the login endpoint. `GRID_LOGIN_URL` is used when this is absent.
+    #[arg(long, value_name = "URL")]
+    login_uri: Option<String>,
+
+    /// Maximum time allowed for login.
+    #[arg(long, default_value_t = 30, value_name = "SECONDS", value_parser = clap::value_parser!(u64).range(1..=300))]
+    login_timeout_seconds: u64,
+}
+
+struct Credentials {
+    first_name: String,
+    last_name: String,
+    password: String,
+    login_uri: Option<String>,
+    login_timeout: Duration,
+}
+
+#[derive(Debug)]
+enum ProgramError {
+    Usage(&'static str),
+    Io { action: String, source: io::Error },
+    InvalidScript { line: usize, reason: &'static str },
+    Client(&'static str),
+    LoginFailed,
+    LoginTimedOut,
+    Signal,
+}
+
+impl ProgramError {
+    const fn exit_code(&self) -> u8 {
+        match self {
+            Self::Usage(_) => EXIT_USAGE,
+            Self::Io { .. } | Self::InvalidScript { .. } => EXIT_INPUT,
+            Self::Client(_) | Self::LoginFailed | Self::LoginTimedOut | Self::Signal => EXIT_CLIENT,
+        }
+    }
+}
+
+impl fmt::Display for ProgramError {
+    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Usage(message) => formatter.write_str(message),
+            Self::Io { action, source } => write!(formatter, "{action}: {source}"),
+            Self::InvalidScript { line, reason } => {
+                write!(
+                    formatter,
+                    "invalid fake-grid script at line {line}: {reason}"
+                )
+            }
+            Self::Client(operation) => {
+                write!(formatter, "native client operation failed: {operation}")
+            }
+            Self::LoginFailed => formatter.write_str("login failed"),
+            Self::LoginTimedOut => formatter.write_str("login timed out"),
+            Self::Signal => formatter.write_str("could not install the Ctrl-C handler"),
+        }
+    }
+}
+
+trait BotLogger: Send + Sync + 'static {
+    fn line(&self, line: String);
+}
+
+struct ConsoleLogger;
+
+impl BotLogger for ConsoleLogger {
+    fn line(&self, line: String) {
+        println!("{}", redact_text(&line));
+    }
+}
+
+#[derive(Default)]
+struct BufferedLogger {
+    lines: Mutex<Vec<String>>,
+}
+
+impl BufferedLogger {
+    fn finish(&self, output: &mut dyn Write) -> Result<(), ProgramError> {
+        for line in lock(&self.lines).iter() {
+            writeln!(output, "{line}").map_err(|source| ProgramError::Io {
+                action: "writing standard output".into(),
+                source,
+            })?;
+        }
+        Ok(())
+    }
+}
+
+impl BotLogger for BufferedLogger {
+    fn line(&self, line: String) {
+        lock(&self.lines).push(redact_text(&line));
+    }
+}
+
+trait BotBackend: Send + Sync + 'static {
+    fn agent_id(&self) -> UUID;
+    fn region_name(&self) -> String;
+    fn position(&self) -> Vector3;
+    fn instant_message(&self, target: UUID, message: String) -> Result<(), &'static str>;
+    fn local_chat(&self, message: String) -> Result<(), &'static str>;
+    fn sit_on_ground(&self) -> Result<(), &'static str>;
+    fn stand(&self) -> Result<(), &'static str>;
+    fn dance(&self) -> Result<(), &'static str>;
+    fn fly(&self, start: bool) -> Result<(), &'static str>;
+    fn jump(&self, start: bool) -> Result<(), &'static str>;
+}
+
+#[derive(Clone)]
+enum InboundEvent {
+    InstantMessage {
+        source_id: UUID,
+        source_name: String,
+        message: String,
+    },
+    LocalChat {
+        source_id: UUID,
+        source_name: String,
+        message: String,
+    },
+    Status(String),
+}
+
+enum DelayedAction {
+    ReleaseJump,
+    Greeting(String),
+}
+
+struct LiveGrid {
+    client: Mutex<GridClient>,
+    network: NetworkManager,
+    subscriptions: Mutex<Vec<Subscription>>,
+    shutdown: AtomicBool,
+    dropped_events: Arc<AtomicUsize>,
+}
+
+impl LiveGrid {
+    fn new(client: GridClient, network: NetworkManager) -> Self {
+        Self {
+            client: Mutex::new(client),
+            network,
+            subscriptions: Mutex::new(Vec::new()),
+            shutdown: AtomicBool::new(false),
+            dropped_events: Arc::new(AtomicUsize::new(0)),
+        }
+    }
+
+    fn install_subscriptions(self: &Arc<Self>, sender: &mpsc::Sender<InboundEvent>) {
+        let mut subscriptions = Vec::with_capacity(4);
+        let im_sender = sender.clone();
+        let dropped = Arc::clone(&self.dropped_events);
+        subscriptions.push(self.with_agent(|agent| {
+            agent.subscribe_im(Arc::new(move |event: InstantMessageEventArgs| {
+                let im = event.im();
+                send_event(
+                    &im_sender,
+                    InboundEvent::InstantMessage {
+                        source_id: im.from_agent_id,
+                        source_name: im.from_agent_name,
+                        message: im.message,
+                    },
+                    &dropped,
+                );
+            }))
+        }));
+
+        let chat_sender = sender.clone();
+        let dropped = Arc::clone(&self.dropped_events);
+        subscriptions.push(self.with_agent(|agent| {
+            agent.subscribe_chat_from_simulator(Arc::new(move |event: ChatEventArgs| {
+                send_event(
+                    &chat_sender,
+                    InboundEvent::LocalChat {
+                        source_id: event.source_id(),
+                        source_name: event.from_name(),
+                        message: event.message(),
+                    },
+                    &dropped,
+                );
+            }))
+        }));
+
+        let progress_sender = sender.clone();
+        let dropped = Arc::clone(&self.dropped_events);
+        subscriptions.push(self.network.subscribe_login_progress(Arc::new(
+            move |event: LoginProgressEventArgs| {
+                send_event(
+                    &progress_sender,
+                    InboundEvent::Status(format!(
+                        "Login {:?}: {}",
+                        event.status(),
+                        event.message()
+                    )),
+                    &dropped,
+                );
+            },
+        )));
+
+        let disconnected_sender = sender.clone();
+        let dropped = Arc::clone(&self.dropped_events);
+        subscriptions.push(self.network.subscribe_disconnected(Arc::new(
+            move |event: DisconnectedEventArgs| {
+                send_event(
+                    &disconnected_sender,
+                    InboundEvent::Status(format!(
+                        "Disconnected: {:?} - {}",
+                        event.reason(),
+                        event.message()
+                    )),
+                    &dropped,
+                );
+            },
+        )));
+        *lock(&self.subscriptions) = subscriptions;
+    }
+
+    fn with_agent<T>(&self, action: impl FnOnce(&mut AgentManager) -> T) -> T {
+        let mut client = lock(&self.client);
+        action(client.self_())
+    }
+
+    fn shutdown(&self) {
+        if self.shutdown.swap(true, Ordering::AcqRel) {
+            return;
+        }
+        lock(&self.subscriptions).clear();
+        let _ = self.network.logout_with_method();
+        let mut client = lock(&self.client);
+        let _ = client.self_().dispose();
+        let _ = client.dispose_with_method();
+    }
+}
+
+impl BotBackend for LiveGrid {
+    fn agent_id(&self) -> UUID {
+        self.with_agent(|agent| agent.agent_id())
+    }
+
+    fn region_name(&self) -> String {
+        self.network
+            .current_sim()
+            .map_or_else(|| "unknown".into(), |simulator| simulator.name.clone())
+    }
+
+    fn position(&self) -> Vector3 {
+        self.with_agent(|agent| agent.sim_position())
+    }
+
+    fn instant_message(&self, target: UUID, message: String) -> Result<(), &'static str> {
+        self.with_agent(|agent| agent.instant_message_with_uuid_string(target, message))
+            .map_err(|_| "send instant message")
+    }
+
+    fn local_chat(&self, message: String) -> Result<(), &'static str> {
+        self.with_agent(|agent| agent.chat(message, 0, ChatType::Normal, Some(false)))
+            .map_err(|_| "send local chat")
+    }
+
+    fn sit_on_ground(&self) -> Result<(), &'static str> {
+        self.with_agent(|agent| agent.sit_on_ground())
+            .map_err(|_| "sit on ground")
+    }
+
+    fn stand(&self) -> Result<(), &'static str> {
+        self.with_agent(|agent| agent.stand())
+            .map(|_| ())
+            .map_err(|_| "stand")
+    }
+
+    fn dance(&self) -> Result<(), &'static str> {
+        self.with_agent(|agent| agent.animation_start(Animations::dance1(), true))
+            .map_err(|_| "start DANCE1 animation")
+    }
+
+    fn fly(&self, start: bool) -> Result<(), &'static str> {
+        self.with_agent(|agent| agent.fly(start))
+            .map_err(|_| if start { "start flying" } else { "stop flying" })
+    }
+
+    fn jump(&self, start: bool) -> Result<(), &'static str> {
+        self.with_agent(|agent| agent.jump(start))
+            .map_err(|_| if start { "start jump" } else { "release jump" })
+    }
+}
+
+struct ScriptBackend {
+    calls: Mutex<Vec<String>>,
+    agent_id: UUID,
+}
+
+impl ScriptBackend {
+    fn new() -> Result<Self, ProgramError> {
+        Ok(Self {
+            calls: Mutex::new(Vec::new()),
+            agent_id: parse_uuid(BOT_ID_TEXT, 0)?,
+        })
+    }
+
+    fn record(&self, call: String) {
+        lock(&self.calls).push(call);
+    }
+
+    fn drain_calls(&self, logger: &dyn BotLogger) {
+        for call in std::mem::take(&mut *lock(&self.calls)) {
+            logger.line(call);
+        }
+    }
+}
+
+impl BotBackend for ScriptBackend {
+    fn agent_id(&self) -> UUID {
+        self.agent_id
+    }
+
+    fn region_name(&self) -> String {
+        FAKE_REGION.into()
+    }
+
+    fn position(&self) -> Vector3 {
+        Vector3 {
+            x: 128.0,
+            y: 64.0,
+            z: 25.0,
+        }
+    }
+
+    fn instant_message(&self, target: UUID, message: String) -> Result<(), &'static str> {
+        self.record(format!("CALL instant-message {target} {message}"));
+        Ok(())
+    }
+
+    fn local_chat(&self, message: String) -> Result<(), &'static str> {
+        self.record(format!("CALL chat channel=0 type=Normal {message}"));
+        Ok(())
+    }
+
+    fn sit_on_ground(&self) -> Result<(), &'static str> {
+        self.record("CALL sit-on-ground".into());
+        Ok(())
+    }
+
+    fn stand(&self) -> Result<(), &'static str> {
+        self.record("CALL stand".into());
+        Ok(())
+    }
+
+    fn dance(&self) -> Result<(), &'static str> {
+        self.record(format!(
+            "CALL animation-start {} reliable=true",
+            Animations::dance1()
+        ));
+        Ok(())
+    }
+
+    fn fly(&self, start: bool) -> Result<(), &'static str> {
+        self.record(format!("CALL fly {start}"));
+        Ok(())
+    }
+
+    fn jump(&self, start: bool) -> Result<(), &'static str> {
+        self.record(format!("CALL jump {start}"));
+        Ok(())
+    }
+}
+
+#[must_use]
+pub fn main_entry() -> ExitCode {
+    let cli = Cli::parse();
+    let Ok(runtime) = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .build()
+    else {
+        eprintln!("simple-bot: could not initialize the async runtime");
+        return ExitCode::from(EXIT_CLIENT);
+    };
+    match runtime.block_on(run(cli)) {
+        Ok(()) => ExitCode::from(EXIT_SUCCESS),
+        Err(error) => {
+            eprintln!("simple-bot: {error}");
+            ExitCode::from(error.exit_code())
+        }
+    }
+}
+
+async fn run(cli: Cli) -> Result<(), ProgramError> {
+    if let Some(script) = cli.fake_script {
+        if cli.first_name.is_some() || cli.last_name.is_some() || cli.password.is_some() {
+            return Err(ProgramError::Usage(
+                "credentials cannot be combined with --fake-script",
+            ));
+        }
+        let stdout = io::stdout();
+        return run_fake_script(&script, &mut stdout.lock());
+    }
+    let credentials = resolve_credentials(cli)?;
+    run_live(credentials).await
+}
+
+fn resolve_credentials(cli: Cli) -> Result<Credentials, ProgramError> {
+    let first_name = cli
+        .first_name
+        .or_else(|| std::env::var("GRID_FIRST_NAME").ok())
+        .filter(|value| !value.is_empty())
+        .ok_or(ProgramError::Usage(
+            "FIRSTNAME is required (or set GRID_FIRST_NAME)",
+        ))?;
+    let last_name = cli
+        .last_name
+        .or_else(|| std::env::var("GRID_LAST_NAME").ok())
+        .filter(|value| !value.is_empty())
+        .ok_or(ProgramError::Usage(
+            "LASTNAME is required (or set GRID_LAST_NAME)",
+        ))?;
+    let password = cli
+        .password
+        .or_else(|| std::env::var("GRID_PASSWORD").ok())
+        .filter(|value| !value.is_empty())
+        .ok_or(ProgramError::Usage(
+            "PASSWORD is required (or set GRID_PASSWORD)",
+        ))?;
+    let login_uri = cli
+        .login_uri
+        .or_else(|| std::env::var("GRID_LOGIN_URL").ok())
+        .filter(|value| !value.is_empty());
+    Ok(Credentials {
+        first_name,
+        last_name,
+        password,
+        login_uri,
+        login_timeout: Duration::from_secs(cli.login_timeout_seconds),
+    })
+}
+
+async fn run_live(mut credentials: Credentials) -> Result<(), ProgramError> {
+    let client = GridClient::new().map_err(|_| ProgramError::Client("construct GridClient"))?;
+    let network = client.network();
+    let mut login = network
+        .default_login_params(
+            std::mem::take(&mut credentials.first_name),
+            std::mem::take(&mut credentials.last_name),
+            std::mem::take(&mut credentials.password),
+            "SimpleBot".into(),
+            env!("CARGO_PKG_VERSION").into(),
+        )
+        .map_err(|_| ProgramError::Client("build login parameters"))?;
+    if let Some(uri) = credentials.login_uri.take() {
+        login.uri = uri;
+    }
+    let login_timeout = credentials.login_timeout;
+    let grid = Arc::new(LiveGrid::new(client, network.clone()));
+    let logger = Arc::new(ConsoleLogger);
+    let (sender, receiver) = mpsc::channel(EVENT_QUEUE_CAPACITY);
+    grid.install_subscriptions(&sender);
+    let cancellation = CancellationTokenSource::new();
+    let worker = tokio::spawn(event_loop(
+        Arc::clone(&grid),
+        Arc::clone(&logger),
+        receiver,
+        cancellation.token(),
+    ));
+
+    logger.line("Logging in...".into());
+    let login_cancellation = CancellationTokenSource::new();
+    let login_result = tokio::select! {
+        result = network.login_with_login_params_cancellation_token(
+            login,
+            Some(login_cancellation.token()),
+        ) => result.map(Some).map_err(|_| ProgramError::LoginFailed),
+        () = tokio::time::sleep(login_timeout) => {
+            login_cancellation.cancel();
+            let _ = network.abort_login();
+            Err(ProgramError::LoginTimedOut)
+        }
+        signal = tokio::signal::ctrl_c() => {
+            login_cancellation.cancel();
+            let _ = network.abort_login();
+            signal.map_err(|_| ProgramError::Signal)?;
+            Ok(None)
+        }
+    };
+
+    let success = match login_result {
+        Ok(Some(success)) => success,
+        Ok(None) => {
+            finish_live(grid, sender, cancellation, worker).await;
+            return Ok(());
+        }
+        Err(error) => {
+            finish_live(grid, sender, cancellation, worker).await;
+            return Err(error);
+        }
+    };
+    if !success {
+        finish_live(grid, sender, cancellation, worker).await;
+        return Err(ProgramError::LoginFailed);
+    }
+
+    logger.line(format!("Logged in to {}", grid.region_name()));
+    logger.line(format!("Position: {}", grid.position().to_string()));
+    logger.line("Bot is ready. Send an IM with 'help'; press Ctrl-C to logout.".into());
+    tokio::signal::ctrl_c()
+        .await
+        .map_err(|_| ProgramError::Signal)?;
+    finish_live(grid, sender, cancellation, worker).await;
+    Ok(())
+}
+
+async fn finish_live(
+    grid: Arc<LiveGrid>,
+    sender: mpsc::Sender<InboundEvent>,
+    cancellation: CancellationTokenSource,
+    worker: tokio::task::JoinHandle<()>,
+) {
+    cancellation.cancel();
+    drop(sender);
+    let _ = worker.await;
+    let dropped = grid.dropped_events.load(Ordering::Acquire);
+    grid.shutdown();
+    if dropped > 0 {
+        eprintln!("simple-bot: dropped {dropped} events because the bounded queue was full");
+    }
+}
+
+async fn event_loop<B: BotBackend, L: BotLogger>(
+    backend: Arc<B>,
+    logger: Arc<L>,
+    mut receiver: mpsc::Receiver<InboundEvent>,
+    cancellation: CancellationToken,
+) {
+    let mut delayed = JoinSet::new();
+    loop {
+        tokio::select! {
+            () = cancellation.cancelled() => break,
+            event = receiver.recv() => {
+                let Some(event) = event else { break };
+                if let Some(action) = handle_event(backend.as_ref(), logger.as_ref(), event) {
+                    spawn_delayed(
+                        &mut delayed,
+                        Arc::clone(&backend),
+                        Arc::clone(&logger),
+                        cancellation.clone(),
+                        action,
+                    );
+                }
+            }
+            result = delayed.join_next(), if !delayed.is_empty() => {
+                let _ = result;
+            }
+        }
+    }
+    receiver.close();
+    while delayed.join_next().await.is_some() {}
+}
+
+fn spawn_delayed<B: BotBackend, L: BotLogger>(
+    tasks: &mut JoinSet<()>,
+    backend: Arc<B>,
+    logger: Arc<L>,
+    cancellation: CancellationToken,
+    action: DelayedAction,
+) {
+    tasks.spawn(async move {
+        match action {
+            DelayedAction::ReleaseJump => {
+                tokio::select! {
+                    () = tokio::time::sleep(JUMP_DURATION) => {}
+                    () = cancellation.cancelled() => {}
+                }
+                if let Err(error) = backend.jump(false) {
+                    logger.line(format!("Error: {error}"));
+                }
+            }
+            DelayedAction::Greeting(message) => {
+                let send = tokio::select! {
+                    () = tokio::time::sleep(GREETING_DELAY) => true,
+                    () = cancellation.cancelled() => false,
+                };
+                if send && let Err(error) = backend.local_chat(message) {
+                    logger.line(format!("Error: {error}"));
+                }
+            }
+        }
+    });
+}
+
+fn handle_event(
+    backend: &dyn BotBackend,
+    logger: &dyn BotLogger,
+    event: InboundEvent,
+) -> Option<DelayedAction> {
+    match event {
+        InboundEvent::InstantMessage {
+            source_id,
+            source_name,
+            message,
+        } => handle_instant_message(backend, logger, source_id, &source_name, &message),
+        InboundEvent::LocalChat {
+            source_id,
+            source_name,
+            message,
+        } => handle_local_chat(backend, logger, source_id, &source_name, &message),
+        InboundEvent::Status(status) => {
+            logger.line(status);
+            None
+        }
+    }
+}
+
+fn handle_instant_message(
+    backend: &dyn BotBackend,
+    logger: &dyn BotLogger,
+    source_id: UUID,
+    source_name: &str,
+    message: &str,
+) -> Option<DelayedAction> {
+    if source_id == backend.agent_id() {
+        return None;
+    }
+    logger.line(format!("[IM] {source_name}: {message}"));
+    let command = message.trim().to_ascii_lowercase();
+    let (response, delayed) = match command.as_str() {
+        "help" | "?" => (
+            "Commands: help, where, sit, stand, dance, fly, walk, jump, hello, hi, hey".into(),
+            None,
+        ),
+        "where" | "location" => (
+            format!(
+                "I'm in {} at {}",
+                backend.region_name(),
+                backend.position().to_string()
+            ),
+            None,
+        ),
+        "sit" => action_response(backend.sit_on_ground(), "Sitting down..."),
+        "stand" => action_response(backend.stand(), "Standing up!"),
+        "dance" => action_response(backend.dance(), "Dancing!"),
+        "fly" => action_response(backend.fly(true), "Taking off!"),
+        "walk" => action_response(backend.fly(false), "Walking now."),
+        "jump" => match backend.jump(true) {
+            Ok(()) => ("Wheee!".into(), Some(DelayedAction::ReleaseJump)),
+            Err(error) => (format!("Error: {error}"), None),
+        },
+        "hello" | "hi" | "hey" => (format!("Hello, {source_name}!"), None),
+        _ => (
+            "I don't understand that command. Try 'help' for a list of commands.".into(),
+            None,
+        ),
+    };
+    match backend.instant_message(source_id, response.clone()) {
+        Ok(()) => logger.line(format!("[IM] -> {source_name}: {response}")),
+        Err(error) => logger.line(format!("Error: {error}")),
+    }
+    delayed
+}
+
+fn action_response(
+    result: Result<(), &'static str>,
+    success: &'static str,
+) -> (String, Option<DelayedAction>) {
+    match result {
+        Ok(()) => (success.into(), None),
+        Err(error) => (format!("Error: {error}"), None),
+    }
+}
+
+fn handle_local_chat(
+    backend: &dyn BotBackend,
+    logger: &dyn BotLogger,
+    source_id: UUID,
+    source_name: &str,
+    message: &str,
+) -> Option<DelayedAction> {
+    if source_id == backend.agent_id() {
+        return None;
+    }
+    logger.line(format!("[Chat] {source_name}: {message}"));
+    let normalized = message.to_ascii_lowercase();
+    (normalized.contains("hello") || normalized.contains("hi "))
+        .then(|| DelayedAction::Greeting(format!("Hello, {source_name}!")))
+}
+
+fn run_fake_script(path: &Path, output: &mut dyn Write) -> Result<(), ProgramError> {
+    let backend = ScriptBackend::new()?;
+    let logger = BufferedLogger::default();
+    logger.line(format!(
+        "Fake grid ready: {} at {}",
+        backend.region_name(),
+        backend.position().to_string()
+    ));
+    let events = read_script(path)?;
+    for event in events {
+        let delayed = handle_event(&backend, &logger, event);
+        backend.drain_calls(&logger);
+        match delayed {
+            Some(DelayedAction::ReleaseJump) => {
+                logger.line(format!("WAIT {}ms", JUMP_DURATION.as_millis()));
+                if let Err(error) = backend.jump(false) {
+                    logger.line(format!("Error: {error}"));
+                }
+            }
+            Some(DelayedAction::Greeting(message)) => {
+                logger.line(format!("WAIT {}ms", GREETING_DELAY.as_millis()));
+                if let Err(error) = backend.local_chat(message) {
+                    logger.line(format!("Error: {error}"));
+                }
+            }
+            None => {}
+        }
+        backend.drain_calls(&logger);
+    }
+    logger.line("Fake grid logout complete; active_tasks=0 open_sockets=0".into());
+    logger.finish(output)
+}
+
+fn read_script(path: &Path) -> Result<Vec<InboundEvent>, ProgramError> {
+    let file = File::open(path).map_err(|source| ProgramError::Io {
+        action: format!("opening fake-grid script {}", path.display()),
+        source,
+    })?;
+    if file.metadata().map_or(0, |metadata| metadata.len()) > MAX_SCRIPT_BYTES {
+        return Err(ProgramError::InvalidScript {
+            line: 0,
+            reason: "script exceeds the 1 MiB limit",
+        });
+    }
+    let mut bytes = Vec::new();
+    BufReader::new(file)
+        .take(MAX_SCRIPT_BYTES + 1)
+        .read_to_end(&mut bytes)
+        .map_err(|source| ProgramError::Io {
+            action: format!("reading fake-grid script {}", path.display()),
+            source,
+        })?;
+    if bytes.len() as u64 > MAX_SCRIPT_BYTES {
+        return Err(ProgramError::InvalidScript {
+            line: 0,
+            reason: "script exceeds the 1 MiB limit",
+        });
+    }
+    let mut events = Vec::new();
+    for (index, line) in BufReader::new(bytes.as_slice()).lines().enumerate() {
+        let line_number = index + 1;
+        let line = line.map_err(|source| ProgramError::Io {
+            action: format!("reading fake-grid script {}", path.display()),
+            source,
+        })?;
+        let line = line.trim_end_matches('\r');
+        if line.is_empty() || line.starts_with('#') {
+            continue;
+        }
+        if events.len() >= MAX_SCRIPT_EVENTS {
+            return Err(ProgramError::InvalidScript {
+                line: line_number,
+                reason: "script contains more than 1024 events",
+            });
+        }
+        events.push(parse_script_line(line, line_number)?);
+    }
+    Ok(events)
+}
+
+fn parse_script_line(line: &str, line_number: usize) -> Result<InboundEvent, ProgramError> {
+    let fields = line.splitn(4, '\t').collect::<Vec<_>>();
+    match fields.as_slice() {
+        [kind @ ("im" | "chat"), source, name, message] => {
+            if name.is_empty() || name.len() > MAX_NAME_BYTES {
+                return Err(ProgramError::InvalidScript {
+                    line: line_number,
+                    reason: "source name is empty or too long",
+                });
+            }
+            if message.len() > MAX_MESSAGE_BYTES {
+                return Err(ProgramError::InvalidScript {
+                    line: line_number,
+                    reason: "message is too long",
+                });
+            }
+            let source_id = parse_uuid(source, line_number)?;
+            if *kind == "im" {
+                Ok(InboundEvent::InstantMessage {
+                    source_id,
+                    source_name: (*name).into(),
+                    message: (*message).into(),
+                })
+            } else {
+                Ok(InboundEvent::LocalChat {
+                    source_id,
+                    source_name: (*name).into(),
+                    message: (*message).into(),
+                })
+            }
+        }
+        ["status", message] if message.len() <= MAX_MESSAGE_BYTES => {
+            Ok(InboundEvent::Status((*message).into()))
+        }
+        _ => Err(ProgramError::InvalidScript {
+            line: line_number,
+            reason: "expected im/chat<TAB>uuid<TAB>name<TAB>message or status<TAB>message",
+        }),
+    }
+}
+
+fn parse_uuid(value: &str, line: usize) -> Result<UUID, ProgramError> {
+    UUID::new_with_string(value.into()).map_err(|_| ProgramError::InvalidScript {
+        line,
+        reason: "source UUID is invalid",
+    })
+}
+
+fn send_event(sender: &mpsc::Sender<InboundEvent>, event: InboundEvent, dropped: &AtomicUsize) {
+    if sender.try_send(event).is_err() {
+        dropped.fetch_add(1, Ordering::Relaxed);
+    }
+}
+
+fn redact_text(value: &str) -> String {
+    let lowercase = value.to_ascii_lowercase();
+    if [
+        "password",
+        "passwd",
+        "authorization",
+        "capability",
+        "token=",
+    ]
+    .iter()
+    .any(|marker| lowercase.contains(marker))
+    {
+        return "<redacted>".into();
+    }
+    value
+        .split_whitespace()
+        .map(|word| {
+            if word.contains("://") {
+                "<redacted-url>"
+            } else {
+                word
+            }
+        })
+        .collect::<Vec<_>>()
+        .join(" ")
+}
+
+fn lock<T>(value: &Mutex<T>) -> std::sync::MutexGuard<'_, T> {
+    value
+        .lock()
+        .unwrap_or_else(std::sync::PoisonError::into_inner)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn cancellation_releases_jump_and_joins_delayed_tasks() {
+        let backend = Arc::new(ScriptBackend::new().expect("script backend"));
+        let logger = Arc::new(BufferedLogger::default());
+        let cancellation = CancellationTokenSource::new();
+        let (sender, receiver) = mpsc::channel(EVENT_QUEUE_CAPACITY);
+        let source = UUID::new_with_string("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee".into())
+            .expect("source UUID");
+        let worker = tokio::spawn(event_loop(
+            Arc::clone(&backend),
+            logger,
+            receiver,
+            cancellation.token(),
+        ));
+        sender
+            .send(InboundEvent::InstantMessage {
+                source_id: source,
+                source_name: "Test Resident".into(),
+                message: "jump".into(),
+            })
+            .await
+            .expect("queue jump");
+        for _ in 0..100 {
+            if lock(&backend.calls)
+                .iter()
+                .any(|call| call == "CALL jump true")
+            {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+        cancellation.cancel();
+        drop(sender);
+        tokio::time::timeout(Duration::from_secs(1), worker)
+            .await
+            .expect("worker stopped before timeout")
+            .expect("worker did not panic");
+        let calls = lock(&backend.calls);
+        assert!(calls.iter().any(|call| call == "CALL jump true"));
+        assert!(calls.iter().any(|call| call == "CALL jump false"));
+    }
+
+    #[test]
+    fn redaction_hides_credentials_capabilities_and_urls() {
+        assert_eq!(redact_text("password=hunter2"), "<redacted>");
+        assert_eq!(redact_text("token=secret"), "<redacted>");
+        assert_eq!(
+            redact_text("endpoint https://example.invalid/cap/secret ready"),
+            "endpoint <redacted-url> ready"
+        );
+    }
+}