johnandersen777.bsky.social/asyncai

Switch repository

Loading…

use std::io::Write;
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use clap::{ArgAction, Args, Parser, Subcommand};
use serde_json::{json, Value};
use tokio::io::{AsyncBufReadExt, BufReader};
use tokio::task::JoinHandle;

use crate::atirc::{
    format_reply, is_no_reply, mentions, synthetic_sub_command, AtircClient, AtircEvent,
    ChannelMessage, ConnectionInfo, LibraryAtircClient,
};
use crate::backend::{AgentBackend, MockBackend};
use crate::code_mode::{
    CodeModeLimits, CodeModeOptions, CodeModeToolExecutor, SandboxKind, WatchLimits, WatchNotice,
};
use crate::error::AgiError;
use crate::kv::MemoryKvStore;
use crate::mcp::{McpToolExecutor, StdioServerConfig};
use crate::monitor::MonitorRunner;
use crate::openai_backend::OpenAiCompatibleBackend;
use crate::orchestrator::{Notice, Orchestrator, OrchestratorConfig, OrchestratorHandle};
use crate::output::OutputMessage;
use crate::session::{ClaudeCodeCliHarness, MonitorEvent, SessionHarness, SessionLogger};
use crate::tools::{ToolExecutor, ToolObserver};
use uuid::Uuid;

#[derive(Debug, Parser)]
#[command(
    name = "asyncai",
    version,
    about = "Tokio port of the core agi.py event loop"
)]
pub struct Cli {
    /// Model name for an OpenAI-compatible endpoint
    #[arg(long, global = true)]
    pub model: Option<String>,

    /// Agent name; defaults to the atirc connection nick or "alice"
    #[arg(long = "agent-name", global = true)]
    pub agent_name: Option<String>,

    /// Base URL for an OpenAI-compatible endpoint
    #[arg(long = "base-url", global = true)]
    pub base_url: Option<String>,

    /// API key for the OpenAI-compatible endpoint
    #[arg(long = "api-key", global = true)]
    pub api_key: Option<String>,

    /// Stdio MCP server command; repeat for multiple servers
    #[arg(long = "mcp-stdio", global = true, value_name = "COMMAND")]
    pub mcp_stdio: Vec<String>,

    /// Where run_code Python runs: auto (pool if a monty worker is found), in-process, pool
    #[arg(long = "code-sandbox", global = true, default_value = "auto")]
    pub code_sandbox: SandboxKind,

    /// monty worker binary for the pool sandbox [env: MONTY_WORKER; default: monty on PATH]
    #[arg(long = "monty-worker", global = true, value_name = "PATH")]
    pub monty_worker: Option<PathBuf>,

    /// Heap cap per pool worker, in bytes
    #[arg(
        long = "code-max-memory",
        global = true,
        value_name = "BYTES",
        default_value_t = 4 * 1024 * 1024 * 1024
    )]
    pub code_max_memory: usize,

    /// Python run in every new run_code scope after the built-in helpers
    #[arg(long = "code-preamble", global = true, value_name = "PATH")]
    pub code_preamble: Option<PathBuf>,

    /// Max tokens the model may generate per request
    #[arg(long = "max-output-tokens", global = true, value_name = "N", default_value_t = crate::openai_backend::DEFAULT_MAX_OUTPUT_TOKENS)]
    pub max_output_tokens: u32,

    /// Wall-clock limit per model request, in seconds
    #[arg(long = "model-timeout", global = true, value_name = "SECS", default_value_t = crate::openai_backend::DEFAULT_REQUEST_TIMEOUT.as_secs())]
    pub model_timeout: u64,

    /// JSON file that keeps watch specs across restarts (written on watch start/stop)
    #[arg(long = "watch-state", global = true, value_name = "PATH")]
    pub watch_state: Option<PathBuf>,

    /// Markdown prompt file; {agent_name} is substituted
    #[arg(
        long = "prompt-file",
        global = true,
        value_name = "PATH",
        conflicts_with = "prompt"
    )]
    pub prompt_file: Option<String>,

    /// Built-in prompt: atirc, interactive, atirc-socialweb-computer
    #[arg(long = "prompt", global = true, value_name = "NAME")]
    pub prompt: Option<String>,

    /// Per-format session root, FORMAT=PATH; repeatable
    #[arg(long = "session-dir", global = true, value_name = "FORMAT=PATH")]
    pub session_dir: Vec<String>,

    /// Session harness format; repeatable; default claude-code-cli
    #[arg(long = "session-format", global = true, value_name = "FORMAT")]
    pub session_format: Vec<String>,

    /// Disable session transcript writing
    #[arg(long = "no-session", global = true)]
    pub no_session: bool,

    /// Do not read stdin; run monitor commands only
    #[arg(long = "no-stdin", global = true)]
    pub no_stdin: bool,

    /// Monitor command; stdout lines become task notifications in session files; repeatable
    #[arg(long = "monitor", global = true, value_name = "COMMAND")]
    pub monitor: Vec<String>,

    #[command(subcommand)]
    pub command: Option<Command>,
}

#[derive(Debug, Subcommand)]
pub enum Command {
    /// Subscribe to atirc channels and reply through the atirc library
    Atirc(AtircArgs),
}

#[derive(Debug, Args)]
pub struct AtircArgs {
    /// Long-lived atirc token [env: ATIRC_TOKEN]
    #[arg(short = 't', long, value_name = "TOKEN")]
    pub token: Option<String>,

    /// Channel filter; repeat for multiple channels; default: account handle
    #[arg(
        short = 'c',
        long = "channel",
        value_name = "CHANNEL",
        action = ArgAction::Append
    )]
    pub channels: Vec<String>,

    /// Last sequence already seen; -1 = only new messages
    #[arg(long, default_value_t = -1, allow_hyphen_values = true)]
    pub cursor: i64,

    /// Stop after N published replies; 0 means unlimited
    #[arg(long, default_value_t = 0)]
    pub max_turns: usize,

    /// Only handle messages that mention the agent nick; default: all messages
    #[arg(long)]
    pub mention_only: bool,

    /// Log connection, inbound messages, publishes, and output records to stderr
    #[arg(long)]
    pub verbose: bool,
}

pub async fn run(mut cli: Cli) -> Result<(), Box<dyn std::error::Error>> {
    let verbose = matches!(&cli.command, Some(Command::Atirc(args)) if args.verbose);
    let tools = connect_tool_executor(&cli, verbose).await?;

    match cli.command.take() {
        Some(Command::Atirc(args)) => {
            let token = resolve_atirc_token(&cli, &args).await?;
            let client =
                Arc::new(LibraryAtircClient::from_token(token).map_err(std::io::Error::other)?);
            run_atirc(cli, args, tools, client).await
        }
        None => run_interactive(cli, tools).await,
    }
}

