supercode_harness/

mail_route.rs

//! One route for every message: which door reaches a receiver, and delivery
//! through it.
//!
//! Every caller that delivers mail — `supercode message send`,
//! `harness.v1.sessions.message`, and idle notices — goes through
//! [`door_for`] and [`deliver`], so the choice of door is made in one place.
//! The doors, best first:
//!
//! 1. **runtime** — a session supercode controls (a hosted runtime of any
//!    harness): its own `steer`/`send_input` ([`crate::runtime_mail`]). This
//!    is the default tier.
//! 2. **native** — a Claude Code session supercode does not control: a Claude
//!    relay sends with Claude's own `SendMessage` ([`crate::claude_relay`]).
//! 3. **hook** — a Codex session supercode does not control, with
//!    supercode's hooks in its hooks file: filed in its mailbox, and the hook
//!    points the session at it.
//! 4. **stored** — no door: filed in its mailbox, seen only when the session
//!    reads it.
//!
//! Tiers 2–4 are the degradation tier, for sessions supercode does not
//! control. An operator (a board, a script) is not a session: its mail is
//! filed in its mailbox and read with `sessions.inbox`.

use std::path::Path;

use crate::claude_peer::{read_registry, registry_dir, ClaudePeerSession, ClaudePeerStatus};
use crate::claude_relay::{send_through_relay, supercode_program, RelayReceipt, RELAY_NAME_PREFIX};
use crate::live_runtime::LiveRuntimeRecord;
use crate::mailbox::{
    local_machine_name, mail_root, Envelope, IdleSubscription, MailAddress, Mailbox, ReplyVia,
};
use crate::runtime_mail::{deliver_to_runtime, RuntimeDelivery};
use crate::HarnessHomes;

/// Arguments a Codex hook entry carries, which is how its presence is found.
pub const CODEX_HOOK_ARGUMENTS: &str = "message hook codex";

/// The door that reaches one receiver.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Door {
    /// A session supercode controls.
    Runtime(Box<LiveRuntimeRecord>),
    /// A Claude Code session supercode does not control.
    Native(Box<ClaudePeerSession>),
    /// A Codex session supercode does not control, with supercode's hooks:
    /// they show it its mailbox at its next tool call or when its turn ends.
    /// An idle one in a daemon pane is woken through its pane.
    Hook {
        /// The daemon pane it runs in.
        pane: Option<String>,
        /// Whether its last turn has ended.
        idle: bool,
    },
    /// A running session with no door.
    Stored,
    /// An operator's mailbox.
    Operator,
}

impl Door {
    /// Stable name of the door, as `message list` and receipts show it.
    pub fn name(&self) -> &'static str {
        match self {
            Self::Runtime(_) => "runtime",
            Self::Native(_) => "native",
            Self::Hook { .. } => "hook",
            Self::Stored => "stored",
            Self::Operator => "operator",
        }
    }
}

/// Why no door reaches an address on this machine.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum NoDoor {
    /// The address names another machine.
    OtherMachine(String),
    /// No running session has that address.
    NotRunning,
}

/// The door that reaches `to` on this machine.
pub fn door_for(homes: &HarnessHomes, to: &MailAddress) -> Result<Door, NoDoor> {
    if to.machine != local_machine_name() {
        return Err(NoDoor::OtherMachine(to.machine.clone()));
    }
    if to.harness == "operator" || to.harness == "board" {
        return Ok(Door::Operator);
    }
    LiveSessions::read(homes)
        .sessions
        .into_iter()
        .find(|session| &session.address == to)
        .map(|session| session.door)
        .ok_or(NoDoor::NotRunning)
}

/// One running session on this machine, as the router reaches it.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LiveSession {
    /// Its address.
    pub address: MailAddress,
    /// The name an agent knows it by (`volter-2c@machine`).
    pub name: String,
    /// Its status as its harness reports it (`busy`, `idle`, `hosted`, …).
    pub status: String,
    /// The door that reaches it.
    pub door: Door,
    /// The process running it, when the harness names one (a hosted
    /// runtime's host, a Claude session's own process).
    pub pid: Option<u32>,
    /// Its working directory.
    pub cwd: Option<std::path::PathBuf>,
    /// The tmux session and pane it runs in, when Claude records one.
    pub tmux: Option<String>,
    /// Its transcript, when the harness names the file (a Codex rollout).
    pub transcript: Option<std::path::PathBuf>,
}

impl LiveSession {
    /// When its transcript last recorded a message (a prompt, a reply, a tool
    /// call or result), in epoch milliseconds: what the session last did,
    /// read from the harness's own records. A queued or injected notice is
    /// not a message, so this is not the transcript's mtime.
    pub fn last_message_at_ms(&self, homes: &HarnessHomes) -> Option<u64> {
        let harness = self.address.harness.as_str();
        let path = match &self.transcript {
            Some(path) => path.clone(),
            None if harness == "claude-code" => {
                let file = format!("{}.jsonl", self.address.session_id);
                std::fs::read_dir(&homes.claude_code)
                    .ok()?
                    .flatten()
                    .map(|project| project.path().join(&file))
                    .find(|path| path.is_file())?
            }
            None => return None,
        };
        last_message_at_ms(&path, harness)
    }
}

