supercode_harness/

mail_watch.rs

//! Idle notices for every harness, from session activity.
//!
//! Knowing that a session finished does not need control of it: supercode's
//! normalized activity ([`crate::session_activity`]) reports busy/idle for a
//! native Claude session, a running Codex, and any runtime supercode hosts.
//! A sender that asked for `--notify-when-idle` gets one notice when the
//! receiver is seen working after the message and then idle, one when the
//! receiver ends, or one saying the subscription expired after
//! [`IDLE_SUBSCRIPTION_LIFETIME`]. The notice is delivered through the one
//! router ([`crate::mail_route`]), so each subscriber gets it by its own door.

use std::time::Duration;

use crate::mail_route::{deliver, door_for};
use crate::mailbox::{
    mail_root, subscribed_mailboxes, Envelope, IdleSubscription, MailAddress, MailKind, ReplyVia,
};
use crate::session_activity::{SessionActivityMonitor, SessionPresence, SessionTurnState};
use crate::{HarnessHomes, HarnessId, SessionLocator, StorageLocator};

/// How long a subscription waits for the receiver to work and go idle.
pub const IDLE_SUBSCRIPTION_LIFETIME: Duration = Duration::from_secs(24 * 60 * 60);

/// Polls the activity of every session someone subscribed to.
#[derive(Debug, Default)]
pub struct IdleWatcher {
    monitor: SessionActivityMonitor,
}

impl IdleWatcher {
    /// A watcher with no state yet.
    pub fn new() -> Self {
        Self::default()
    }

    /// Whether any subscription is waiting.
    pub fn has_subscriptions() -> bool {
        !subscribed_mailboxes(&mail_root()).is_empty()
    }

    /// One pass: settle every subscription whose receiver finished a turn,
    /// ended, or outlived the subscription.
    pub async fn tick(&mut self, homes: &HarnessHomes) {
        for mailbox in subscribed_mailboxes(&mail_root()) {
            let target = mailbox.address().clone();
            let Ok(subscriptions) = mailbox.subscriptions() else {
                continue;
            };
            if subscriptions.is_empty() {
                continue;
            }
            // A runtime supercode hosts reports its own turn; for the rest,
            // session activity reads the harness's lifecycle.
            let (working, idle, ended) = if subscriptions.iter().any(|s| s.final_reply) {
                match crate::runtime_mail::controlled_runtime(&target.harness, &target.session_id) {
                    Some(record) => match crate::runtime_mail::runtime_turn_state(&record).await {
                        Ok(crate::frontend::FrontendTurnState::Busy) => (true, false, false),
                        Ok(crate::frontend::FrontendTurnState::Idle) => (false, true, false),
                        Err(_) => (false, false, false),
                    },
                    None => (false, false, true),
                }
            } else {
                let locator = locator_for(homes, &target);
                let activity = match self.monitor.resolve(&[locator], homes).await {
                    Ok(mut activities) => activities.pop(),
                    Err(_) => None,
                };
                match &activity {
                    Some(activity) => (
                        activity.turn == SessionTurnState::Working,
                        activity.turn == SessionTurnState::Idle,
                        activity.presence == SessionPresence::Persisted,
                    ),
                    None => (false, false, false),
                }
            };
            let runtime =
                crate::runtime_mail::controlled_runtime(&target.harness, &target.session_id);
            // One turn settles every message a sender subscribed with; the sender hears it once,
            // about its latest message, and its earlier subscriptions settle with it silently.
            let mut subscriptions = subscriptions;
            subscriptions.sort_by(|a, b| b.created_at_ms.cmp(&a.created_at_ms));
            let mut told: std::collections::HashSet<(String, u8)> =
                std::collections::HashSet::new();
            for mut subscription in subscriptions {
                let expired = subscription.age() > IDLE_SUBSCRIPTION_LIFETIME;
                // A reply is the runtime's answer to this message: it settles
                // once the runtime is idle and has answered, however fast
                // the turn was.
                if subscription.final_reply && !ended && !expired {
                    let answer = match (&runtime, idle) {
                        (Some(record), true) => {
                            crate::runtime_mail::answer_to(homes, record, &subscription.message_id)
                        }
                        _ => None,
                    };
                    if let Some(answer) = answer {
                        if mailbox
                            .remove_subscription(&subscription.message_id)
                            .is_ok()
                        {
                            send_reply(homes, &target, &subscription, Some(answer)).await;
                            if subscription.notice {
                                notify(homes, &target, &subscription, Settled::Idle).await;
                            }
                        }
                    }
                    continue;
                }
                let settles = if ended {
                    Some(Settled::Ended)
                } else if expired {
                    Some(Settled::Expired)
                } else if idle && subscription.seen_working {
                    Some(Settled::Idle)
                } else {
                    None
                };
                match settles {
                    Some(settled) => {
                        if mailbox
                            .remove_subscription(&subscription.message_id)
                            .is_ok()
                        {
                            if subscription.final_reply {
                                send_reply(homes, &target, &subscription, None).await;
                            }
                            let first =
                                told.insert((subscription.subscriber.to_string(), settled as u8));
                            if subscription.notice && first {
                                notify(homes, &target, &subscription, settled).await;
                            }
                        }
                    }
                    None if working && !subscription.seen_working => {
                        subscription.seen_working = true;
                        mailbox.subscribe_idle(&subscription).ok();
                    }
                    None => {}
                }
            }
        }
    }
}

