supercode_harness/

mail_send.rs

//! One send for every sender: `supercode message send`, the native agent's
//! `send_message`, and a request arriving through another machine's mail
//! door. It resolves the receiver, routes to another machine through Teams or
//! to the router's door here, guards against loops and repeats, and words
//! the outcome for the sending agent.

use std::path::PathBuf;

use crate::mail_route::{Caller, Door, LiveSessions, Unresolved};
use crate::mailbox::{
    local_machine_name, mail_root, Envelope, MailAddress, MailKind, Mailbox, ReplyVia,
};
use crate::HarnessHomes;

/// Exit code of a refused send.
pub const EXIT_REFUSED: i32 = 2;
/// Exit code of an unknown or stale receiver.
pub const EXIT_UNKNOWN: i32 = 3;
/// Exit code of a message stored with no door to show it.
pub const EXIT_STORED: i32 = 4;
/// Exit code of a failed send.
pub const EXIT_FAILED: i32 = 5;

/// What a delivery came to: the exit code, and the text the sending agent reads.
pub struct Outcome {
    /// Exit code for scripts: 0 sent, [`EXIT_REFUSED`], [`EXIT_UNKNOWN`],
    /// [`EXIT_STORED`] or [`EXIT_FAILED`].
    pub code: i32,
    /// What the sending agent reads.
    pub text: String,
    /// Native delivery evidence, or the retained terminal queue state.
    pub receipt: Option<serde_json::Value>,
}

impl Outcome {
    /// An outcome with this code and text.
    pub fn new(code: i32, text: impl Into<String>) -> Self {
        Self {
            code,
            text: text.into(),
            receipt: None,
        }
    }
}

/// Options of one send.
#[derive(Debug, Clone, Default)]
pub struct SendOptions {
    /// A short subject, separate from the unchanged message body.
    pub subject: Option<String>,
    /// Id of the message this one answers.
    pub in_reply_to: Option<String>,
    /// Also send one notice when the receiver's next turn ends.
    pub notify_when_idle: bool,
    /// Only enqueue; never start a turn in an idle receiver.
    pub queue: bool,
    /// Idempotency key: repeating a send with it does not send twice.
    pub idempotency_key: Option<String>,
    /// An automated notice (a Teams transition), not a person's or a session's message:
    /// filed as `notice`, with nothing to answer.
    pub notice: bool,
}

/// Send `body` from `caller` to `to` (an address, `name@machine`, or a name
/// unique on this machine): through Teams when `to` is on another machine,
/// else through the router's door here. The one send every sender uses.
pub async fn send(
    homes: &HarnessHomes,
    caller: &Caller,
    to: &str,
    body: &str,
    options: SendOptions,
) -> std::io::Result<Outcome> {
    let SendOptions {
        subject,
        in_reply_to,
        notify_when_idle,
        queue,
        idempotency_key,
        notice,
    } = options;
    if let Some(machine) = remote_machine(to) {
        let request = serde_json::json!({
            "op": "send",
            "from": caller.address.to_string(),
            "from_name": caller.name,
            "to": to,
            "body": body,
            "subject": subject,
            "in_reply_to": in_reply_to,
            "notify_when_idle": notify_when_idle,
            "queue": queue,
            "id": idempotency_key,
            "notice": notice,
        });
        return Ok(remote_send(&machine, &request));
    }
    deliver(
        homes,
        caller,
        to,
        body,
        subject,
        in_reply_to,
        notify_when_idle,
        queue,
        idempotency_key,
        notice,
    )
    .await
}

/// The machine `to` names when it is not this one.
fn remote_machine(to: &str) -> Option<String> {
    let local = local_machine_name();
    let machine = match MailAddress::parse(to) {
        Ok(address) => address.machine,
        Err(_) => to.rsplit_once('@')?.1.to_string(),
    };
    (machine != local).then_some(machine)
}