async fn resolve_atirc_token(
    cli: &Cli,
    args: &AtircArgs,
) -> Result<String, Box<dyn std::error::Error>> {
    if let Some(token) = args.token.clone().or_else(|| env_first(&["ATIRC_TOKEN"])) {
        return Ok(token);
    }

    let username = env_first(&["ATP_USERNAME", "ATIRC_USERNAME"]);
    let password = env_first(&["ATP_PASSWORD", "ATIRC_PASSWORD"]);
    let (Some(username), Some(password)) = (username, password) else {
        return Err("atirc requires --token, ATIRC_TOKEN, or ATP_USERNAME + ATP_PASSWORD".into());
    };
    let nick = cli
        .agent_name
        .clone()
        .ok_or("auto-mint requires --agent-name for the atirc nick")?;
    let endpoint_value = env_first(&["ATIRC_ENDPOINT"])
        .unwrap_or_else(|| atirc_client::DEFAULT_ENDPOINT.to_string());
    let endpoint =
        atirc_client::endpoint::Endpoint::parse(&endpoint_value).map_err(std::io::Error::other)?;
    let pds = env_first(&["ATIRC_PDS"]);
    atirc_client::transport::mint_long_token(&endpoint, &username, &password, &nick, pds.as_deref())
        .await
        .map_err(std::io::Error::other)
        .map_err(Into::into)
}

fn env_first(keys: &[&str]) -> Option<String> {
    keys.iter()
        .find_map(|key| std::env::var(key).ok().filter(|value| !value.is_empty()))
}

const DEFAULT_INTERACTIVE_PROMPT: &str = include_str!("../prompts/interactive.md");
const DEFAULT_ATIRC_PROMPT: &str = include_str!("../prompts/atirc.md");
/// Addendum to the atirc prompt for socialweb-computer nodes.
const SOCIALWEB_COMPUTER_ADDENDUM: &str = include_str!("../prompts/atirc-socialweb-computer.md");

/// Names accepted by `--prompt`.
pub const PROMPT_NAMES: &[&str] = &["atirc", "interactive", "atirc-socialweb-computer"];

/// A built-in prompt by name.
pub fn builtin_prompt(name: &str) -> Option<String> {
    match name {
        "atirc" => Some(DEFAULT_ATIRC_PROMPT.to_string()),
        "interactive" => Some(DEFAULT_INTERACTIVE_PROMPT.to_string()),
        "atirc-socialweb-computer" => Some(format!(
            "{DEFAULT_ATIRC_PROMPT}{SOCIALWEB_COMPUTER_ADDENDUM}"
        )),
        _ => None,
    }
}

/// Instruction appended to the `On Start` section when it is delivered.
pub const ON_START_INSTRUCTION: &str = "This is the operator's start-up instruction, not a user message. \
Do it now. Post only what the channel needs (what you set up, or a problem); otherwise reply exactly NO_REPLY as your message text (not inside run_code).";

/// Splits a prompt into the system prompt and its `On Start` section.
///
/// A markdown heading whose text is `On Start` (any level, any case) starts
/// the section; it runs until the next heading of the same or a higher
/// level. The section (heading included) is removed from the system prompt.
pub fn split_on_start(prompt: &str) -> (String, Option<String>) {
    let heading = |line: &str| -> Option<(usize, String)> {
        let level = line.chars().take_while(|c| *c == '#').count();
        let rest = &line[level..];
        (level > 0 && rest.starts_with(' ')).then(|| (level, rest.trim().to_string()))
    };
    let lines: Vec<&str> = prompt.lines().collect();
    let Some((start, level)) = lines.iter().enumerate().find_map(|(index, line)| {
        heading(line)
            .filter(|(_, text)| text.eq_ignore_ascii_case("on start"))
            .map(|(level, _)| (index, level))
    }) else {
        return (prompt.to_string(), None);
    };
    let end = lines[start + 1..]
        .iter()
        .position(|line| heading(line).is_some_and(|(other, _)| other <= level))
        .map(|offset| start + 1 + offset)
        .unwrap_or(lines.len());
    let section = lines[start + 1..end].join("\n").trim().to_string();
    let mut system: Vec<&str> = lines[..start].to_vec();
    while system.last().is_some_and(|line| line.trim().is_empty()) {
        system.pop();
    }
    if end < lines.len() {
        system.push("");
        system.extend(&lines[end..]);
    }
    let mut system = system.join("\n");
    system.push('\n');
    (system, (!section.is_empty()).then_some(section))
}

/// The system prompt plus the `On Start` text to deliver once at startup.
fn load_prompt(
    cli: &Cli,
    default_prompt: &str,
    agent_name: &str,
) -> Result<(String, Option<String>), Box<dyn std::error::Error>> {
    let template = match (&cli.prompt_file, &cli.prompt) {
        (Some(path), _) => std::fs::read_to_string(path).map_err(|error| {
            std::io::Error::other(format!("failed to read prompt file {path}: {error}"))
        })?,
        (None, Some(name)) => builtin_prompt(name).ok_or_else(|| {
            std::io::Error::other(format!(
                "unknown --prompt {name:?}; use one of {}",
                PROMPT_NAMES.join(", ")
            ))
        })?,
        (None, None) => default_prompt.to_string(),
    };
    Ok(split_on_start(
        &template.replace("{agent_name}", agent_name),
    ))
}

fn base_url(cli: &Cli) -> Option<String> {
    cli.base_url
        .clone()
        .or_else(|| env_first(&["OPENAI_BASE_URL", "ASYNC_LLM_BASE_URL"]))
}

fn model_name(cli: &Cli) -> String {
    cli.model
        .clone()
        .or_else(|| env_first(&["OPENAI_MODEL", "ASYNC_LLM_MODEL"]))
        .unwrap_or_else(|| "local".to_string())
}

fn parse_session_overrides(
    values: &[String],
) -> Result<std::collections::HashMap<String, PathBuf>, Box<dyn std::error::Error>> {
    let mut overrides = std::collections::HashMap::new();
    for value in values {
        let Some((format, path)) = value.split_once('=') else {
            return Err("--session-dir expects FORMAT=PATH".into());
        };
        if format.is_empty() || path.is_empty() {
            return Err("--session-dir expects FORMAT=PATH".into());
        }
        overrides.insert(format.to_string(), PathBuf::from(path));
    }
    Ok(overrides)
}

