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>,
}

/// 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,
    } = 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,
        });
        return Ok(remote_send(&machine, &request));
    }
    deliver(
        homes,
        caller,
        to,
        body,
        subject,
        in_reply_to,
        notify_when_idle,
        queue,
        idempotency_key,
    )
    .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>,
) -> 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)
        }
        _ => 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.",
        ));
    }
    let door = match door_for(homes, &address) {
        Ok(door) => door,
        Err(_) => {
            return Ok(Outcome::new(
                EXIT_UNKNOWN,
                format!(
                    "{name} is no longer running. Nothing was sent. Run supercode message list for \
                     the live sessions."
                ),
            ))
        }
    };
    let message_id = match &idempotency_key {
        // A key that is itself a whole message id is the message's id: a sender that chose it beforehand (a board
        // blocking a card on the answer to the question it is about to send) knows what the answer will name.
        Some(key)
            if key.len() == 26
                && key.starts_with("m-")
                && key[2..]
                    .bytes()
                    .all(|byte| byte.is_ascii_hexdigit() && !byte.is_ascii_uppercase()) =>
        {
            key.clone()
        }
        Some(key) => format!(
            "m-{}",
            &blake3::hash(format!("{}\0{key}", caller.address).as_bytes()).to_hex()[..24]
        ),
        None => crate::mailbox::new_message_id()?,
    };
    let sent_marker = sent_marker(&caller.address, &message_id);
    let terminal = matches!(&door, 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 mut envelope = Envelope::new(
        caller.address.clone(),
        caller.name.clone(),
        MailKind::Peer,
        ReplyVia::Command,
        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;
    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::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)
}

/// 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"})
}