/// Hand a request to another machine's mail door through Teams.
fn remote_send(machine: &str, request: &serde_json::Value) -> Outcome {
    match crate::mailbox::teams_mail(machine, request) {
        Ok(answer) => {
            let mut outcome = Outcome::new(
                answer["code"]
                    .as_i64()
                    .map_or(EXIT_FAILED, |code| code as i32),
                answer["text"]
                    .as_str()
                    .or_else(|| answer["detail"].as_str())
                    .unwrap_or("the other machine answered nothing readable")
                    .to_string(),
            );
            outcome.receipt = answer.get("receipt").cloned();
            outcome
        }
        Err(detail) if detail.contains("no mail grant") => Outcome::new(
            EXIT_REFUSED,
            format!(
                "Not sent: machine {machine} does not accept mail from you (no mail grant). Only \
                 an operator can grant it; tell your user. Don't retry."
            ),
        ),
        Err(detail) => Outcome::new(
            EXIT_FAILED,
            format!("Not sent to machine {machine}: {detail}"),
        ),
    }
}

/// Deliver from `caller` to `to` on this machine, through the one router
/// every sender uses (`crate::mail_route`).
#[allow(clippy::too_many_arguments)]
async fn deliver(
    homes: &HarnessHomes,
    caller: &Caller,
    to: &str,
    body: &str,
    subject: Option<String>,
    in_reply_to: Option<String>,
    notify_when_idle: bool,
    queue: bool,
    idempotency_key: Option<String>,
    notice: bool,
) -> std::io::Result<Outcome> {
    use crate::mail_route::{deliver as route, door_for, Delivered, Refused};
    // An operator (a board, a script) is not a listed session; its address is
    // taken as given.
    // `operator:<name>` is that operator on this machine: a tool files what it posts on the
    // session's behalf without knowing the machine's name.
    let local_operator = to
        .strip_prefix("operator:")
        .and_then(|name| MailAddress::new(local_machine_name(), "operator", name).ok());
    let (address, name) = match local_operator
        .map(Ok)
        .unwrap_or_else(|| MailAddress::parse(to))
    {
        // a board is an operator too: its dispatcher reads its mailbox (answers to the questions it asks)
        Ok(address) if address.harness == "operator" || address.harness == "board" => {
            let name = format!("{}@{}", address.session_id, address.machine);
            (address, name)
        }
        // an agent's one mailbox: the mailbox decides which of its sessions receives it
        Ok(address) if address.harness == crate::mail_agent::AGENT_HARNESS => {
            if crate::mail_agent::load(&address.session_id).is_none() {
                return Ok(Outcome::new(
                    EXIT_UNKNOWN,
                    format!(
                        "No agent named {} is declared on this machine. Nothing was sent. Run \
                         supercode agent show for the declared agents.",
                        address.session_id
                    ),
                ));
            }
            let name = format!("agent {}@{}", address.session_id, address.machine);
            (address, name)
        }
        _ => match LiveSessions::read(homes).resolve(to) {
            Ok(session) => (session.address.clone(), session.name.clone()),
            Err(Unresolved::Stale(message) | Unresolved::Unknown(message)) => {
                return Ok(Outcome::new(EXIT_UNKNOWN, message));
            }
        },
    };
    if address == caller.address {
        return Ok(Outcome::new(
            EXIT_REFUSED,
            "Not sent: that address is your own session.",
        ));
    }
    // A message an agent's mailbox routes (a thread's reply, a root to an agent) may reach a
    // session that has stopped: it is resumed. Anything else needs its receiver running.
    let door = door_for(homes, &address).ok();
    let message_id = crate::mail_file::message_id_for(&caller.address, idempotency_key.as_deref())?;
    let sent_marker = sent_marker(&caller.address, &message_id);
    let terminal = matches!(&door, Some(Door::Hook { pane: Some(_), .. }));
    if let Ok(previous) = std::fs::read_to_string(&sent_marker) {
        if terminal && !queue {
            Mailbox::open(&mail_root(), &address)?.request_wake(&message_id)?;
        }
        let mut outcome = Outcome::new(
            0,
            format!(
                "Already sent with --id {}: {previous} Nothing was sent again.",
                idempotency_key.as_deref().unwrap_or_default()
            ),
        );
        outcome.receipt = Some(delivery_receipt(&address, &message_id, terminal));
        return Ok(outcome);
    }
    // An answered message is named as the agent was shown it, by a prefix of its id: the one in
    // the caller's mailbox it names.
    // A prefix names the one message it matches there, or nothing is sent; a whole id is kept as
    // given (it may answer a message filed elsewhere).
    let in_reply_to = match in_reply_to {
        Some(answered) if answered.len() < FULL_ID_LEN => {
            let found = Mailbox::open(&mail_root(), &caller.address)
                .and_then(|mailbox| mailbox.find_prefix(&answered))
                .unwrap_or_default();
            match found.as_slice() {
                [only] => Some(only.envelope.id.clone()),
                [] => {
                    return Ok(Outcome::new(
                        EXIT_REFUSED,
                        format!(
                            "Not sent: no message in your mailbox has an id starting {answered}. \
                             Name the message you answer as its envelope shows it."
                        ),
                    ))
                }
                many => {
                    let ids: Vec<&str> = many
                        .iter()
                        .map(|stored| stored.envelope.id.as_str())
                        .collect();
                    return Ok(Outcome::new(
                        EXIT_REFUSED,
                        format!(
                            "Not sent: {answered} names {} messages in your mailbox ({}). Give \
                             more of its id.",
                            ids.len(),
                            ids.join(", ")
                        ),
                    ));
                }
            }
        }
        other => other,
    };
    if let Some(answered) = in_reply_to.as_deref().filter(|id| id.starts_with("q-")) {
        return crate::mail_question::reply(caller, &address, answered, body).await;
    }
    if let Some(answered) = in_reply_to.as_deref() {
        if reply_chain_depth(&caller.address, &address, answered) >= REPLY_CHAIN_LIMIT {
            return Ok(Outcome::new(
                EXIT_REFUSED,
                format!(
                    "Not sent: you and {name} have answered each other {REPLY_CHAIN_LIMIT} times \
                     in a row. If you are trading acknowledgements or status, stop; to go on, \
                     send a new message (without --re), or tell your user."
                ),
            ));
        }
    }
    let (kind, reply_via) = if notice {
        (MailKind::Notice, ReplyVia::None)
    } else {
        (MailKind::Peer, ReplyVia::Command)
    };
    let mut envelope = Envelope::new(
        caller.address.clone(),
        caller.name.clone(),
        kind,
        reply_via,
        body,
    )?;
    envelope.subject = subject.as_deref().map(|value| {
        value
            .lines()
            .next()
            .unwrap_or("")
            .chars()
            .take(200)
            .collect()
    });
    envelope.id = message_id.clone();
    envelope.thread = crate::mailbox::thread_of_reply(in_reply_to.as_deref());
    envelope.in_reply_to = in_reply_to;
    if let Some(plan) = crate::mail_agent::plan(
        &mut envelope,
        &address,
        &crate::mail_agent::Channel::default(),
    )? {
        return Ok(deliver_planned(homes, caller, &envelope, &plan, queue, notify_when_idle).await);
    }
    let Some(door) = door else {
        return Ok(Outcome::new(
            EXIT_UNKNOWN,
            format!(
                "{name} is no longer running. Nothing was sent. Run supercode message list for \
                 the live sessions."
            ),
        ));
    };
    let tier = door.name();
    let idle_note = if notify_when_idle {
        " Subscribed: one idle notice reaches you when its next turn ends."
    } else {
        ""
    };
    let (code, what) = match route(&envelope, &address, &door, !queue, notify_when_idle).await {
        Err(detail) => {
            return Ok(Outcome::new(
                EXIT_FAILED,
                format!("Not sent to {name}: {detail}"),
            ))
        }
        Ok(Err(Refused::CannotQueueNative)) => {
            return Ok(Outcome::new(
                EXIT_REFUSED,
                format!(
                    "Not sent: {name} is idle, and a Claude session always starts a turn when a \
                     message arrives, so --queue cannot hold it. Send without --queue to wake it."
                ),
            ))
        }
        Ok(Err(Refused::TooLong(bytes))) => {
            return Ok(Outcome::new(
                EXIT_REFUSED,
                format!(
                    "Not sent: the message is {bytes} bytes; the limit is {}. Write it to a file \
                     and send its path instead.",
                    crate::mail_route::MAX_RELAYED_BYTES
                ),
            ))
        }
        Ok(Ok(Delivered::Steered)) => (0, "steered into its running turn"),
        Ok(Ok(Delivered::Started)) => (0, "it was idle, so the message started a turn"),
        Ok(Ok(Delivered::Native { busy: true })) => (0, "it will read it at its next tool call"),
        Ok(Ok(Delivered::Native { busy: false })) => {
            (0, "it was idle, so the message starts its next turn")
        }
        Ok(Ok(Delivered::Hooked)) => (
            0,
            "queued in its mailbox; terminal delivery waits for native transcript confirmation",
        ),
        Ok(Ok(Delivered::HookWoken)) => (
            0,
            "it was idle, so its pane was told to read its mailbox; its hook shows it the message in \
             that turn",
        ),
        Ok(Ok(Delivered::Queued)) => (
            0,
            "it is idle and --queue leaves it so; the message waits in its mailbox",
        ),
        Ok(Ok(Delivered::Operator)) => (0, "filed in its mailbox"),
        Ok(Ok(Delivered::Already)) => (0, "it already had this message; nothing was sent again"),
        Ok(Ok(Delivered::Stored)) => (
            EXIT_STORED,
            "Stored (not failed): it has no delivery door, so it sees this only if it runs \
             supercode message inbox (a Codex session gets one with: supercode message setup \
             codex). Don't resend, and don't wait for a reply",
        ),
    };
    record_send(&caller.address, &address, &message_id);
    let shown_id = crate::mailbox::short_id(&message_id);
    let text = if code == 0 {
        format!(
            "sent to {name} ({}, {tier}): {what}. Message id {shown_id}.{idle_note} Don't poll; \
             carry on.",
            address.harness
        )
    } else {
        format!(
            "{name} ({}): {what}. Message id {shown_id}.{idle_note}",
            address.harness
        )
    };
    if code == 0 && idempotency_key.is_some() {
        if let Some(parent) = sent_marker.parent() {
            std::fs::create_dir_all(parent).ok();
        }
        std::fs::write(&sent_marker, &text).ok();
    }
    let mut outcome = Outcome::new(code, text);
    if code == 0 {
        outcome.receipt = Some(delivery_receipt(&address, &message_id, terminal));
    }
    Ok(outcome)
}