impl LiveSession {
    /// What a session waiting on its user is asking, read from its transcript: the newest tool call with no result
    /// yet. An `AskUserQuestion` gives its questions, headers and option labels; any other tool its name and a
    /// shortened input. `None` when the session is not waiting or its transcript shows nothing pending.
    pub fn pending_request(&self, homes: &HarnessHomes) -> Option<serde_json::Value> {
        if self.status != "waiting" {
            return None;
        }
        // A Codex prompt writes nothing to its rollout: what it asks is read off its pane.
        if self.address.harness == "codex" {
            let prompt = daemon_pane(self).and_then(|pane| pane_prompt(&pane))?;
            return Some(serde_json::json!({"prompt": prompt, "source": "screen"}));
        }
        if self.address.harness != "claude-code" {
            return None;
        }
        let file = format!("{}.jsonl", self.address.session_id);
        let from_transcript = std::fs::read_dir(&homes.claude_code)
            .ok()
            .and_then(|projects| {
                projects
                    .flatten()
                    .map(|project| project.path().join(&file))
                    .find(|path| path.is_file())
            })
            .and_then(|path| pending_request(&path));
        // Claude Code writes a question's tool call to its transcript only once it is answered
        // (2.1.280): until then, what it asks is on its pane.
        from_transcript.or_else(|| {
            let prompt = daemon_pane(self).and_then(|pane| pane_prompt(&pane))?;
            Some(serde_json::json!({"prompt": prompt, "source": "screen"}))
        })
    }
}

/// The newest tool call in a Claude transcript that has no result yet, as [`LiveSession::pending_request`] shows it.
pub fn pending_request(path: &Path) -> Option<serde_json::Value> {
    use std::io::{Read, Seek, SeekFrom};
    let mut file = std::fs::File::open(path).ok()?;
    let length = file.metadata().ok()?.len();
    let start = length.saturating_sub(512 * 1024);
    file.seek(SeekFrom::Start(start)).ok()?;
    let mut bytes = Vec::new();
    file.read_to_end(&mut bytes).ok()?;
    let text = String::from_utf8_lossy(&bytes);
    let mut answered = std::collections::HashSet::new();
    for line in text.lines().rev() {
        let Ok(record) = serde_json::from_str::<serde_json::Value>(line) else {
            continue;
        };
        if record["isSidechain"] == true {
            continue;
        }
        let Some(content) = record
            .pointer("/message/content")
            .and_then(serde_json::Value::as_array)
        else {
            continue;
        };
        match record["type"].as_str() {
            Some("user") => {
                for item in content.iter().filter(|item| item["type"] == "tool_result") {
                    if let Some(id) = item["tool_use_id"].as_str() {
                        answered.insert(id.to_string());
                    }
                }
            }
            Some("assistant") => {
                let Some(call) = content.iter().rev().find(|item| item["type"] == "tool_use")
                else {
                    continue;
                };
                if call["id"].as_str().is_some_and(|id| answered.contains(id)) {
                    return None;
                }
                let tool = call["name"].as_str().unwrap_or_default();
                if tool == "AskUserQuestion" {
                    let questions = call["input"]["questions"]
                        .as_array()
                        .map(|questions| {
                            questions
                                .iter()
                                .map(|question| {
                                    serde_json::json!({
                                        "question": question["question"],
                                        "header": question["header"],
                                        "options": question["options"].as_array().map(|options| options.iter().map(|option| option["label"].clone()).collect::<Vec<_>>()).unwrap_or_default(),
                                    })
                                })
                                .collect::<Vec<_>>()
                        })
                        .unwrap_or_default();
                    return Some(serde_json::json!({"tool": tool, "questions": questions}));
                }
                let input = call["input"].to_string();
                let input: String = input.chars().take(500).collect();
                return Some(serde_json::json!({"tool": tool, "input": input}));
            }
            _ => {}
        }
    }
    None
}

/// The newest message record's timestamp in a Claude or Codex transcript.
fn last_message_at_ms(path: &Path, harness: &str) -> Option<u64> {
    use std::io::{Read, Seek, SeekFrom};
    // A tool result can be large; widen the window once before giving up.
    for window in [256 * 1024_u64, 8 * 1024 * 1024] {
        let mut file = std::fs::File::open(path).ok()?;
        let length = file.metadata().ok()?.len();
        let start = length.saturating_sub(window);
        file.seek(SeekFrom::Start(start)).ok()?;
        let mut bytes = Vec::new();
        file.read_to_end(&mut bytes).ok()?;
        let text = String::from_utf8_lossy(&bytes);
        let mut lines = text.lines().rev().collect::<Vec<_>>();
        if start > 0 {
            lines.pop(); // the first line may begin mid-record
        }
        for line in lines {
            let Ok(record) = serde_json::from_str::<serde_json::Value>(line) else {
                continue;
            };
            let message = match harness {
                "codex" => record["type"] == "response_item",
                _ => {
                    matches!(record["type"].as_str(), Some("user" | "assistant"))
                        && record["isSidechain"] != true
                }
            };
            if !message {
                continue;
            }
            if let Some(at) = record["timestamp"]
                .as_str()
                .and_then(supercode_interchange::sidecar::rfc3339_to_ms)
                .and_then(|at| u64::try_from(at).ok())
            {
                return Some(at);
            }
        }
        if start == 0 {
            return None;
        }
    }
    None
}

/// Every running session on this machine with its door, read once: what
/// `message list` shows, what names resolve against, and what discovery
/// projects onto each discovered session as `delivery`.
#[derive(Debug, Default)]
pub struct LiveSessions {
    sessions: Vec<LiveSession>,
}

/// Why a receiver named by an agent was not found.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Unresolved {
    /// The address names a session that is not running.
    Stale(String),
    /// No running session has that name (the text suggests near names).
    Unknown(String),
}

impl LiveSessions {
    /// Read every running session now. Sessions supercode controls come
    /// first, and a degraded-tier row for the same session is dropped: the
    /// runtime is the default tier.
    pub fn read(homes: &HarnessHomes) -> Self {
        crate::slow_log::timed("read live sessions", || Self::read_now(homes))
    }

