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