/// Deliver `envelope` to every receiver `plan` names: a `to` receiver through its door (a
/// stopped session is resumed first), a `cc` receiver filed unread and not woken. The one
/// delivery every agent-routed message takes, whichever door it came in by.
pub async fn deliver_planned(
    homes: &HarnessHomes,
    caller: &Caller,
    envelope: &Envelope,
    plan: &crate::mail_agent::Plan,
    queue: bool,
    notify_when_idle: bool,
) -> Outcome {
    let mut reached: Vec<String> = Vec::new();
    let mut copied: Vec<String> = Vec::new();
    let mut failed: Vec<String> = Vec::new();
    if let Some(agent) = &plan.agent {
        // The agent's own mailbox keeps the record of what was addressed to it.
        if let Err(error) =
            Mailbox::open(&mail_root(), agent).and_then(|mailbox| mailbox.deliver_read(envelope))
        {
            failed.push(format!("the agent's mailbox ({error})"));
        }
    }
    for recipient in plan
        .recipients
        .iter()
        .filter(|recipient| recipient.address != caller.address)
    {
        let label = recipient_label(homes, &recipient.address);
        record_send(&caller.address, &recipient.address, &envelope.id);
        if !recipient.wake {
            match crate::mail_agent::file_unread(&recipient.address, envelope) {
                Ok(()) => copied.push(label),
                Err(error) => failed.push(format!("{label} ({error})")),
            }
            continue;
        }
        match deliver_woken(homes, envelope, &recipient.address, queue, notify_when_idle).await {
            Ok(how) => reached.push(format!("{label} ({how})")),
            Err(detail) => failed.push(format!("{label} ({detail})")),
        }
    }
    let shown_id = crate::mailbox::short_id(&envelope.id);
    let thread = plan
        .thread
        .as_ref()
        .map(|thread| format!(" Thread {}.", crate::mailbox::short_id(&thread.id)))
        .unwrap_or_default();
    let mut text = if reached.is_empty() && copied.is_empty() {
        format!("Not sent: nobody in its thread could be reached.{thread}")
    } else {
        let cc = if copied.is_empty() {
            String::new()
        } else {
            format!(" CC (filed, not woken): {}.", copied.join(", "))
        };
        format!(
            "sent to {}.{cc} Message id {shown_id}.{thread} Don't poll; carry on.",
            if reached.is_empty() {
                "no one woken".to_string()
            } else {
                reached.join(", ")
            }
        )
    };
    if !failed.is_empty() {
        text.push_str(&format!(" Not delivered to: {}.", failed.join("; ")));
    }
    let code = if reached.is_empty() && copied.is_empty() {
        EXIT_FAILED
    } else {
        0
    };
    let mut outcome = Outcome::new(code, text);
    if code == 0 {
        outcome.receipt = Some(
            serde_json::json!({"delivered": true, "message_ids": [envelope.id], "receipt": "agent-mailbox"}),
        );
    }
    outcome
}