    fn read_now(homes: &HarnessHomes) -> Self {
        let machine = local_machine_name();
        let registry = read_registry(&registry_dir(homes));
        // A hosted Claude session is also in Claude's registry, which knows
        // its name; a relay is not a session and is not listed.
        let registered = |record: &crate::live_runtime::LiveRuntimeRecord| {
            registry.iter().find(|session| {
                session.session_id == record.source.session_id
                    || session.session_id == record.runtime_session_id
            })
        };
        let records = crate::runtime_mail::controlled_runtimes();
        // A runtime supercode hosts says whether a turn runs: its label is that, not only its door.
        let turns = crate::runtime_mail::runtime_turn_states(&records);
        let mut sessions: Vec<LiveSession> = records
            .into_iter()
            .zip(turns)
            .filter_map(|(record, turn)| {
                let registered = registered(&record);
                if registered.is_some_and(|session| session.name.starts_with(RELAY_NAME_PREFIX)) {
                    return None;
                }
                let address =
                    MailAddress::new(&machine, &record.source.harness, &record.source.session_id)
                        .ok()?;
                let short: String = record.source.session_id.chars().take(8).collect();
                let name = match registered {
                    Some(session) if !session.name.is_empty() => session.name.clone(),
                    _ => format!("{}-{short}", record.source.harness),
                };
                Some(LiveSession {
                    name: format!("{name}@{machine}"),
                    address,
                    // `hosted` only when its runtime did not answer: no label rather than a false one.
                    status: match turn {
                        Some(crate::frontend::FrontendTurnState::Busy) => "busy".into(),
                        Some(crate::frontend::FrontendTurnState::Idle) => "idle".into(),
                        None => "hosted".into(),
                    },
                    pid: Some(record.pid),
                    cwd: Some(record.source.workspace.clone()),
                    tmux: None,
                    transcript: None,
                    door: Door::Runtime(Box::new(record)),
                })
            })
            .collect();
        let controlled = |address: &MailAddress, sessions: &[LiveSession]| {
            sessions.iter().any(|session| &session.address == address)
        };
        for session in registry {
            if session.name.starts_with(RELAY_NAME_PREFIX) {
                continue;
            }
            let Ok(address) = MailAddress::new(&machine, "claude-code", &session.session_id) else {
                continue;
            };
            if controlled(&address, &sessions) {
                continue;
            }
            sessions.push(LiveSession {
                address,
                name: format!("{}@{machine}", session.name),
                status: session
                    .status
                    .as_ref()
                    .map(|status| status.as_str().to_string())
                    .unwrap_or_else(|| "unknown".into()),
                pid: Some(session.pid),
                cwd: session.cwd.clone(),
                tmux: session.tmux.clone(),
                transcript: None,
                door: Door::Native(Box::new(session)),
            });
        }
        let user_hook = codex_user_hook_installed();
        let rollouts = crate::codex_peer::live_rollouts(&homes.codex);
        let panes = crate::codex_peer::session_panes();
        // A shared native app-server can retain a rollout after its terminal has gone.
        // Pane absence rules out the pane, not the native writer holding the thread.
        let live_panes = (!panes.is_empty())
            .then(|| live_daemon_panes(panes.values()))
            .flatten();
        for (path, status) in rollouts {
            let Some((session_id, None)) = crate::codex_peer::rollout_session(&path) else {
                continue;
            };
            let Ok(address) = MailAddress::new(&machine, "codex", &session_id) else {
                continue;
            };
            if controlled(&address, &sessions) {
                continue;
            }
            let cwd = crate::codex_peer::rollout_cwd(&path);
            let hooked = user_hook || cwd.as_deref().is_some_and(codex_project_hook_installed);
            let pane = panes
                .get(&session_id)
                .filter(|pane| {
                    live_panes
                        .as_ref()
                        .is_none_or(|live| live.contains_key(*pane))
                })
                .cloned();
            // A Codex turn waiting on an approval or a choice writes nothing to its rollout, which
            // reads that turn as working; in a daemon pane its screen shows the prompt.
            let status = match status {
                crate::codex_peer::CodexPeerStatus::Busy
                | crate::codex_peer::CodexPeerStatus::Running
                    if pane.as_ref().is_some_and(|p| {
                        live_panes.as_ref().and_then(|live| live.get(p)) == Some(&true)
                    }) =>
                {
                    "idle".to_string()
                }

                crate::codex_peer::CodexPeerStatus::Busy
                | crate::codex_peer::CodexPeerStatus::Running
                    if pane.as_deref().and_then(pane_prompt).is_some() =>
                {
                    "waiting".to_string()
                }
                status => status.as_str().to_string(),
            };
            let door = if hooked {
                Door::Hook {
                    pane: pane.clone(),
                    idle: status == "idle",
                }
            } else {
                Door::Stored
            };
            sessions.push(LiveSession {
                name: format!("{}@{machine}", codex_name(&session_id)),
                address,
                status,
                pid: None,
                cwd,
                // The daemon pane it runs in, as a Claude session's tmux names its own.
                tmux: pane,
                transcript: Some(path),
                door,
            });
        }
        // A conversation resumed in a daemon pane and idle since holds no rollout open; its pane and the
        // process's own `resume <id>` still name it, so it is reachable as an idle session there.
        for (session_id, pane) in &panes {
            let Ok(address) = MailAddress::new(&machine, "codex", session_id) else {
                continue;
            };
            if controlled(&address, &sessions)
                || !live_panes
                    .as_ref()
                    .is_some_and(|live| live.contains_key(pane))
            {
                continue;
            }
            let Some(path) = crate::codex_peer::rollout_of_session(&homes.codex, session_id) else {
                continue;
            };
            let cwd = crate::codex_peer::rollout_cwd(&path);
            let hooked = user_hook || cwd.as_deref().is_some_and(codex_project_hook_installed);
            sessions.push(LiveSession {
                name: format!("{}@{machine}", codex_name(session_id)),
                address,
                status: "idle".to_string(),
                pid: None,
                cwd,
                tmux: Some(pane.clone()),
                transcript: Some(path),
                door: if hooked {
                    Door::Hook {
                        pane: Some(pane.clone()),
                        idle: true,
                    }
                } else {
                    Door::Stored
                },
            });
        }
        Self { sessions }
    }