fn create_session(
    cli: &Cli,
    agent_name: &str,
) -> Result<Option<SessionLogger>, Box<dyn std::error::Error>> {
    if cli.no_session {
        return Ok(None);
    }
    let cwd = std::env::current_dir()?;
    let model = model_name(cli);
    let overrides = parse_session_overrides(&cli.session_dir)?;
    let formats = if cli.session_format.is_empty() {
        vec!["claude-code-cli".to_string()]
    } else {
        cli.session_format.clone()
    };

    let mut harnesses: Vec<Box<dyn SessionHarness>> = Vec::new();
    for format in formats {
        match format.as_str() {
            "claude-code-cli" => harnesses.push(Box::new(ClaudeCodeCliHarness::create(
                overrides.get("claude-code-cli").map(PathBuf::as_path),
                &cwd,
                &model,
            )?)),
            other => {
                return Err(format!(
                    "unknown session format {other:?}; available: claude-code-cli"
                )
                .into());
            }
        }
    }

    let session = SessionLogger::start(harnesses, agent_name)?;
    println!("{}", session.first_line()?);
    std::io::stdout().flush()?;
    Ok(Some(session))
}

fn start_monitors(
    cli: &Cli,
    session: Option<&SessionLogger>,
) -> Result<Vec<MonitorRunner>, Box<dyn std::error::Error>> {
    if cli.monitor.is_empty() {
        return Ok(Vec::new());
    }
    let session =
        session.ok_or_else(|| std::io::Error::other("--monitor requires session logging"))?;
    let tx = session.transcript_sender();
    let mut runners = Vec::new();
    for command in &cli.monitor {
        runners.push(MonitorRunner::spawn(
            command.clone(),
            format!("asyncai monitor: {command}"),
            tx.clone(),
        )?);
    }
    Ok(runners)
}

fn log_atirc_event(
    session: &SessionLogger,
    task_id: &str,
    description: &str,
    message: &ChannelMessage,
) {
    let event = MonitorEvent {
        task_id: task_id.to_string(),
        description: description.to_string(),
        event: message.text.clone(),
        source_uuid: Uuid::new_v4().to_string(),
        delivery_id: Uuid::new_v4().to_string(),
        human_turn: false,
        note: None,
    };
    session.monitor_event(event.clone());
    session.monitor_consumed(event, "absorbed_mid_turn");
}

/// Tools for the agent plus the watch notices they produce (ADR 0005).
#[derive(Default)]
pub struct ToolSetup {
    pub executor: Option<Arc<dyn ToolExecutor>>,
    pub notices: Option<tokio::sync::mpsc::UnboundedReceiver<WatchNotice>>,
}

/// Next watch notice, or never when watches are off.
async fn next_notice(
    notices: &mut Option<tokio::sync::mpsc::UnboundedReceiver<WatchNotice>>,
) -> Option<WatchNotice> {
    match notices {
        Some(receiver) => receiver.recv().await,
        None => std::future::pending().await,
    }
}

/// Tracks inputs (user lines, channel messages, watch notices) against runs.
///
/// The runtime coalesces inputs that arrive during an active run into one
/// queued follow-up run, so inputs and runs are not 1:1. Each
/// `new_thread_run_created` claims every input not yet claimed; replies of
/// that run go to the route of the most recent claimed input.
#[derive(Debug, Default)]
struct TurnTracker {
    unclaimed: Vec<Option<String>>,
    run_route: Option<String>,
    active_runs: usize,
}

impl TurnTracker {
    fn input(&mut self, route: Option<String>) {
        self.unclaimed.push(route);
    }

    fn observe(&mut self, message: &OutputMessage) {
        match event_type_of(message) {
            Some("new_thread_run_created") => {
                self.active_runs += 1;
                if let Some(route) = self.unclaimed.drain(..).rev().flatten().next() {
                    self.run_route = Some(route);
                }
            }
            Some("thread_run_complete" | "thread_run_failed") => {
                self.active_runs = self.active_runs.saturating_sub(1);
            }
            _ => {}
        }
    }

    fn reply_route(&self) -> Option<&str> {
        self.run_route.as_deref()
    }

    /// No input waits for a run and no run is active.
    fn idle(&self) -> bool {
        self.unclaimed.is_empty() && self.active_runs == 0
    }
}

/// Queues the prompt's `On Start` section as the first input: a session
/// monitor event (never a user record) and an orchestrator notice.
///
/// Watch notices already queued at startup (the `--watch-state` restore
/// notice) are folded into the same message, so startup is one turn instead
/// of two near-identical ones.
fn deliver_on_start(
    text: &str,
    notices: &mut Option<tokio::sync::mpsc::UnboundedReceiver<WatchNotice>>,
    session: Option<&SessionLogger>,
    handle: &OrchestratorHandle,
) -> bool {
    let mut queued = Vec::new();
    if let Some(receiver) = notices.as_mut() {
        while let Ok(notice) = receiver.try_recv() {
            queued.push(notice);
        }
    }
    crate::log::emit(
        "info",
        "prompt.on_start",
        json!({"chars": text.chars().count(), "folded_notices": queued.len()}),
    );
    let mut message = format!("<on-start>\n{text}\n</on-start>\n{ON_START_INSTRUCTION}");
    for notice in &queued {
        record_notice(notice, session);
        message.push_str("\n\n");
        message.push_str(&notice.render());
    }
    if let Some(session) = session {
        let event = MonitorEvent {
            task_id: "on-start".to_string(),
            description: "prompt On Start".to_string(),
            event: text.to_string(),
            source_uuid: Uuid::new_v4().to_string(),
            delivery_id: Uuid::new_v4().to_string(),
            human_turn: false,
            note: Some(ON_START_INSTRUCTION.to_string()),
        };
        session.monitor_event(event.clone());
        session.monitor_consumed(event, "delivered_as_turn");
    }
    handle
        .send_notice(Notice {
            source: "prompt:on-start".to_string(),
            text: message,
        })
        .is_ok()
}

/// Records a watch notice in the session as a monitor event (never a user
/// record) and hands it to the orchestrator.
fn deliver_notice(
    notice: &WatchNotice,
    session: Option<&SessionLogger>,
    handle: &OrchestratorHandle,
) -> bool {
    record_notice(notice, session);
    handle
        .send_notice(Notice {
            source: notice.source(),
            text: notice.render(),
        })
        .is_ok()
}