/// What a receiver is called in an outcome: its session name when it runs here, else its address.
fn recipient_label(homes: &HarnessHomes, address: &MailAddress) -> String {
    if address.harness == "operator" || address.harness == "board" {
        return format!("{}@{}", address.session_id, address.machine);
    }
    LiveSessions::read(homes)
        .all()
        .iter()
        .find(|session| &session.address == address)
        .map(|session| session.name.clone())
        .unwrap_or_else(|| address.to_string())
}

/// Deliver `envelope` to `to` through its door, waking it: a session on another machine
/// through Teams, a stopped one here resumed first. Says how it went.
async fn deliver_woken(
    homes: &HarnessHomes,
    envelope: &Envelope,
    to: &MailAddress,
    queue: bool,
    notify_when_idle: bool,
) -> Result<&'static str, String> {
    use crate::mail_route::{deliver as route, door_for, Delivered, NoDoor};
    let door = match door_for(homes, to) {
        Ok(door) => door,
        Err(NoDoor::OtherMachine(_)) => {
            crate::mailbox::deliver_to(to, envelope).map_err(|error| error.to_string())?;
            return Ok("filed on its machine");
        }
        Err(NoDoor::NotRunning) => {
            // A reply to a closed session resumes it; it is never a new root. Only a session the
            // mailbox launched is resumed, the way it was launched; any other waits with its mail.
            crate::mail_agent::file_unread(to, envelope).map_err(|error| error.to_string())?;
            if !crate::mail_agent::resumable(to) {
                return Ok(
                    "stopped; not resumed, since the mailbox did not launch it: the \
                           message waits in its mailbox",
                );
            }
            crate::mail_agent::resume(to).map_err(|error| format!("not resumed: {error}"))?;
            let started = std::time::Instant::now();
            loop {
                if let Ok(door) = door_for(homes, to) {
                    break door;
                }
                if started.elapsed() > crate::mail_agent::RESUME_WAIT {
                    return Ok("resumed; the message waits in its mailbox");
                }
                tokio::time::sleep(std::time::Duration::from_secs(1)).await;
            }
        }
    };
    match route(envelope, to, &door, !queue, notify_when_idle).await? {
        Ok(Delivered::Steered) => Ok("steered into its running turn"),
        Ok(Delivered::Started) => Ok("started a turn"),
        Ok(Delivered::Native { busy: true }) => Ok("read at its next tool call"),
        Ok(Delivered::Native { busy: false }) => Ok("starts its next turn"),
        Ok(Delivered::Hooked | Delivered::HookWoken) => Ok("in its mailbox, its hook shows it"),
        Ok(Delivered::Queued) => Ok("waits in its mailbox"),
        Ok(Delivered::Operator) => Ok("filed"),
        Ok(Delivered::Already) => Ok("it already had it"),
        Ok(Delivered::Stored) => Ok("stored; it has no delivery door"),
        Err(crate::mail_route::Refused::CannotQueueNative) => {
            Err("idle, and --queue cannot hold a Claude session's mail".into())
        }
        Err(crate::mail_route::Refused::TooLong(bytes)) => Err(format!(
            "{bytes} bytes; the limit is {}",
            crate::mail_route::MAX_RELAYED_BYTES
        )),
    }
}