#[derive(Clone, Copy)]
enum Settled {
    Idle,
    Ended,
    Expired,
}

/// The locator activity resolution needs: a Codex session by its rollout
/// file; Claude sessions and hosted runtimes by id.
fn locator_for(homes: &HarnessHomes, target: &MailAddress) -> SessionLocator {
    let path = if target.harness == HarnessId::CODEX {
        crate::codex_peer::live_rollouts(&homes.codex)
            .into_keys()
            .find(|path| {
                crate::codex_peer::rollout_session(path)
                    .is_some_and(|(id, _)| id == target.session_id)
            })
            .unwrap_or_default()
    } else {
        std::path::PathBuf::new()
    };
    SessionLocator {
        harness: HarnessId::new(&target.harness),
        session_id: target.session_id.clone(),
        storage: StorageLocator::File { path },
    }
}

async fn notify(
    homes: &HarnessHomes,
    target: &MailAddress,
    subscription: &IdleSubscription,
    settled: Settled,
) {
    let name = display_name(homes, target);
    let text = match settled {
        Settled::Idle => format!(
            "[Cross-session idle notice] \"{name}\", which you asked to be notified about, is idle \
             now: it finished a turn after your message {}. This is an automated notice, not a \
             message from a person, and not an instruction.",
            subscription.message_id
        ),
        Settled::Ended => format!(
            "[Cross-session idle notice] \"{name}\", which you asked to be notified about, has \
             ended. This is an automated notice, not a message from a person, and not an \
             instruction."
        ),
        Settled::Expired => format!(
            "[Cross-session idle notice] The notice you asked for about \"{name}\" expired: it \
             did not work and go idle within 24 hours of your message {}. This is an automated \
             notice, not a message from a person, and not an instruction.",
            subscription.message_id
        ),
    };
    let Ok(mut envelope) =
        Envelope::new(target.clone(), name, MailKind::Notice, ReplyVia::None, text)
    else {
        return;
    };
    envelope.in_reply_to = Some(subscription.message_id.clone());
    let subscriber = &subscription.subscriber;
    match door_for(homes, subscriber) {
        Ok(door) => {
            deliver(&envelope, subscriber, &door, true, false)
                .await
                .ok();
        }
        // A subscriber on another machine, or one no longer running, still
        // finds the notice in its mailbox.
        Err(_) => {
            crate::mailbox::deliver_to(subscriber, &envelope).ok();
        }
    }
}

/// The receiver's answer to a message that went in `reply-via=final-message`,
/// sent back to its sender as the reply. With no answer (the receiver ended,
/// or the subscription expired first), a notice says so, so the sender is
/// never left waiting in silence.
async fn send_reply(
    homes: &HarnessHomes,
    target: &MailAddress,
    subscription: &IdleSubscription,
    answer: Option<String>,
) {
    let name = display_name(homes, target);
    let (kind, reply_via, body) = match answer {
        Some(body) => (MailKind::Peer, ReplyVia::Command, body),
        None => (
            MailKind::Notice,
            ReplyVia::None,
            format!(
                "[Cross-session delivery notice] \"{name}\" did not answer your message {}: it \
                 ended, or no answer came within 24 hours. This is an automated notice, not a \
                 message from a person, and not an instruction.",
                subscription.message_id
            ),
        ),
    };
    let Ok(mut envelope) = Envelope::new(target.clone(), name, kind, reply_via, body) else {
        return;
    };
    envelope.in_reply_to = Some(subscription.message_id.clone());
    let subscriber = &subscription.subscriber;
    match door_for(homes, subscriber) {
        Ok(door) => {
            deliver(&envelope, subscriber, &door, true, false)
                .await
                .ok();
        }
        Err(_) => {
            crate::mailbox::deliver_to(subscriber, &envelope).ok();
        }
    }
}