/// A watch notice as a session monitor event.
fn record_notice(notice: &WatchNotice, session: Option<&SessionLogger>) {
    if let Some(session) = session {
        let event = MonitorEvent {
            task_id: format!("watch-{}", notice.name),
            description: format!("watch {} {}", notice.name, notice.uri),
            event: notice.lines.join("\n"),
            source_uuid: Uuid::new_v4().to_string(),
            delivery_id: Uuid::new_v4().to_string(),
            human_turn: false,
            note: Some(crate::code_mode::watch::WATCH_INSTRUCTION.to_string()),
        };
        session.monitor_event(event.clone());
        session.monitor_consumed(event, "delivered_as_turn");
    }
}

/// Connects the `--mcp-stdio` servers and puts them behind code mode: the
/// model gets `run_code`, and every MCP call is made from Monty.
async fn connect_tool_executor(
    cli: &Cli,
    verbose: bool,
) -> Result<ToolSetup, Box<dyn std::error::Error>> {
    if cli.mcp_stdio.is_empty() {
        return Ok(ToolSetup::default());
    }

    let configs = cli
        .mcp_stdio
        .iter()
        .map(|spec| StdioServerConfig::parse(spec))
        .collect::<Result<Vec<_>, _>>()?;
    let mcp = Arc::new(McpToolExecutor::connect_stdio(configs).await?);
    for duplicate in mcp.duplicates() {
        crate::log::emit(
            "warn",
            "mcp.duplicate_tool",
            json!({"tool": duplicate.name, "kept": duplicate.kept, "ignored": duplicate.ignored}),
        );
    }
    let preamble = match &cli.code_preamble {
        Some(path) => Some(
            std::fs::read_to_string(path)
                .map_err(|error| format!("--code-preamble {}: {error}", path.display()))?,
        ),
        None => None,
    };
    let options = CodeModeOptions {
        limits: CodeModeLimits {
            max_memory: cli.code_max_memory,
            ..CodeModeLimits::default()
        },
        verbose,
        sandbox: cli.code_sandbox,
        monty_worker: cli.monty_worker.clone(),
        preamble,
        resources: Some(mcp.clone()),
        watch_limits: WatchLimits::default(),
        watch_state: cli.watch_state.clone(),
    };
    let code_mode = CodeModeToolExecutor::with_options(mcp, options).await?;
    if verbose {
        crate::log::emit(
            "info",
            "code_mode.sandbox",
            json!({
                "sandbox": code_mode.sandbox_name(),
                "max_memory": (code_mode.sandbox_name() == "pool").then_some(cli.code_max_memory),
                "tools": code_mode.catalog().entries().len(),
            }),
        );
    }
    let notices = code_mode.take_watch_notices();
    // Notices queue on an unbounded channel until the agent loop reads them.
    code_mode.restore_watches().await;
    Ok(ToolSetup {
        executor: Some(Arc::new(code_mode)),
        notices,
    })
}

fn make_backend(
    cli: &Cli,
    tool_executor: Option<Arc<dyn ToolExecutor>>,
    tool_observer: Option<Arc<dyn ToolObserver>>,
    verbose_logging: bool,
) -> Result<Arc<dyn AgentBackend>, Box<dyn std::error::Error>> {
    let base_url = base_url(cli);

    match base_url {
        Some(base_url) => {
            let model = model_name(cli);
            let mut backend = OpenAiCompatibleBackend::new(base_url, model);
            if let Some(api_key) = cli
                .api_key
                .clone()
                .or_else(|| env_first(&["OPENAI_API_KEY", "ASYNC_LLM_API_KEY"]))
            {
                backend = backend.with_api_key(api_key);
            }
            if let Some(executor) = tool_executor {
                backend = backend.with_tool_executor(executor);
            }
            if let Some(observer) = tool_observer {
                backend = backend.with_tool_observer(observer);
            }
            backend = backend
                .with_verbose_logging(verbose_logging)
                .with_max_output_tokens(cli.max_output_tokens)
                .with_request_timeout(Duration::from_secs(cli.model_timeout));
            Ok(Arc::new(backend))
        }
        None if tool_executor.is_some() => {
            Err("--mcp-stdio requires --base-url or OPENAI_BASE_URL for tool calling".into())
        }
        None => Ok(Arc::new(MockBackend::echo())),
    }
}

fn orchestrator_handle(
    backend: Arc<dyn AgentBackend>,
    agent_name: String,
    instructions: String,
) -> (OrchestratorHandle, JoinHandle<Result<(), AgiError>>) {
    let config = OrchestratorConfig {
        agi_name: agent_name,
        agent_instructions: instructions,
        ..Default::default()
    };
    let (orchestrator, handle) =
        Orchestrator::new(backend, Arc::new(MemoryKvStore::default()), config);
    (handle, tokio::spawn(orchestrator.run()))
}

async fn run_interactive(cli: Cli, tools: ToolSetup) -> Result<(), Box<dyn std::error::Error>> {
    let ToolSetup {
        executor: tool_executor,
        mut notices,
    } = tools;
    let agent_name = cli
        .agent_name
        .clone()
        .unwrap_or_else(|| "alice".to_string());
    let (instructions, on_start) = load_prompt(&cli, DEFAULT_INTERACTIVE_PROMPT, &agent_name)?;
    let session = create_session(&cli, &agent_name)?;
    if cli.no_stdin {
        if cli.monitor.is_empty() {
            return Err("--no-stdin requires --monitor".into());
        }
        let monitors = start_monitors(&cli, session.as_ref())?;
        for monitor in monitors {
            monitor.wait().await;
        }
        if let Some(session) = session {
            session.close().await;
        }
        return Ok(());
    }
    let backend = make_backend(
        &cli,
        tool_executor,
        session.as_ref().map(|session| session.tool_observer()),
        false,
    )?;
    let monitors = start_monitors(&cli, session.as_ref())?;
    let (mut handle, mut run) = orchestrator_handle(backend, agent_name, instructions);

    let mut lines = BufReader::new(tokio::io::stdin()).lines();
    let mut input_open = true;
    let mut turns = TurnTracker::default();
    if let Some(text) = &on_start {
        turns.input(None);
        deliver_on_start(text, &mut notices, session.as_ref(), &handle);
    }

    loop {
        tokio::select! {
            line = lines.next_line(), if input_open => {
                match line {
                    Ok(Some(line)) => {
                        turns.input(None);
                        if handle.input_tx.send(line.clone()).is_err() {
                            break;
                        }
                        if let Some(session) = &session {
                            session.user_message(line);
                        }
                    }
                    Ok(None) => {
                        input_open = false;
                        if turns.idle() {
                            break;
                        }
                    }
                    Err(error) => {
                        crate::log::emit(
                            "error",
                            "stdin.read_failed",
                            json!({"error": error.to_string()}),
                        );
                        input_open = false;
                        if turns.idle() {
                            break;
                        }
                    }
                }
            }
            notice = next_notice(&mut notices) => {
                match notice {
                    Some(notice) => {
                        turns.input(None);
                        if !deliver_notice(&notice, session.as_ref(), &handle) {
                            break;
                        }
                    }
                    None => notices = None,
                }
            }
            output = handle.output_rx.recv() => {
                match output {
                    Some(message) => {
                        if event_type_of(&message) == Some("new_thread_message")
                            && event_data(&message)["message_role"] == "assistant"
                        {
                            if let (Some(session), Some(text)) =
                                (&session, assistant_text(&message))
                            {
                                session.assistant_message(text);
                            }
                        }
                        turns.observe(&message);
                        match message.to_ndjson() {
                            Ok(line) => println!("{line}"),
                            Err(error) => crate::log::emit(
                                "error",
                                "output.serialize_failed",
                                json!({"error": error.to_string()}),
                            ),
                        }
                        if !input_open && turns.idle() {
                            break;
                        }
                    }
                    None => break,
                }
            }
            result = &mut run => {
                report_run_result(result);
                break;
            }
        }
    }

    run.abort();
    let _ = run.await;
    for monitor in monitors {
        monitor.finish(Duration::from_millis(250)).await;
    }
    if let Some(session) = session {
        session.close().await;
    }
    Ok(())
}