/// Replies in one unbroken chain between two sessions beyond which another
/// reply is refused: two agents trading acknowledgements never stop on their
/// own, and every message of such a loop answers the one before it. A new
/// message (no `--re`) starts a chain, so volume alone is never refused.
const REPLY_CHAIN_LIMIT: usize = 32;

/// Length of a whole message id: its kind, `-`, and 24 hex digits.
const FULL_ID_LEN: usize = 26;

/// How many messages deep the reply chain ending at `answered` runs, between
/// `a` and `b`: each message's `in_reply_to`, followed back through both
/// sessions' mailboxes.
fn reply_chain_depth(a: &MailAddress, b: &MailAddress, answered: &str) -> usize {
    let mut links = std::collections::HashMap::new();
    for address in [a, b] {
        let Ok(mailbox) = Mailbox::open(&mail_root(), address) else {
            continue;
        };
        for stored in mailbox.list().unwrap_or_default() {
            links.insert(
                stored.envelope.id.clone(),
                stored.envelope.in_reply_to.clone(),
            );
        }
    }
    let mut depth = 0;
    let mut current = Some(answered.to_string());
    while let Some(id) = current {
        if depth > REPLY_CHAIN_LIMIT || !links.contains_key(&id) {
            break;
        }
        depth += 1;
        current = links.get(&id).cloned().flatten();
    }
    depth
}