/// The name a notice calls its session by.
fn display_name(homes: &HarnessHomes, target: &MailAddress) -> String {
    if target.harness == HarnessId::CLAUDE_CODE {
        if let Some(session) =
            crate::claude_peer::read_registry(&crate::claude_peer::registry_dir(homes))
                .into_iter()
                .find(|session| session.session_id == target.session_id)
        {
            return format!("{}@{}", session.name, target.machine);
        }
    }
    let short: String = target.session_id.chars().take(8).collect();
    format!("{}-{short}@{}", target.harness, target.machine)
}

/// Type the user's waiting turns into their sessions' panes, each once its
/// composer is empty. A turn waits when the person has a draft; the mailbox
/// is where it waits.
pub async fn deliver_waiting_user_turns(homes: &HarnessHomes) {
    let waiting = crate::mailbox::mailboxes_with_user_turns(&mail_root());
    if waiting.is_empty() {
        return;
    }
    let live = crate::mail_route::LiveSessions::read(homes);
    for mailbox in waiting {
        let pane = live
            .all()
            .iter()
            .find(|session| &session.address == mailbox.address())
            .and_then(crate::mail_route::daemon_pane);
        if let Some(pane) = pane {
            crate::mail_route::type_user_turns(&mailbox, &pane).await;
        }
    }
}

/// How long a wake waits after a failed attempt, at first and at most (doubling between).
const FIRST_RETRY: Duration = Duration::from_secs(2);
const LAST_RETRY: Duration = Duration::from_secs(300);
/// Counted failures a wake has at least before it may expire.
pub const WAKE_ATTEMPTS: u32 = 8;
/// How long a wake to a session on this machine keeps trying, from its first counted failure. Its basis is a session
/// restart: the board relaunches or resumes a lost card session within its launch window (300 s, `launchWindowSeconds`),
/// the machine daemon restarts its watch loop within 5 s, and a person resuming a closed session does so within
/// minutes; thirty minutes outlasts several of those. Past it the session is not coming back soon, and the message
/// waits unread for whenever it does (a session reads its unread mail first).
pub const LOCAL_WAKE_LIFETIME: Duration = Duration::from_secs(30 * 60);
/// How long a wake to another machine keeps trying, from its first counted failure (a refusal by that machine or by
/// Teams, or Teams unreachable). Three days outlasts a weekend with the server down. A refusal saying the machine is
/// offline (a reboot, a closed lid) is never counted: that mail waits for the machine, however long, and is carried
/// within [`LAST_RETRY`] of its link coming back.
pub const REMOTE_WAKE_LIFETIME: Duration = Duration::from_secs(3 * 24 * 60 * 60);

/// The mailbox's own delivery (D140): every message filed with a wake (by a filer that does not wait, such as a board's
/// dispatcher, or by a send to a hooked session) is handed to its session by the session's own door, by the daemon that
/// runs this watch loop. Mail for another machine is carried there through Teams (`supercode teams mail`), the only
/// cross-machine leg, and filed with the same wake, so that machine's watch loop delivers it. A failed attempt waits
/// [`FIRST_RETRY`] doubling to [`LAST_RETRY`], kept in the wake itself, so a restarted watcher waits as the last one
/// did. A wake expires only once it has failed [`WAKE_ATTEMPTS`] counted times and its first counted failure is older
/// than its bound ([`LOCAL_WAKE_LIFETIME`] here, [`REMOTE_WAKE_LIFETIME`] for another machine): the message stays filed
/// and unread, and the expiry is recorded in its mailbox (`supercode message expired`), which the board's diagnostics
/// read. Mail for a machine that is offline is not failing: it is offered again every [`LAST_RETRY`] until the machine
/// links, and never expires for that.
#[derive(Debug, Default)]
pub struct MailCarrier;