async fn run_atirc(
    cli: Cli,
    args: AtircArgs,
    tools: ToolSetup,
    client: Arc<dyn AtircClient>,
) -> Result<(), Box<dyn std::error::Error>> {
    let ToolSetup {
        executor: tool_executor,
        mut notices,
    } = tools;
    let client_identity = client.identity();
    let (event_tx, mut event_rx) = tokio::sync::mpsc::unbounded_channel();
    let subscription_client = Arc::clone(&client);
    let channels = args.channels.clone();
    let cursor = args.cursor;
    let subscription = tokio::spawn(async move {
        if let Err(error) = subscription_client
            .subscribe(channels, cursor, event_tx)
            .await
        {
            crate::log::emit(
                "error",
                "atirc.subscription_ended",
                json!({"error": error.to_string()}),
            );
        }
    });

    let connection = wait_for_connection(&mut event_rx).await?;
    if args.verbose {
        eprintln!(
            "{}",
            serde_json::json!({
                "level": "info",
                "event": "atirc.connected",
                "agent_name": connection.subscribed_as.clone(),
                "subhandle": client_identity.subhandle.clone(),
                "channels": connection.channels.clone(),
                "transport": connection.transport.clone(),
            })
        );
    }
    let agent_name = cli
        .agent_name
        .clone()
        .unwrap_or_else(|| connection.subscribed_as.clone());
    let (instructions, on_start) = load_prompt(&cli, DEFAULT_ATIRC_PROMPT, &agent_name)?;
    let session = create_session(&cli, &agent_name)?;
    let backend = make_backend(
        &cli,
        tool_executor,
        session.as_ref().map(|session| session.tool_observer()),
        args.verbose,
    )?;
    let monitors = start_monitors(&cli, session.as_ref())?;
    let (mut handle, mut run) = orchestrator_handle(backend, agent_name.clone(), instructions);
    let atirc_task_id = format!("atirc-{}", &Uuid::new_v4().simple().to_string()[..8]);
    let sub_command = synthetic_sub_command(&args.channels, args.cursor);

    let mut subscription_open = true;
    let mut published_turns = 0usize;
    let mut turns = TurnTracker::default();
    // Watch notices reply where the agent was last addressed.
    let mut last_channel: Option<String> = connection.channels.first().cloned();
    if let Some(text) = &on_start {
        turns.input(last_channel.clone());
        deliver_on_start(text, &mut notices, session.as_ref(), &handle);
    }

    loop {
        let at_limit = args.max_turns > 0 && published_turns >= args.max_turns;
        if turns.idle() && (!subscription_open || at_limit) {
            break;
        }

        tokio::select! {
            event = event_rx.recv() => {
                match event {
                    Some(AtircEvent::Message(message)) => {
                        if args.verbose {
                            eprintln!(
                                "{}",
                                serde_json::json!({
                                    "level": "info",
                                    "event": "atirc.message",
                                    "channel": message.channel.clone(),
                                    "identity": message.identity.clone(),
                                    "text": message.text.clone(),
                                })
                            );
                        }
                        let addressed =
                            !args.mention_only || mentions(&message.text, &agent_name);
                        let under_limit = args.max_turns == 0
                            || published_turns < args.max_turns;

                        if !client_identity.is_self(message.identity.as_deref())
                            && addressed
                            && under_limit
                        {
                            if let Some(session) = &session {
                                log_atirc_event(
                                    session,
                                    &atirc_task_id,
                                    &sub_command,
                                    &message,
                                );
                            }
                            last_channel = Some(message.channel.clone());
                            turns.input(Some(message.channel));
                            if handle.input_tx.send(message.text).is_err() {
                                break;
                            }
                        }
                    }
                    Some(AtircEvent::Connected(_)) => {}
                    None => subscription_open = false,
                }
            }
            notice = next_notice(&mut notices) => {
                match notice {
                    Some(notice) => {
                        turns.input(last_channel.clone());
                        if !deliver_notice(&notice, session.as_ref(), &handle) {
                            break;
                        }
                    }
                    None => notices = None,
                }
            }
            output = handle.output_rx.recv() => {
                match output {
                    Some(message) => {
                        if args.verbose {
                            match message.to_ndjson() {
                                Ok(line) => eprintln!("{line}"),
                                Err(error) => crate::log::emit(
                                "error",
                                "output.serialize_failed",
                                json!({"error": error.to_string()}),
                            ),
                            }
                        }

                        turns.observe(&message);
                        match event_type_of(&message) {
                            Some("new_thread_message")
                                if event_data(&message)["message_role"] == "assistant" =>
                            {
                                if let Some(content) = assistant_text(&message) {
                                    if let Some(session) = &session {
                                        session.assistant_message(content.as_str());
                                    }
                                    if let Some(channel) = turns.reply_route().map(str::to_string) {
                                        let under_limit = args.max_turns == 0
                                            || published_turns < args.max_turns;
                                        if under_limit && is_no_reply(&content) {
                                            crate::log::emit(
                                                "info",
                                                "reply.suppressed",
                                                json!({"channel": channel, "chars": content.chars().count()}),
                                            );
                                        } else if under_limit {
                                            let replies = format_reply(&content);
                                            let parts = replies.len();
                                            let mut all_published = true;
                                            for (index, reply) in replies.iter().enumerate() {
                                                match client.publish(&channel, reply).await {
                                                    Ok(()) => {
                                                        if args.verbose {
                                                            eprintln!(
                                                                "{}",
                                                                serde_json::json!({
                                                                    "level": "info",
                                                                    "event": "atirc.published",
                                                                    "channel": channel,
                                                                    "part": index + 1,
                                                                    "parts": parts,
                                                                    "bytes": reply.len(),
                                                                })
                                                            );
                                                        }
                                                    }
                                                    Err(error) => {
                                                        all_published = false;
                                                        eprintln!(
                                                            "{}",
                                                            serde_json::json!({
                                                                "level": "error",
                                                                "event": "atirc.publish_failed",
                                                                "channel": channel,
                                                                "part": index + 1,
                                                                "parts": parts,
                                                                "error": error,
                                                            })
                                                        );
                                                    }
                                                }
                                            }
                                            if all_published && parts > 0 {
                                                published_turns += 1;
                                            }
                                        }
                                    }
                                }
                            }
                            _ => {}
                        }
                    }
                    None => break,
                }
            }
            result = &mut run => {
                report_run_result(result);
                break;
            }
        }
    }

    run.abort();
    let _ = run.await;
    subscription.abort();
    for monitor in monitors {
        monitor.finish(Duration::from_millis(250)).await;
    }
    if let Some(session) = session {
        session.close().await;
    }
    Ok(())
}