    /// Every running session.
    pub fn all(&self) -> &[LiveSession] {
        &self.sessions
    }

    /// The door that reaches `harness`'s session `session_id`, when it is
    /// running.
    pub fn door(&self, harness: &str, session_id: &str) -> Option<&'static str> {
        self.sessions
            .iter()
            .find(|session| {
                session.address.harness == harness && session.address.session_id == session_id
            })
            .map(|session| session.door.name())
    }

    /// The running session an agent means by `to`: an address, `name@machine`
    /// on this machine, or a name unique here.
    pub fn resolve(&self, to: &str) -> Result<&LiveSession, Unresolved> {
        let machine = local_machine_name();
        if let Ok(address) = MailAddress::parse(to) {
            return self
                .sessions
                .iter()
                .find(|session| session.address == address)
                .ok_or_else(|| {
                    Unresolved::Stale(format!(
                        "{to} is no longer running. Nothing was sent. Run supercode message list \
                         for the live sessions."
                    ))
                });
        }
        let wanted = match to.split_once('@') {
            Some((name, at)) if at == machine => name.to_string(),
            Some((_, at)) => {
                return Err(Unresolved::Unknown(format!(
                    "Not sent: {to} is on machine {at}, not this one. Nothing was sent."
                )))
            }
            None => to.to_string(),
        };
        let short = |session: &LiveSession| {
            session
                .name
                .split('@')
                .next()
                .unwrap_or_default()
                .to_string()
        };
        let matching: Vec<&LiveSession> = self
            .sessions
            .iter()
            .filter(|session| short(session) == wanted)
            .collect();
        if let [only] = matching.as_slice() {
            return Ok(only);
        }
        let hint = if matching.len() > 1 {
            format!(
                " {} sessions are named {wanted}; use its address.",
                matching.len()
            )
        } else {
            let near: Vec<String> = self
                .sessions
                .iter()
                .filter(|session| {
                    let name = short(session);
                    name.contains(&wanted)
                        || wanted.contains(&name)
                        || name
                            .chars()
                            .zip(wanted.chars())
                            .take_while(|(a, b)| a == b)
                            .count()
                            >= 4
                })
                .take(3)
                .map(|session| {
                    format!(
                        "{} ({}, {})",
                        session.name, session.address.harness, session.status
                    )
                })
                .collect();
            if near.is_empty() {
                String::new()
            } else {
                format!(" Did you mean: {}?", near.join(", "))
            }
        };
        Err(Unresolved::Unknown(format!(
            "No session named \"{wanted}\" is reachable.{hint} Run supercode message list. Nothing \
             was sent."
        )))
    }
}

/// Whether the session process `pid` has supercode's messaging tools: a
/// harness starts each MCP server as a child of the session, so the tools are
/// loaded exactly when a live `supercode message mcp` is one of its children.
pub fn has_message_tools(pid: u32) -> bool {
    let Ok(output) = std::process::Command::new("ps")
        .args(["-A", "-o", "ppid=,command="])
        .output()
    else {
        return false;
    };
    String::from_utf8_lossy(&output.stdout).lines().any(|line| {
        let line = line.trim_start();
        let Some((ppid, command)) = line.split_once(' ') else {
            return false;
        };
        ppid.parse::<u32>() == Ok(pid) && command.trim_end().ends_with(" message mcp")
    })
}

/// Display name of a Codex conversation: `codex-` and the start of its id.
pub fn codex_name(session_id: &str) -> String {
    format!("codex-{}", session_id.chars().take(8).collect::<String>())
}

/// The session a process belongs to: the sender a message is from.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct Caller {
    /// Its address, where replies go.
    pub address: MailAddress,
    /// The name it is known by.
    pub name: String,
}

/// Why no session could be found behind a process.
pub const CALLER_UNRESOLVED: &str = "Can't tell which session is running this command, so \
    replies would have nowhere to go. Nothing was sent. Run it from your agent session's own \
    shell tool.";

/// The session behind a process, from its ancestry `pids` (nearest first):
/// the first that owns a hosted runtime, a Claude session or a Codex
/// conversation. Never declared by the caller; an environment id only
/// corroborates, and a mismatch refuses.
pub fn resolve_caller(homes: &HarnessHomes, pids: &[u32]) -> Result<Caller, String> {
    crate::slow_log::timed("resolve the caller", || resolve_caller_now(homes, pids))
}