fn send_log(sender: &MailAddress) -> PathBuf {
    let hash = blake3::hash(sender.to_string().as_bytes()).to_hex();
    mail_root().join("sent").join(&hash[..24]).join("log.jsonl")
}

fn record_send(sender: &MailAddress, to: &MailAddress, message_id: &str) {
    let log = send_log(sender);
    if let Some(parent) = log.parent() {
        std::fs::create_dir_all(parent).ok();
    }
    if let Ok(mut file) = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(&log)
    {
        use std::io::Write as _;
        let entry = serde_json::json!({"to": to.to_string(), "at_ms": now_ms(), "id": message_id});
        writeln!(file, "{entry}").ok();
    }
}

fn now_ms() -> u64 {
    std::time::SystemTime::now()
        .duration_since(std::time::UNIX_EPOCH)
        .map(|elapsed| elapsed.as_millis() as u64)
        .unwrap_or_default()
}

/// Marker recording that `message_id` was sent by `sender`, for `--id`.
fn sent_marker(sender: &MailAddress, message_id: &str) -> PathBuf {
    let hash = blake3::hash(sender.to_string().as_bytes()).to_hex();
    mail_root().join("sent").join(&hash[..24]).join(message_id)
}

/// Terminal acknowledgements come from the daemon's retained mailbox receipts,
/// not the fact that an envelope was filed or a composer became empty.
fn delivery_receipt(address: &MailAddress, id: &str, terminal: bool) -> serde_json::Value {
    if !terminal {
        return serde_json::json!({"delivered": true, "message_ids": [id], "receipt": "native-mail"});
    }
    let recipient = format!("{}:{}", address.harness, address.session_id);
    if let Ok(entries) = std::fs::read_dir(mail_root().join("terminal-delivery")) {
        for entry in entries.flatten() {
            let Ok(bytes) = std::fs::read(entry.path().join("queue.json")) else {
                continue;
            };
            let Ok(state) = serde_json::from_slice::<serde_json::Value>(&bytes) else {
                continue;
            };
            if state["recipient"].as_str() != Some(recipient.as_str()) {
                continue;
            }
            if let Some(receipt) = state["receipts"].as_array().and_then(|receipts| {
                receipts.iter().find(|r| {
                    r["message_ids"]
                        .as_array()
                        .is_some_and(|ids| ids.iter().any(|value| value.as_str() == Some(id)))
                })
            }) {
                return receipt.clone();
            }
            return serde_json::json!({"delivered": false, "queued": true, "message_ids": [id], "reason": state["active"]["error"]});
        }
    }
    serde_json::json!({"delivered": false, "queued": true, "message_ids": [id], "reason": "awaiting terminal mailbox"})
}