/// What one attempt at a wake came to.
enum Carried {
    /// Delivered: the wake is done and the message is read.
    Delivered,
    /// Filed where its reader looks for it (an operator's mailbox, a hook that reads it at the next tool call, a
    /// session with no door): the wake is done and the message stays unread for its reader.
    Filed,
    /// Not now, and why (its session is not running, the attempt failed): a counted failure, tried again after a wait.
    Later(String),
    /// Its machine is offline (not linked to Teams): not a failure; offered again at [`LAST_RETRY`].
    Offline(String),
}

/// Whether a cross-machine answer says the receiving machine is offline (not linked), rather than refusing the mail.
fn offline(text: &str) -> bool {
    text.contains("machine_offline")
}

impl MailCarrier {
    /// A carrier.
    pub fn new() -> Self {
        Self
    }

    /// One pass over every mailbox holding a wake that is due.
    pub async fn tick(&mut self, homes: &HarnessHomes) {
        let local = crate::mailbox::local_machine_name();
        let now = crate::mailbox::now_ms();
        let mut offline_now: std::collections::HashMap<String, String> =
            std::collections::HashMap::new();
        for mailbox in crate::mailbox::mailboxes_with_wake_requests(&mail_root()) {
            let Ok(pending) = mailbox.pending_wakes() else {
                continue;
            };
            let address = mailbox.address().clone();
            let pending: Vec<String> = pending
                .into_iter()
                .filter(|id| mailbox.wake_state(id).next_at_ms <= now)
                .collect();
            if pending.is_empty() {
                continue;
            }
            // a hooked session in a daemon pane is woken once for all its waiting mail, its hook showing it all
            if address.machine == local {
                if let Ok(crate::mail_route::Door::Hook {
                    pane: Some(pane), ..
                }) = door_for(homes, &address)
                {
                    let woken =
                        crate::mail_route::wake_hook_mailbox(&mailbox, &pane, &pending).await;
                    for id in pending {
                        let carried = match &woken {
                            Ok(true) => Carried::Delivered,
                            Ok(false) => Carried::Later("its pane was not at its prompt".into()),
                            Err(error) => Carried::Later(error.clone()),
                        };
                        settle(&mailbox, &id, carried, &local);
                    }
                    continue;
                }
            }
            for id in pending {
                // a machine found offline in this pass is not asked again for its other mail
                if let Some(reason) = offline_now.get(&address.machine) {
                    settle(&mailbox, &id, Carried::Offline(reason.clone()), &local);
                    continue;
                }
                let carried = match mailbox.find(&id) {
                    // its recipient has it already (its own read, or a hand-over by an attempt that ended before it was settled):
                    // never delivered twice; a read by anyone else does not count
                    Ok(Some(stored))
                        if stored.state == crate::mailbox::MailState::Read
                            && mailbox.delivered_to_recipient(&stored.envelope.id) =>
                    {
                        Carried::Filed
                    }
                    Ok(Some(stored)) => carry(homes, &address, stored.envelope, &local).await,
                    _ => Carried::Filed,
                };
                if let Carried::Offline(reason) = &carried {
                    offline_now.insert(address.machine.clone(), reason.clone());
                }
                settle(&mailbox, &id, carried, &local);
            }
        }
    }
}