fn resolve_caller_now(homes: &HarnessHomes, pids: &[u32]) -> Result<Caller, String> {
    let machine = local_machine_name();
    let registry = read_registry(&registry_dir(homes));
    let hosted = crate::runtime_mail::controlled_runtimes();
    for &pid in pids {
        if let Some(record) = hosted.iter().find(|record| record.pid == pid) {
            let short: String = record.source.session_id.chars().take(8).collect();
            return Ok(Caller {
                address: MailAddress::new(
                    &machine,
                    &record.source.harness,
                    &record.source.session_id,
                )
                .map_err(|error| error.to_string())?,
                name: format!("{}-{short}@{machine}", record.source.harness),
            });
        }
        if let Some(session) = registry.iter().find(|session| session.pid == pid) {
            if let Ok(claimed) = std::env::var("CLAUDE_CODE_SESSION_ID") {
                if !claimed.is_empty() && claimed != session.session_id {
                    return Err(format!(
                        "{CALLER_UNRESOLVED} (CLAUDE_CODE_SESSION_ID names {claimed}, but the Claude \
                         process {pid} above this command is session {})",
                        session.session_id
                    ));
                }
            }
            return Ok(Caller {
                address: MailAddress::new(&machine, "claude-code", &session.session_id)
                    .map_err(|error| error.to_string())?,
                name: format!("{}@{machine}", session.name),
            });
        }
        // Codex names the thread a command runs for (`CODEX_THREAD_ID`); the
        // claim stands when this process holds that thread's rollout, as
        // Codex's shared app-server daemon does for every thread it runs.
        if let Some(thread) = std::env::var("CODEX_THREAD_ID")
            .ok()
            .filter(|thread| !thread.is_empty())
            .filter(|thread| crate::codex_peer::holds_session(pid, thread))
        {
            return Ok(Caller {
                address: MailAddress::new(&machine, "codex", &thread)
                    .map_err(|error| error.to_string())?,
                name: format!("{}@{machine}", codex_name(&thread)),
            });
        }
        if let Some((session_id, _)) = crate::codex_peer::session_of_process(pid) {
            return Ok(Caller {
                address: MailAddress::new(&machine, "codex", &session_id)
                    .map_err(|error| error.to_string())?,
                name: format!("{}@{machine}", codex_name(&session_id)),
            });
        }
    }
    #[cfg(windows)]
    if let Some(session) = msys_cut_claim(&registry, pids) {
        return Ok(Caller {
            address: MailAddress::new(&machine, "claude-code", &session.session_id)
                .map_err(|error| error.to_string())?,
            name: format!("{}@{machine}", session.name),
        });
    }
    Err(format!(
        "{CALLER_UNRESOLVED} (looked for this command's processes {pids:?} among {} Claude \
         sessions in {} and {} hosted runtimes)",
        registry.len(),
        registry_dir(homes).display(),
        hosted.len()
    ))
}

/// Git Bash (MSYS) runs a command in a forked process that hands over to it. When the command is
/// itself an MSYS program (a `sh` script, as npm's command shims are), the forked process exits once
/// it has handed over, so the Windows parent chain is cut just above that shell and never reaches
/// the Claude session running it. When the ancestry ends at exactly such a cut (its last pid is gone,
/// the one before it is an MSYS shell), the session Claude names in `CLAUDE_CODE_SESSION_ID` stands
/// if it is a live Claude session on this machine.
#[cfg(windows)]
fn msys_cut_claim<'a>(
    registry: &'a [ClaudePeerSession],
    pids: &[u32],
) -> Option<&'a ClaudePeerSession> {
    let table = process_table();
    let [.., shell, cut] = pids else {
        return None;
    };
    if table.contains_key(cut) {
        return None;
    }
    let name = table.get(shell)?.1.to_ascii_lowercase();
    if !matches!(name.as_str(), "sh.exe" | "bash.exe" | "dash.exe") {
        return None;
    }
    let claimed = std::env::var("CLAUDE_CODE_SESSION_ID").ok()?;
    registry
        .iter()
        .find(|session| !claimed.is_empty() && session.session_id == claimed)
}

/// This process's ancestry, nearest first (itself included).
pub fn process_ancestry() -> Vec<u32> {
    crate::slow_log::timed("read the process ancestry", || {
        ancestry_of(std::process::id())
    })
}

/// A process's ancestry, nearest first (itself included).
pub fn ancestry_of(pid: u32) -> Vec<u32> {
    let parents = parent_pids();
    let mut chain = vec![pid];
    let mut current = pid;
    while let Some(&parent) = parents.get(&current) {
        if parent <= 1 || chain.contains(&parent) {
            break;
        }
        chain.push(parent);
        current = parent;
    }
    chain
}

/// Every process's parent: pid → parent pid, from `ps`.
#[cfg(not(windows))]
fn parent_pids() -> std::collections::HashMap<u32, u32> {
    let Ok(output) = std::process::Command::new("ps")
        .args(["-axo", "pid=,ppid="])
        .output()
    else {
        return Default::default();
    };
    String::from_utf8_lossy(&output.stdout)
        .lines()
        .filter_map(|line| {
            let mut fields = line.split_whitespace();
            Some((fields.next()?.parse().ok()?, fields.next()?.parse().ok()?))
        })
        .collect()
}

/// Every process's parent: pid → parent pid, from a process snapshot.
#[cfg(windows)]
fn parent_pids() -> std::collections::HashMap<u32, u32> {
    process_table()
        .into_iter()
        .map(|(pid, (parent, _))| (pid, parent))
        .collect()
}

/// Every process: pid → (parent pid, executable file name), from a process snapshot.
#[cfg(windows)]
fn process_table() -> std::collections::HashMap<u32, (u32, String)> {
    use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE};
    use windows_sys::Win32::System::Diagnostics::ToolHelp::{
        CreateToolhelp32Snapshot, Process32FirstW, Process32NextW, PROCESSENTRY32W,
        TH32CS_SNAPPROCESS,
    };

    let mut parents = std::collections::HashMap::new();
    let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0) };
    if snapshot == INVALID_HANDLE_VALUE {
        return parents;
    }
    let mut entry: PROCESSENTRY32W = unsafe { std::mem::zeroed() };
    entry.dwSize = std::mem::size_of::<PROCESSENTRY32W>() as u32;
    let mut has_entry = unsafe { Process32FirstW(snapshot, &mut entry) } != 0;
    while has_entry {
        let length = entry
            .szExeFile
            .iter()
            .position(|&unit| unit == 0)
            .unwrap_or(entry.szExeFile.len());
        let name = String::from_utf16_lossy(&entry.szExeFile[..length]);
        parents.insert(entry.th32ProcessID, (entry.th32ParentProcessID, name));
        has_entry = unsafe { Process32NextW(snapshot, &mut entry) } != 0;
    }
    unsafe {
        CloseHandle(snapshot);
    }
    parents
}