async fn wait_for_connection(
    rx: &mut tokio::sync::mpsc::UnboundedReceiver<AtircEvent>,
) -> Result<ConnectionInfo, Box<dyn std::error::Error>> {
    loop {
        match tokio::time::timeout(Duration::from_secs(30), rx.recv()).await {
            Ok(Some(AtircEvent::Connected(connection))) => return Ok(connection),
            Ok(Some(AtircEvent::Message(_))) => continue,
            Ok(None) => return Err("atirc subscription ended before connecting".into()),
            Err(_) => return Err("timed out waiting for atirc connection".into()),
        }
    }
}

fn event_type_of(message: &OutputMessage) -> Option<&str> {
    message.result.get("event_type").and_then(Value::as_str)
}

fn event_data(message: &OutputMessage) -> &Value {
    &message.result["event_data"]
}

fn assistant_text(message: &OutputMessage) -> Option<String> {
    let content = event_data(message).get("message_content")?;
    if let Some(text) = content.get("text").and_then(Value::as_str) {
        return Some(text.to_string());
    }
    content.get("json").map(Value::to_string)
}

fn report_run_result(result: Result<Result<(), AgiError>, tokio::task::JoinError>) {
    match result {
        Ok(Ok(())) => {}
        Ok(Err(error)) => crate::log::emit(
            "error",
            "orchestrator.failed",
            json!({"error": error.to_string()}),
        ),
        Err(error) => crate::log::emit(
            "error",
            "orchestrator.task_failed",
            json!({"error": error.to_string()}),
        ),
    }
}

#[cfg(test)]
mod tests {
    use std::path::{Path, PathBuf};
    use std::sync::Mutex;

    fn agent_event(event_type: &str) -> OutputMessage {
        OutputMessage::new(
            "main.agent.events",
            serde_json::json!({"event_type": event_type, "event_data": {}}),
        )
    }

    #[test]
    fn turn_tracker_routes_coalesced_inputs_to_the_latest_channel() {
        let mut turns = TurnTracker::default();
        assert!(turns.idle());

        turns.input(Some("#a".to_string()));
        assert!(!turns.idle());
        turns.observe(&agent_event("new_thread_run_created"));
        assert_eq!(turns.reply_route(), Some("#a"));

        // Two inputs arrive during the run: the runtime coalesces them into
        // one follow-up run, which claims both.
        turns.input(Some("#b".to_string()));
        turns.input(None);
        turns.observe(&agent_event("thread_run_in_progress"));
        turns.observe(&agent_event("thread_run_complete"));
        assert!(!turns.idle(), "unclaimed inputs keep the loop alive");
        turns.observe(&agent_event("new_thread_run_created"));
        assert_eq!(
            turns.reply_route(),
            Some("#b"),
            "a notice without a route keeps the last channel"
        );
        turns.observe(&agent_event("thread_run_failed"));
        assert!(turns.idle());
    }

    use async_trait::async_trait;
    use tokio::sync::mpsc::UnboundedSender;

    use super::*;
    use crate::atirc::{AtircEvent, AtircIdentity, ChannelMessage, ConnectionInfo};

    #[test]
    fn parses_model_and_agent_name_flags() {
        let cli = Cli::try_parse_from([
            "asyncai",
            "--model",
            "some-model",
            "--agent-name",
            "alice",
            "atirc",
            "--channel",
            "#test_asyncai",
            "--cursor",
            "-1",
            "--max-turns",
            "2",
        ])
        .unwrap();

        assert_eq!(cli.model.as_deref(), Some("some-model"));
        assert_eq!(cli.agent_name.as_deref(), Some("alice"));
        let Some(Command::Atirc(args)) = cli.command else {
            panic!("expected atirc command");
        };
        assert_eq!(args.channels, vec!["#test_asyncai"]);
        assert_eq!(args.cursor, -1);
        assert_eq!(args.max_turns, 2);
    }

    #[test]
    fn parses_plain_interactive_mode() {
        let cli = Cli::try_parse_from(["asyncai", "--agent-name", "bob"]).unwrap();
        assert_eq!(cli.agent_name.as_deref(), Some("bob"));
        assert!(cli.command.is_none());
    }

    fn test_cli(session_root: &Path) -> Cli {
        Cli {
            model: None,
            agent_name: Some("sess-agent".to_string()),
            base_url: None,
            api_key: None,
            mcp_stdio: Vec::new(),
            code_sandbox: SandboxKind::InProcess,
            monty_worker: None,
            code_max_memory: CodeModeLimits::default().max_memory,
            code_preamble: None,
            watch_state: None,
            max_output_tokens: crate::openai_backend::DEFAULT_MAX_OUTPUT_TOKENS,
            model_timeout: 600,
            prompt_file: None,
            prompt: None,
            session_dir: vec![format!("claude-code-cli={}", session_root.display())],
            session_format: Vec::new(),
            no_session: false,
            no_stdin: false,
            monitor: Vec::new(),
            command: None,
        }
    }

    fn transcript_files(root: &Path) -> Vec<PathBuf> {
        let mut files = Vec::new();
        let Ok(projects) = std::fs::read_dir(root.join("projects")) else {
            return files;
        };
        for project in projects.flatten() {
            let Ok(entries) = std::fs::read_dir(project.path()) else {
                continue;
            };
            for entry in entries.flatten() {
                let path = entry.path();
                if path.extension().and_then(|value| value.to_str()) == Some("jsonl") {
                    files.push(path);
                }
            }
        }
        files
    }