/// One message toward its reader.
async fn carry(
    homes: &HarnessHomes,
    to: &MailAddress,
    mut envelope: Envelope,
    local: &str,
) -> Carried {
    if to.machine != local {
        let request = serde_json::json!({"op": "file", "to": to.to_string(), "envelope": envelope, "wake": true});
        return match crate::mailbox::teams_mail(&to.machine, &request) {
            Ok(answer) if answer["code"].as_i64() == Some(0) => Carried::Delivered,
            Ok(answer) => {
                let reason = format!(
                    "{} did not take it: {}",
                    to.machine,
                    answer["text"]
                        .as_str()
                        .or(answer["detail"].as_str())
                        .unwrap_or("no reason given")
                );
                if offline(&answer.to_string()) {
                    Carried::Offline(reason)
                } else {
                    Carried::Later(reason)
                }
            }
            Err(error) if offline(&error) => {
                Carried::Offline(format!("{} is offline: {error}", to.machine))
            }
            Err(error) => {
                Carried::Later(format!("Teams did not carry it to {}: {error}", to.machine))
            }
        };
    }
    // an agent's one mailbox: its plan decides which of its sessions receives it (ADR 0008 decisions 5-9)
    if to.harness == crate::mail_agent::AGENT_HARNESS {
        let plan =
            crate::mail_agent::plan(&mut envelope, to, &crate::mail_agent::Channel::default());
        let Ok(Some(plan)) = plan else {
            return Carried::Filed;
        };
        let caller = crate::mail_route::Caller {
            address: envelope.from.clone(),
            name: envelope.from_name.clone(),
        };
        let outcome =
            crate::mail_send::deliver_planned(homes, &caller, &envelope, &plan, false, false).await;
        return if outcome.code == 0 {
            Carried::Delivered
        } else {
            Carried::Later(outcome.text)
        };
    }
    match door_for(homes, to) {
        Ok(door @ (crate::mail_route::Door::Native(_) | crate::mail_route::Door::Runtime(_))) => {
            // Idempotent by the message's id across a crash: the handover is recorded before it starts, and a wake whose
            // last handover never settled (its watcher died in it) is not handed over again; it stays unread, which its
            // session reads first.
            let Ok(mailbox) = crate::mailbox::Mailbox::open(&mail_root(), to) else {
                return Carried::Later("its mailbox could not be opened".into());
            };
            let mut state = mailbox.wake_state(&envelope.id);
            if state.handover_at_ms.is_some() {
                return Carried::Filed;
            }
            state.handover_at_ms = Some(crate::mailbox::now_ms());
            if mailbox.set_wake_state(&envelope.id, &state).is_err() {
                return Carried::Later("its handover could not be recorded".into());
            }
            let delivered = deliver(&envelope, to, &door, true, false).await;
            state.handover_at_ms = None;
            mailbox.set_wake_state(&envelope.id, &state).ok();
            match delivered {
                Ok(Ok(_)) => Carried::Delivered,
                // a message its door can never take (too long for a relay) stays filed for the session to read
                Ok(Err(_)) => Carried::Filed,
                Err(error) => Carried::Later(error),
            }
        }
        Ok(_) => Carried::Filed,
        Err(crate::mail_route::NoDoor::NotRunning) => {
            Carried::Later("its session is not running".into())
        }
        Err(crate::mail_route::NoDoor::OtherMachine(machine)) => {
            Carried::Later(format!("it is on {machine}"))
        }
    }
}

/// Record how a wake went: done, waiting for its next attempt (an offline machine's mail uncounted), or expired at its
/// bound.
fn settle(mailbox: &crate::mailbox::Mailbox, id: &str, carried: Carried, local: &str) {
    match carried {
        Carried::Delivered | Carried::Filed => {
            mailbox.acknowledge_wake(id);
            if matches!(carried, Carried::Delivered) {
                if let Ok(Some(stored)) = mailbox.find(id) {
                    mailbox.mark_read(&stored).ok();
                }
            }
        }
        Carried::Offline(reason) => {
            let mut state = mailbox.wake_state(id);
            state.next_at_ms = crate::mailbox::now_ms() + LAST_RETRY.as_millis() as u64;
            state.last_error = Some(reason);
            mailbox.set_wake_state(id, &state).ok();
        }
        Carried::Later(reason) => {
            let now = crate::mailbox::now_ms();
            let mut state = mailbox.wake_state(id);
            state.attempts += 1;
            let since = *state.failing_since_ms.get_or_insert(now);
            let lifetime = if mailbox.address().machine == local {
                LOCAL_WAKE_LIFETIME
            } else {
                REMOTE_WAKE_LIFETIME
            };
            if state.attempts >= WAKE_ATTEMPTS
                && now.saturating_sub(since) >= lifetime.as_millis() as u64
            {
                let envelope = mailbox
                    .find(id)
                    .ok()
                    .flatten()
                    .map(|stored| stored.envelope);
                let expiry = crate::mailbox::WakeExpiry {
                    id: id.to_string(),
                    to: mailbox.address().to_string(),
                    from: envelope
                        .as_ref()
                        .map(|e| e.from.to_string())
                        .unwrap_or_default(),
                    subject: envelope.as_ref().and_then(|e| e.subject.clone()),
                    filed_at_ms: envelope.as_ref().map_or(0, |e| e.created_at_ms),
                    expired_at_ms: now,
                    attempts: state.attempts,
                    reason,
                };
                mailbox.expire_wake(id, &expiry).ok();
                return;
            }
            let wait = FIRST_RETRY
                .saturating_mul(1 << (state.attempts - 1).min(16))
                .min(LAST_RETRY);
            state.next_at_ms = now + wait.as_millis() as u64;
            state.last_error = Some(reason);
            mailbox.set_wake_state(id, &state).ok();
        }
    }
}