/// Codex's user-level hooks file.
pub fn codex_hooks_path() -> std::path::PathBuf {
    std::env::var_os("CODEX_HOME")
        .map(std::path::PathBuf::from)
        .or_else(|| {
            supercode_interchange::user_home()
                .map(std::path::PathBuf::into_os_string)
                .map(|home| std::path::PathBuf::from(home).join(".codex"))
        })
        .unwrap_or_else(|| std::path::PathBuf::from(".codex"))
        .join("hooks.json")
}

/// Whether supercode's mail hook is in Codex's user hooks file. (Codex runs
/// it only once its user has trusted it.)
pub fn codex_user_hook_installed() -> bool {
    std::fs::read_to_string(codex_hooks_path())
        .is_ok_and(|text| text.contains(CODEX_HOOK_ARGUMENTS))
}

/// Whether supercode's mail hook is in a project's Codex hooks file, in the
/// session's directory or one above it.
pub fn codex_project_hook_installed(cwd: &Path) -> bool {
    cwd.ancestors().any(|directory| {
        std::fs::read_to_string(directory.join(".codex").join("hooks.json"))
            .is_ok_and(|text| text.contains(CODEX_HOOK_ARGUMENTS))
    })
}

/// How a delivery ended.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Delivered {
    /// The receiver has it now: steered into a running turn.
    Steered,
    /// The receiver has it now: it started a turn in an idle session.
    Started,
    /// Claude reported it in the receiver's inbox; `busy` says whether the
    /// receiver reads it at its next tool call (true) or it starts a turn.
    Native {
        /// True when the receiver was in a turn.
        busy: bool,
    },
    /// Filed in the receiver's mailbox, which a hook points it at.
    Hooked,
    /// Filed in an idle receiver's mailbox, and its pane told to read it: a
    /// turn has started, in which its hook shows it the message.
    HookWoken,
    /// Filed in the receiver's mailbox, waiting to be read.
    Queued,
    /// Filed with no door to show it.
    Stored,
    /// Filed in an operator's mailbox.
    Operator,
}

/// A delivery refused before anything was sent.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Refused {
    /// `--queue` cannot hold a message for an idle Claude session: Claude
    /// starts a turn for every message it delivers.
    CannotQueueNative,
    /// The relayed text would not fit one relay turn.
    TooLong(usize),
}

/// Largest message relayed into a Claude session. The relay copies it into a
/// model turn byte for byte, so it must fit comfortably in one.
pub const MAX_RELAYED_BYTES: usize = 100_000;

/// Deliver `envelope` to `to` through `door`. With `wake` false an idle
/// receiver is not started. With `notify_when_idle` the sender gets one idle
/// notice after the receiver's next turn ends.
pub async fn deliver(
    envelope: &Envelope,
    to: &MailAddress,
    door: &Door,
    wake: bool,
    notify_when_idle: bool,
) -> Result<Result<Delivered, Refused>, String> {
    let mailbox = Mailbox::open(&mail_root(), to).map_err(|error| error.to_string())?;
    // A runtime answers with its turn's final message, which the watcher
    // sends back: not every runtime can run a command (a hosted agent may
    // have no shell), and each can end a turn.
    let mut final_reply = false;
    let delivered = match door {
        Door::Runtime(record) => {
            let mut sent = envelope.clone();
            let answers = matches!(envelope.reply_via, ReplyVia::Command);
            if answers {
                sent.reply_via = ReplyVia::FinalMessage {
                    destination: envelope.from_name.clone(),
                };
            }
            match deliver_to_runtime(record, sent.render(), wake).await? {
                RuntimeDelivery::Steered => {
                    mailbox
                        .deliver_read(&sent)
                        .map_err(|error| error.to_string())?;
                    final_reply = answers;
                    Delivered::Steered
                }
                RuntimeDelivery::Started => {
                    mailbox
                        .deliver_read(&sent)
                        .map_err(|error| error.to_string())?;
                    final_reply = answers;
                    Delivered::Started
                }
                RuntimeDelivery::NotWoken => {
                    mailbox
                        .deliver(envelope)
                        .map_err(|error| error.to_string())?;
                    Delivered::Queued
                }
            }
        }
        Door::Native(session) => {
            let busy = session.status != Some(ClaudePeerStatus::Idle);
            if !wake && !busy {
                return Ok(Err(Refused::CannotQueueNative));
            }
            // The same envelope every door delivers; Claude wraps it in its
            // own, which names the relay. Its reply line names the door this
            // session really has: its tools when it runs supercode's MCP server.
            let mut envelope = envelope.clone();
            if envelope.reply_via == ReplyVia::Command && has_message_tools(session.pid) {
                envelope.reply_via = ReplyVia::Tool;
            }
            let envelope = &envelope;
            let text = envelope.render();
            if text.len() > MAX_RELAYED_BYTES {
                return Ok(Err(Refused::TooLong(text.len())));
            }
            // Filed before it is handed over: a session that reads its mailbox on its first tool call (to reply by
            // the message's id) finds it there. A relay that fails takes back the copy this delivery filed.
            let filed_here = mailbox
                .find(&envelope.id)
                .map_err(|error| error.to_string())?
                .is_none();
            let filed = mailbox
                .deliver_read(envelope)
                .map_err(|error| error.to_string())?;
            match send_through_relay(
                &envelope.from,
                &envelope.from_name,
                &session.name,
                text,
                &envelope.id,
            )
            .await
            {
                RelayReceipt::Delivered { .. } => Delivered::Native { busy },
                RelayReceipt::Failed { detail } => {
                    if filed_here {
                        std::fs::remove_file(&filed).ok();
                    }
                    return Err(detail);
                }
            }
        }
        Door::Hook { .. } => {
            mailbox
                .deliver(envelope)
                .map_err(|error| error.to_string())?;
            if wake {
                mailbox
                    .request_wake(&envelope.id)
                    .map_err(|error| error.to_string())?;
            }
            Delivered::Hooked
        }
        Door::Stored => {
            mailbox
                .deliver(envelope)
                .map_err(|error| error.to_string())?;
            Delivered::Stored
        }
        Door::Operator => {
            mailbox
                .deliver(envelope)
                .map_err(|error| error.to_string())?;
            Delivered::Operator
        }
    };
    let notice = notify_when_idle && !matches!(door, Door::Operator);
    if notice || final_reply {
        let mut subscription = IdleSubscription::new(envelope.id.clone(), envelope.from.clone());
        subscription.notice = notice;
        subscription.final_reply = final_reply;
        mailbox
            .subscribe_idle(&subscription)
            .map_err(|error| error.to_string())?;
        // The watcher that settles it runs beside the machine daemon.
        if let Ok(program) = supercode_program() {
            crate::claude_relay::ensure_machine_daemon(&program)
                .await
                .ok();
        }
    }
    Ok(Ok(delivered))
}