    fn transcript_records(root: &Path) -> Vec<serde_json::Value> {
        let files = transcript_files(root);
        assert_eq!(files.len(), 1, "expected one transcript: {files:?}");
        std::fs::read_to_string(&files[0])
            .unwrap()
            .lines()
            .map(|line| serde_json::from_str(line).unwrap())
            .collect()
    }

    struct MockAtircClient {
        identity: AtircIdentity,
        events: Mutex<Option<Vec<AtircEvent>>>,
        published: Mutex<Vec<(String, String)>>,
    }

    #[async_trait]
    impl AtircClient for MockAtircClient {
        fn identity(&self) -> AtircIdentity {
            self.identity.clone()
        }

        async fn subscribe(
            &self,
            _channels: Vec<String>,
            _cursor: i64,
            tx: UnboundedSender<AtircEvent>,
        ) -> Result<(), String> {
            let events = self.events.lock().unwrap().take().unwrap_or_default();
            for event in events {
                let _ = tx.send(event);
            }
            Ok(())
        }

        async fn publish(&self, channel: &str, text: &str) -> Result<(), String> {
            self.published
                .lock()
                .unwrap()
                .push((channel.to_string(), text.to_string()));
            Ok(())
        }
    }

    #[tokio::test]
    async fn atirc_mode_subscribes_and_publishes_via_client() {
        let temp = tempfile::tempdir().unwrap();
        let identity = AtircIdentity {
            nick: "sess-agent".to_string(),
            handle: "example.test".to_string(),
            subhandle: "sess-agent.example.test".to_string(),
        };
        let connection = ConnectionInfo {
            subscribed_as: "sess-agent".to_string(),
            owner_nick: Some("owner".to_string()),
            your_user_is: Some("example.test".to_string()),
            transport: Some("http".to_string()),
            identities: vec!["sess-agent@sess-agent.example.test".to_string()],
            channels: vec!["#test-asyncai".to_string()],
        };
        let message = ChannelMessage {
            channel: "#test-asyncai".to_string(),
            identity: Some("tester.example.test".to_string()),
            text: "hello".to_string(),
        };
        let self_message = ChannelMessage {
            channel: "#test-asyncai".to_string(),
            identity: Some("sess-agent.example.test".to_string()),
            text: "self message".to_string(),
        };
        let client = Arc::new(MockAtircClient {
            identity,
            events: Mutex::new(Some(vec![
                AtircEvent::Connected(connection),
                AtircEvent::Message(message),
                AtircEvent::Message(self_message),
            ])),
            published: Mutex::new(Vec::new()),
        });

        run_atirc(
            test_cli(temp.path()),
            AtircArgs {
                token: None,
                channels: vec!["#test-asyncai".to_string()],
                cursor: -1,
                max_turns: 1,
                mention_only: false,
                verbose: false,
            },
            ToolSetup::default(),
            client.clone(),
        )
        .await
        .unwrap();

        assert_eq!(
            *client.published.lock().unwrap(),
            vec![("#test-asyncai".to_string(), "hello".to_string())]
        );

        let records = transcript_records(temp.path());
        assert!(records.iter().any(|record| {
            record["type"] == "attachment" && record["attachment"]["type"] == "queued_command"
        }));
        assert!(records.iter().any(|record| {
            record["type"] == "assistant"
                && record["message"]["content"]
                    .as_array()
                    .is_some_and(|content| content.iter().any(|block| block["text"] == "hello"))
        }));
        assert!(!records.iter().any(|record| {
            record["type"] == "user"
                && record["message"]["content"]
                    .as_str()
                    .is_some_and(|content| content == "hello")
        }));
    }

    #[tokio::test]
    async fn atirc_mode_publishes_multiline_reply_as_separate_messages() {
        let temp = tempfile::tempdir().unwrap();
        let identity = AtircIdentity {
            nick: "sess-agent".to_string(),
            handle: "example.test".to_string(),
            subhandle: "sess-agent.example.test".to_string(),
        };
        let connection = ConnectionInfo {
            subscribed_as: "sess-agent".to_string(),
            owner_nick: Some("owner".to_string()),
            your_user_is: Some("example.test".to_string()),
            transport: Some("http".to_string()),
            identities: vec!["sess-agent@sess-agent.example.test".to_string()],
            channels: vec!["#test-asyncai".to_string()],
        };
        let message = ChannelMessage {
            channel: "#test-asyncai".to_string(),
            identity: Some("tester.example.test".to_string()),
            text: "line one\nline two\n\n  line three  ".to_string(),
        };
        let client = Arc::new(MockAtircClient {
            identity,
            events: Mutex::new(Some(vec![
                AtircEvent::Connected(connection),
                AtircEvent::Message(message),
            ])),
            published: Mutex::new(Vec::new()),
        });

        run_atirc(
            test_cli(temp.path()),
            AtircArgs {
                token: None,
                channels: vec!["#test-asyncai".to_string()],
                cursor: -1,
                max_turns: 1,
                mention_only: false,
                verbose: false,
            },
            ToolSetup::default(),
            client.clone(),
        )
        .await
        .unwrap();

        assert_eq!(
            *client.published.lock().unwrap(),
            vec![
                ("#test-asyncai".to_string(), "line one".to_string()),
                ("#test-asyncai".to_string(), "line two".to_string()),
                ("#test-asyncai".to_string(), "line three".to_string()),
            ]
        );
    }

    #[test]
    fn on_start_sections_are_split_out() {
        let prompt = "You are x.\n\n## Rules\n- a\n\n## On start\n\nWatch bids.\n### detail\nkeep this in on-start\n\n## Later\n- b\n";
        let (system, on_start) = split_on_start(prompt);
        assert_eq!(system, "You are x.\n\n## Rules\n- a\n\n## Later\n- b\n");
        assert_eq!(
            on_start.as_deref(),
            Some("Watch bids.\n### detail\nkeep this in on-start")
        );

        let (system, on_start) = split_on_start("# P\n\n# ON START\n\ndo it\n");
        assert_eq!(
            (system.as_str(), on_start.as_deref()),
            ("# P\n", Some("do it"))
        );

        let (system, on_start) = split_on_start("no section\n#On Start is not a heading\n");
        assert_eq!(system, "no section\n#On Start is not a heading\n");
        assert_eq!(on_start, None);

        let (_, empty) = split_on_start("x\n## On Start\n\n## Next\n");
        assert_eq!(empty, None, "empty section delivers nothing");

        let (system, on_start) =
            split_on_start(&builtin_prompt("atirc-socialweb-computer").unwrap());
        assert!(!system.to_lowercase().contains("## on start"), "{system}");
        assert!(on_start.unwrap().contains("monitors"));
    }