/// How the user's own turn reached its session.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UserTurn {
    /// A hosted runtime was in a turn; the words were steered into it.
    Steered,
    /// A hosted runtime was idle; the words started a turn.
    Started,
    /// Typed into the session's pane as its own submitted turn.
    Typed,
    /// The pane's composer holds a draft, or its program is not at its
    /// composer: the turn waits in the session's mailbox and is typed once
    /// the composer is empty.
    Waiting,
}

impl UserTurn {
    /// Stable wire spelling.
    pub const fn as_str(self) -> &'static str {
        match self {
            Self::Steered => "steered",
            Self::Started => "started",
            Self::Typed => "typed",
            Self::Waiting => "waiting",
        }
    }
}

/// The choice or permission prompt a daemon pane's screen shows, as its lines (the question above
/// it and its options), or `None`. The terminal substrate's own check (`classifyAttentionSignal`):
/// a selector (❯ or ›) on a numbered option is a prompt blocked on an answer; the idle input line
/// has no number after its selector.
pub fn pane_prompt(pane: &str) -> Option<Vec<String>> {
    let screen = pane_screen(pane)?;
    let lines: Vec<&str> = screen.lines().map(str::trim_end).collect();
    let is_option = |line: &str| {
        let line = line.trim_start();
        // Windows consoles draw Claude's selector as `>`.
        let rest = line
            .strip_prefix('❯')
            .or_else(|| line.strip_prefix('›'))
            .or_else(|| line.strip_prefix('>'))
            .unwrap_or(line)
            .trim_start();
        let digits = rest.chars().take_while(char::is_ascii_digit).count();
        digits > 0 && matches!(rest[digits..].chars().next(), Some('.' | ')'))
    };
    let selected = lines.iter().position(|line| {
        let line = line.trim_start();
        (line.starts_with('❯') || line.starts_with('›') || line.starts_with('>')) && is_option(line)
    })?;
    // The prompt: the options' block of non-empty lines and the block above it (the question).
    let block_start = |end: usize| {
        lines[..end]
            .iter()
            .rposition(|line| line.trim().is_empty())
            .map_or(0, |blank| blank + 1)
    };
    let options = block_start(selected);
    let above = lines[..options]
        .iter()
        .rposition(|line| !line.trim().is_empty())
        .map(|last| block_start(last));
    let start = above.unwrap_or(options);
    let end = lines[selected..]
        .iter()
        .position(|line| line.trim().is_empty())
        .map_or(lines.len(), |blank| selected + blank);
    Some(
        lines[start..end]
            .iter()
            .map(|line| line.trim().to_string())
            .filter(|line| !line.is_empty())
            .take(20)
            .collect(),
    )
}

/// Read this machine's actual panes once for the session list. Unavailability is unknown.
/// Only `wanted` are read: the daemon captures each pane it is asked about, so asking about every
/// pane cost a screen capture of all of them on every live-session read (t_23e47b73).
fn live_daemon_panes<'a>(
    wanted: impl Iterator<Item = &'a String>,
) -> Option<std::collections::HashMap<String, bool>> {
    let wanted: Vec<&str> = wanted.map(String::as_str).collect();
    crate::slow_log::timed("teams panes ls --presence-only", || {
        live_daemon_panes_now(&wanted.join(","))
    })
}

fn live_daemon_panes_now(wanted: &str) -> Option<std::collections::HashMap<String, bool>> {
    let entry = crate::teams_entry().ok()?;
    let node = std::env::var(crate::orchestrator_door::NODE_BIN_ENV)
        .ok()
        .filter(|value| !value.trim().is_empty())
        .unwrap_or_else(|| "node".into());
    let output = std::process::Command::new(node)
        .arg(entry)
        .args(["panes", "ls", "--presence-only", "--panes", wanted])
        .stdin(std::process::Stdio::null())
        .output()
        .ok()?;
    if !output.status.success() {
        return None;
    }
    let rows: Vec<serde_json::Value> = serde_json::from_slice(&output.stdout).ok()?;
    rows.iter()
        .map(|row| {
            Some((
                row["pane"].as_str()?.to_owned(),
                row["idleComposer"].as_bool()?,
            ))
        })
        .collect()
}

/// A daemon pane's screen: tmux's capture where the daemon's host is tmux, else the machine daemon's
/// own capture (`teams panes capture`), as on Windows, where panes are ConPTY.
fn pane_screen(pane: &str) -> Option<String> {
    #[cfg(unix)]
    {
        let output = std::process::Command::new("tmux")
            .args(["capture-pane", "-p", "-t", &format!("{pane}:")])
            .output()
            .ok()?;
        output
            .status
            .success()
            .then(|| String::from_utf8_lossy(&output.stdout).into_owned())
    }
    #[cfg(not(unix))]
    {
        let entry = crate::teams_entry().ok()?;
        let node = std::env::var(crate::orchestrator_door::NODE_BIN_ENV)
            .ok()
            .filter(|value| !value.trim().is_empty())
            .unwrap_or_else(|| "node".into());
        let output = std::process::Command::new(node)
            .arg(entry)
            .args(["panes", "capture", pane, "--lines", "60"])
            .stdin(std::process::Stdio::null())
            .output()
            .ok()?;
        output
            .status
            .success()
            .then(|| String::from_utf8_lossy(&output.stdout).into_owned())
    }
}

/// The daemon pane a session runs in, when it runs in one.
pub fn daemon_pane(session: &LiveSession) -> Option<String> {
    let name = session.tmux.as_deref()?.split(':').next()?;
    let rest = name.strip_prefix("p_")?;
    (!rest.is_empty()
        && rest
            .chars()
            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-'))
    .then(|| name.to_string())
}

/// Deliver the user's own turn to `to` through the one door that carries the
/// user's authority: a hosted runtime's own input, or the session's pane.
/// A session with neither (running outside supercode) is refused, never
/// reached as a peer instead. Callers hold the owner's authority already:
/// this is reached only through owner doors (the machine's own harness.v1).
pub async fn deliver_user_turn(
    homes: &HarnessHomes,
    envelope: &Envelope,
    to: &MailAddress,
) -> Result<UserTurn, String> {
    let session = LiveSessions::read(homes)
        .sessions
        .into_iter()
        .find(|session| &session.address == to)
        .ok_or_else(|| format!("{to} is not running; nothing was sent"))?;
    let mailbox = Mailbox::open(&mail_root(), to).map_err(|error| error.to_string())?;
    if let Door::Runtime(record) = &session.door {
        let delivered = deliver_to_runtime(record, envelope.body.clone(), true).await?;
        mailbox
            .deliver_read(envelope)
            .map_err(|error| error.to_string())?;
        return Ok(match delivered {
            RuntimeDelivery::Steered => UserTurn::Steered,
            _ => UserTurn::Started,
        });
    }
    let Some(pane) = daemon_pane(&session) else {
        let name = session.name.split('@').next().unwrap_or(&session.name);
        return Err(format!(
            "{name} runs outside supercode, where nothing can speak as its user. Open it in a \
             pane with `supercode open {name}`; nothing was sent."
        ));
    };
    mailbox
        .deliver(envelope)
        .map_err(|error| error.to_string())?;
    let typed = type_user_turns(&mailbox, &pane).await;
    Ok(if typed.contains(&envelope.id) {
        UserTurn::Typed
    } else {
        UserTurn::Waiting
    })
}

/// Type the user's waiting turns into `pane`, oldest first, each only when
/// the pane's composer is empty. Stops at the first that has to wait.
/// Returns the ids typed.
pub async fn type_user_turns(mailbox: &Mailbox, pane: &str) -> Vec<String> {
    let mut typed = Vec::new();
    for waiting in mailbox.user_turns().unwrap_or_default() {
        // Claimed before it is typed: the machine's watcher and a direct delivery both type waiting turns, and
        // two typers filling one composer sent a turn's text twice in one prompt. A turn another typer holds is
        // that typer's, and so is every turn after it.
        let Ok(Some(stored)) = mailbox.claim_user_turn(&waiting) else {
            break;
        };
        match submit_mail_batch(pane, &stored.envelope.body, &[stored.envelope.id.clone()]).await {
            Ok(true) => {
                mailbox.acknowledge(&stored).ok();
                typed.push(stored.envelope.id.clone());
            }
            Ok(false) => {
                mailbox.release(&stored).ok();
                break;
            }
            Err(error) => {
                mailbox.release(&stored).ok();
                eprintln!(
                    "supercode: the user's turn {} for {} waits: {error}",
                    stored.envelope.id,
                    mailbox.address()
                );
                break;
            }
        }
    }
    typed
}

/// The daemon drains a recipient's waiting envelopes as one paste. The machine
/// retains the batch and its ids until the native transcript confirms receipt.
pub(crate) async fn wake_hook_mailbox(
    mailbox: &Mailbox,
    pane: &str,
    ids: &[String],
) -> Result<bool, String> {
    let messages = ids
        .iter()
        .map(|id| mailbox.find(id))
        .collect::<Result<Vec<_>, _>>()
        .map_err(|error| error.to_string())?;
    let messages: Vec<_> = messages.into_iter().flatten().collect();
    if messages.is_empty() {
        return Ok(false);
    }
    let mut delivered = true;
    for message in messages {
        delivered &= submit_mail_batch(
            pane,
            &message.envelope.render(),
            &[message.envelope.id.clone()],
        )
        .await?;
    }
    Ok(delivered)
}

async fn submit_mail_batch(pane: &str, text: &str, ids: &[String]) -> Result<bool, String> {
    let entry = crate::teams_entry().map_err(|error| error.to_string())?;
    let node = std::env::var(crate::orchestrator_door::NODE_BIN_ENV)
        .ok()
        .filter(|value| !value.trim().is_empty())
        .unwrap_or_else(|| "node".into());
    let output = tokio::process::Command::new(node)
        .arg(entry)
        .args([
            "input",
            pane,
            text,
            "--when-composer-empty",
            "--message-ids",
            &ids.join(","),
        ])
        // A message is the session's own door, not a session acting on a pane: the pane doors read that from this
        // process's place under the daemon's `message watch`, not from anything named here.
        .env_remove("SUPERCODE_CALLER")
        .stdin(std::process::Stdio::null())
        .output()
        .await
        .map_err(|error| error.to_string())?;
    if !output.status.success() {
        return Err(crate::mailbox::error_line(&String::from_utf8_lossy(
            &output.stderr,
        )));
    }
    let answer: serde_json::Value =
        serde_json::from_slice(&output.stdout).map_err(|error| error.to_string())?;
    Ok(answer["delivered"] == true)
}