    #[tokio::test]
    async fn atirc_mode_delivers_on_start_first() {
        let temp = tempfile::tempdir().unwrap();
        let prompt = temp.path().join("prompt.md");
        std::fs::write(
            &prompt,
            "You are {agent_name}.\n\n## On Start\n\nset up monitors\n",
        )
        .unwrap();
        let mut cli = test_cli(temp.path());
        cli.prompt_file = Some(prompt.display().to_string());
        let identity = AtircIdentity {
            nick: "sess-agent".to_string(),
            handle: "example.test".to_string(),
            subhandle: "sess-agent.example.test".to_string(),
        };
        let connection = ConnectionInfo {
            subscribed_as: "sess-agent".to_string(),
            owner_nick: Some("owner".to_string()),
            your_user_is: Some("example.test".to_string()),
            transport: Some("http".to_string()),
            identities: vec!["sess-agent@sess-agent.example.test".to_string()],
            channels: vec!["#test-asyncai".to_string()],
        };
        let client = Arc::new(MockAtircClient {
            identity,
            events: Mutex::new(Some(vec![AtircEvent::Connected(connection)])),
            published: Mutex::new(Vec::new()),
        });

        run_atirc(
            cli,
            AtircArgs {
                token: None,
                channels: vec!["#test-asyncai".to_string()],
                cursor: -1,
                max_turns: 1,
                mention_only: false,
                verbose: false,
            },
            {
                // A restore notice already queued at startup.
                let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
                sender
                    .send(WatchNotice {
                        name: "restore".to_string(),
                        uri: "watches.json".to_string(),
                        lines: vec!["restored bids kcp-events://watch/root/bids".to_string()],
                        dropped: 0,
                    })
                    .unwrap();
                ToolSetup {
                    executor: None,
                    notices: Some(receiver),
                }
            },
            client.clone(),
        )
        .await
        .unwrap();

        // The mock backend echoes its input: On Start and the queued restore
        // notice ran as one turn (max_turns 1) and replied to the first channel.
        let published = client.published.lock().unwrap().clone();
        assert!(
            published
                .iter()
                .any(|(channel, text)| channel == "#test-asyncai" && text == "set up monitors"),
            "{published:?}"
        );
        assert!(
            published
                .iter()
                .any(|(_, text)| text == "restored bids kcp-events://watch/root/bids"),
            "restore notice folded into the On Start turn: {published:?}"
        );

        let transcript = std::fs::read_dir(temp.path())
            .unwrap()
            .flatten()
            .flat_map(|entry| walk(&entry.path()))
            .find(|path| path.extension().is_some_and(|ext| ext == "jsonl"))
            .map(|path| std::fs::read_to_string(path).unwrap())
            .expect("session transcript");
        assert!(transcript.contains("prompt On Start"), "{transcript}");
        assert!(
            !transcript.contains("\"type\":\"user\""),
            "On Start is a monitor event, not a user record"
        );
    }

    fn walk(path: &std::path::Path) -> Vec<std::path::PathBuf> {
        if path.is_dir() {
            std::fs::read_dir(path)
                .unwrap()
                .flatten()
                .flat_map(|entry| walk(&entry.path()))
                .collect()
        } else {
            vec![path.to_path_buf()]
        }
    }

    #[test]
    fn builtin_prompts_compose() {
        let socialweb = builtin_prompt("atirc-socialweb-computer").unwrap();
        assert!(socialweb.starts_with(DEFAULT_ATIRC_PROMPT));
        assert!(socialweb.contains("kcp-events://watch/root/"));
        assert_eq!(builtin_prompt("atirc").unwrap(), DEFAULT_ATIRC_PROMPT);
        assert!(builtin_prompt("nope").is_none());
        for name in PROMPT_NAMES {
            assert!(builtin_prompt(name).is_some(), "{name}");
        }
    }

    #[tokio::test]
    async fn atirc_mode_no_reply_publishes_nothing() {
        let temp = tempfile::tempdir().unwrap();
        let identity = AtircIdentity {
            nick: "sess-agent".to_string(),
            handle: "example.test".to_string(),
            subhandle: "sess-agent.example.test".to_string(),
        };
        let connection = ConnectionInfo {
            subscribed_as: "sess-agent".to_string(),
            owner_nick: Some("owner".to_string()),
            your_user_is: Some("example.test".to_string()),
            transport: Some("http".to_string()),
            identities: vec!["sess-agent@sess-agent.example.test".to_string()],
            channels: vec!["#test-asyncai".to_string()],
        };
        let message = |text: &str| {
            AtircEvent::Message(ChannelMessage {
                channel: "#test-asyncai".to_string(),
                identity: Some("tester.example.test".to_string()),
                text: text.to_string(),
            })
        };
        let client = Arc::new(MockAtircClient {
            identity,
            events: Mutex::new(Some(vec![
                AtircEvent::Connected(connection),
                message("`NO_REPLY`"),
            ])),
            published: Mutex::new(Vec::new()),
        });

        run_atirc(
            test_cli(temp.path()),
            AtircArgs {
                token: None,
                channels: vec!["#test-asyncai".to_string()],
                cursor: -1,
                max_turns: 1,
                mention_only: false,
                verbose: false,
            },
            ToolSetup::default(),
            client.clone(),
        )
        .await
        .unwrap();

        assert!(client.published.lock().unwrap().is_empty());
        assert!(crate::atirc::is_no_reply("NO_REPLY."));
        assert!(crate::atirc::is_no_reply("nothing new\nNO_REPLY"));
        assert!(crate::atirc::is_no_reply("  "));
        assert!(!crate::atirc::is_no_reply("reply NO_REPLY if idle"));
    }

    #[tokio::test]
    async fn no_stdin_runs_monitors_without_user_records() {
        let temp = tempfile::tempdir().unwrap();
        let mut cli = test_cli(temp.path());
        cli.no_stdin = true;
        cli.monitor = vec!["printf 'monitor-only\\n'".to_string()];

        run_interactive(cli, ToolSetup::default()).await.unwrap();

        let records = transcript_records(temp.path());
        assert!(records.iter().any(|record| {
            record["type"] == "queue-operation" && record["operation"] == "enqueue"
        }));
        assert!(!records.iter().any(|record| record["type"] == "user"));
    }
}