Skip to main content

supercode_harness/
mail_route.rs

1//! One route for every message: which door reaches a receiver, and delivery
2//! through it.
3//!
4//! Every caller that delivers mail — `supercode message send`,
5//! `harness.v1.sessions.message`, and idle notices — goes through
6//! [`door_for`] and [`deliver`], so the choice of door is made in one place.
7//! The doors, best first:
8//!
9//! 1. **runtime** — a session supercode controls (a hosted runtime of any
10//!    harness): its own `steer`/`send_input` ([`crate::runtime_mail`]). This
11//!    is the default tier.
12//! 2. **native** — a Claude Code session supercode does not control: a Claude
13//!    relay sends with Claude's own `SendMessage` ([`crate::claude_relay`]).
14//! 3. **hook** — a Codex session supercode does not control, with
15//!    supercode's hooks in its hooks file: filed in its mailbox, and the hook
16//!    points the session at it.
17//! 4. **stored** — no door: filed in its mailbox, seen only when the session
18//!    reads it.
19//!
20//! Tiers 2–4 are the degradation tier, for sessions supercode does not
21//! control. An operator (a board, a script) is not a session: its mail is
22//! filed in its mailbox and read with `sessions.inbox`.
23
24use std::path::Path;
25
26use crate::claude_peer::{read_registry, registry_dir, ClaudePeerSession, ClaudePeerStatus};
27use crate::claude_relay::{send_through_relay, supercode_program, RelayReceipt, RELAY_NAME_PREFIX};
28use crate::live_runtime::LiveRuntimeRecord;
29use crate::mailbox::{
30    local_machine_name, mail_root, Envelope, IdleSubscription, MailAddress, Mailbox, ReplyVia,
31};
32use crate::runtime_mail::{deliver_to_runtime, RuntimeDelivery};
33use crate::HarnessHomes;
34
35/// Arguments a Codex hook entry carries, which is how its presence is found.
36pub const CODEX_HOOK_ARGUMENTS: &str = "message hook codex";
37
38/// The door that reaches one receiver.
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub enum Door {
41    /// A session supercode controls.
42    Runtime(Box<LiveRuntimeRecord>),
43    /// A Claude Code session supercode does not control.
44    Native(Box<ClaudePeerSession>),
45    /// A Codex session supercode does not control, with supercode's hooks.
46    Hook,
47    /// A running session with no door.
48    Stored,
49    /// An operator's mailbox.
50    Operator,
51}
52
53impl Door {
54    /// Stable name of the door, as `message list` and receipts show it.
55    pub fn name(&self) -> &'static str {
56        match self {
57            Self::Runtime(_) => "runtime",
58            Self::Native(_) => "native",
59            Self::Hook => "hook",
60            Self::Stored => "stored",
61            Self::Operator => "operator",
62        }
63    }
64}
65
66/// Why no door reaches an address on this machine.
67#[derive(Debug, Clone, PartialEq, Eq)]
68pub enum NoDoor {
69    /// The address names another machine.
70    OtherMachine(String),
71    /// No running session has that address.
72    NotRunning,
73}
74
75/// The door that reaches `to` on this machine.
76pub fn door_for(homes: &HarnessHomes, to: &MailAddress) -> Result<Door, NoDoor> {
77    if to.machine != local_machine_name() {
78        return Err(NoDoor::OtherMachine(to.machine.clone()));
79    }
80    if to.harness == "operator" {
81        return Ok(Door::Operator);
82    }
83    LiveSessions::read(homes)
84        .sessions
85        .into_iter()
86        .find(|session| &session.address == to)
87        .map(|session| session.door)
88        .ok_or(NoDoor::NotRunning)
89}
90
91/// One running session on this machine, as the router reaches it.
92#[derive(Debug, Clone, PartialEq, Eq)]
93pub struct LiveSession {
94    /// Its address.
95    pub address: MailAddress,
96    /// The name an agent knows it by (`volter-2c@machine`).
97    pub name: String,
98    /// Its status as its harness reports it (`busy`, `idle`, `hosted`, …).
99    pub status: String,
100    /// The door that reaches it.
101    pub door: Door,
102    /// The process running it, when the harness names one (a hosted
103    /// runtime's host, a Claude session's own process).
104    pub pid: Option<u32>,
105    /// Its working directory.
106    pub cwd: Option<std::path::PathBuf>,
107    /// The tmux session and pane it runs in, when Claude records one.
108    pub tmux: Option<String>,
109    /// Its transcript, when the harness names the file (a Codex rollout).
110    pub transcript: Option<std::path::PathBuf>,
111}
112
113impl LiveSession {
114    /// When its transcript last recorded a message (a prompt, a reply, a tool
115    /// call or result), in epoch milliseconds: what the session last did,
116    /// read from the harness's own records. A queued or injected notice is
117    /// not a message, so this is not the transcript's mtime.
118    pub fn last_message_at_ms(&self, homes: &HarnessHomes) -> Option<u64> {
119        let harness = self.address.harness.as_str();
120        let path = match &self.transcript {
121            Some(path) => path.clone(),
122            None if harness == "claude-code" => {
123                let file = format!("{}.jsonl", self.address.session_id);
124                std::fs::read_dir(&homes.claude_code)
125                    .ok()?
126                    .flatten()
127                    .map(|project| project.path().join(&file))
128                    .find(|path| path.is_file())?
129            }
130            None => return None,
131        };
132        last_message_at_ms(&path, harness)
133    }
134}
135
136/// The newest message record's timestamp in a Claude or Codex transcript.
137fn last_message_at_ms(path: &Path, harness: &str) -> Option<u64> {
138    use std::io::{Read, Seek, SeekFrom};
139    // A tool result can be large; widen the window once before giving up.
140    for window in [256 * 1024_u64, 8 * 1024 * 1024] {
141        let mut file = std::fs::File::open(path).ok()?;
142        let length = file.metadata().ok()?.len();
143        let start = length.saturating_sub(window);
144        file.seek(SeekFrom::Start(start)).ok()?;
145        let mut bytes = Vec::new();
146        file.read_to_end(&mut bytes).ok()?;
147        let text = String::from_utf8_lossy(&bytes);
148        let mut lines = text.lines().rev().collect::<Vec<_>>();
149        if start > 0 {
150            lines.pop(); // the first line may begin mid-record
151        }
152        for line in lines {
153            let Ok(record) = serde_json::from_str::<serde_json::Value>(line) else {
154                continue;
155            };
156            let message = match harness {
157                "codex" => record["type"] == "response_item",
158                _ => {
159                    matches!(record["type"].as_str(), Some("user" | "assistant"))
160                        && record["isSidechain"] != true
161                }
162            };
163            if !message {
164                continue;
165            }
166            if let Some(at) = record["timestamp"]
167                .as_str()
168                .and_then(supercode_interchange::sidecar::rfc3339_to_ms)
169                .and_then(|at| u64::try_from(at).ok())
170            {
171                return Some(at);
172            }
173        }
174        if start == 0 {
175            return None;
176        }
177    }
178    None
179}
180
181/// Every running session on this machine with its door, read once: what
182/// `message list` shows, what names resolve against, and what discovery
183/// projects onto each discovered session as `delivery`.
184#[derive(Debug, Default)]
185pub struct LiveSessions {
186    sessions: Vec<LiveSession>,
187}
188
189/// Why a receiver named by an agent was not found.
190#[derive(Debug, Clone, PartialEq, Eq)]
191pub enum Unresolved {
192    /// The address names a session that is not running.
193    Stale(String),
194    /// No running session has that name (the text suggests near names).
195    Unknown(String),
196}
197
198impl LiveSessions {
199    /// Read every running session now. Sessions supercode controls come
200    /// first, and a degraded-tier row for the same session is dropped: the
201    /// runtime is the default tier.
202    pub fn read(homes: &HarnessHomes) -> Self {
203        let machine = local_machine_name();
204        let registry = read_registry(&registry_dir(homes));
205        // A hosted Claude session is also in Claude's registry, which knows
206        // its name; a relay is not a session and is not listed.
207        let registered = |record: &crate::live_runtime::LiveRuntimeRecord| {
208            registry.iter().find(|session| {
209                session.session_id == record.source.session_id
210                    || session.session_id == record.runtime_session_id
211            })
212        };
213        let mut sessions: Vec<LiveSession> = crate::runtime_mail::controlled_runtimes()
214            .into_iter()
215            .filter_map(|record| {
216                let registered = registered(&record);
217                if registered.is_some_and(|session| session.name.starts_with(RELAY_NAME_PREFIX)) {
218                    return None;
219                }
220                let address =
221                    MailAddress::new(&machine, &record.source.harness, &record.source.session_id)
222                        .ok()?;
223                let short: String = record.source.session_id.chars().take(8).collect();
224                let name = match registered {
225                    Some(session) if !session.name.is_empty() => session.name.clone(),
226                    _ => format!("{}-{short}", record.source.harness),
227                };
228                Some(LiveSession {
229                    name: format!("{name}@{machine}"),
230                    address,
231                    status: "hosted".into(),
232                    pid: Some(record.pid),
233                    cwd: Some(record.source.workspace.clone()),
234                    tmux: None,
235                    transcript: None,
236                    door: Door::Runtime(Box::new(record)),
237                })
238            })
239            .collect();
240        let controlled = |address: &MailAddress, sessions: &[LiveSession]| {
241            sessions.iter().any(|session| &session.address == address)
242        };
243        for session in registry {
244            if session.name.starts_with(RELAY_NAME_PREFIX) {
245                continue;
246            }
247            let Ok(address) = MailAddress::new(&machine, "claude-code", &session.session_id) else {
248                continue;
249            };
250            if controlled(&address, &sessions) {
251                continue;
252            }
253            sessions.push(LiveSession {
254                address,
255                name: format!("{}@{machine}", session.name),
256                status: session
257                    .status
258                    .as_ref()
259                    .map(|status| status.as_str().to_string())
260                    .unwrap_or_else(|| "unknown".into()),
261                pid: Some(session.pid),
262                cwd: session.cwd.clone(),
263                tmux: session.tmux.clone(),
264                transcript: None,
265                door: Door::Native(Box::new(session)),
266            });
267        }
268        let user_hook = codex_user_hook_installed();
269        for (path, status) in crate::codex_peer::live_rollouts(&homes.codex) {
270            let Some((session_id, None)) = crate::codex_peer::rollout_session(&path) else {
271                continue;
272            };
273            let Ok(address) = MailAddress::new(&machine, "codex", &session_id) else {
274                continue;
275            };
276            if controlled(&address, &sessions) {
277                continue;
278            }
279            let cwd = crate::codex_peer::rollout_cwd(&path);
280            let hooked = user_hook || cwd.as_deref().is_some_and(codex_project_hook_installed);
281            sessions.push(LiveSession {
282                name: format!("{}@{machine}", codex_name(&session_id)),
283                address,
284                status: status.as_str().to_string(),
285                pid: None,
286                cwd,
287                tmux: None,
288                transcript: Some(path),
289                door: if hooked { Door::Hook } else { Door::Stored },
290            });
291        }
292        Self { sessions }
293    }
294
295    /// Every running session.
296    pub fn all(&self) -> &[LiveSession] {
297        &self.sessions
298    }
299
300    /// The door that reaches `harness`'s session `session_id`, when it is
301    /// running.
302    pub fn door(&self, harness: &str, session_id: &str) -> Option<&'static str> {
303        self.sessions
304            .iter()
305            .find(|session| {
306                session.address.harness == harness && session.address.session_id == session_id
307            })
308            .map(|session| session.door.name())
309    }
310
311    /// The running session an agent means by `to`: an address, `name@machine`
312    /// on this machine, or a name unique here.
313    pub fn resolve(&self, to: &str) -> Result<&LiveSession, Unresolved> {
314        let machine = local_machine_name();
315        if let Ok(address) = MailAddress::parse(to) {
316            return self
317                .sessions
318                .iter()
319                .find(|session| session.address == address)
320                .ok_or_else(|| {
321                    Unresolved::Stale(format!(
322                        "{to} is no longer running. Nothing was sent. Run supercode message list \
323                         for the live sessions."
324                    ))
325                });
326        }
327        let wanted = match to.split_once('@') {
328            Some((name, at)) if at == machine => name.to_string(),
329            Some((_, at)) => {
330                return Err(Unresolved::Unknown(format!(
331                    "Not sent: {to} is on machine {at}, not this one. Nothing was sent."
332                )))
333            }
334            None => to.to_string(),
335        };
336        let short = |session: &LiveSession| {
337            session
338                .name
339                .split('@')
340                .next()
341                .unwrap_or_default()
342                .to_string()
343        };
344        let matching: Vec<&LiveSession> = self
345            .sessions
346            .iter()
347            .filter(|session| short(session) == wanted)
348            .collect();
349        if let [only] = matching.as_slice() {
350            return Ok(only);
351        }
352        let hint = if matching.len() > 1 {
353            format!(
354                " {} sessions are named {wanted}; use its address.",
355                matching.len()
356            )
357        } else {
358            let near: Vec<String> = self
359                .sessions
360                .iter()
361                .filter(|session| {
362                    let name = short(session);
363                    name.contains(&wanted)
364                        || wanted.contains(&name)
365                        || name
366                            .chars()
367                            .zip(wanted.chars())
368                            .take_while(|(a, b)| a == b)
369                            .count()
370                            >= 4
371                })
372                .take(3)
373                .map(|session| {
374                    format!(
375                        "{} ({}, {})",
376                        session.name, session.address.harness, session.status
377                    )
378                })
379                .collect();
380            if near.is_empty() {
381                String::new()
382            } else {
383                format!(" Did you mean: {}?", near.join(", "))
384            }
385        };
386        Err(Unresolved::Unknown(format!(
387            "No session named \"{wanted}\" is reachable.{hint} Run supercode message list. Nothing \
388             was sent."
389        )))
390    }
391}
392
393/// Whether the session process `pid` has supercode's messaging tools: a
394/// harness starts each MCP server as a child of the session, so the tools are
395/// loaded exactly when a live `supercode message mcp` is one of its children.
396pub fn has_message_tools(pid: u32) -> bool {
397    let Ok(output) = std::process::Command::new("ps")
398        .args(["-A", "-o", "ppid=,command="])
399        .output()
400    else {
401        return false;
402    };
403    String::from_utf8_lossy(&output.stdout).lines().any(|line| {
404        let line = line.trim_start();
405        let Some((ppid, command)) = line.split_once(' ') else {
406            return false;
407        };
408        ppid.parse::<u32>() == Ok(pid) && command.trim_end().ends_with(" message mcp")
409    })
410}
411
412/// Display name of a Codex conversation: `codex-` and the start of its id.
413pub fn codex_name(session_id: &str) -> String {
414    format!("codex-{}", session_id.chars().take(8).collect::<String>())
415}
416
417/// The session a process belongs to: the sender a message is from.
418#[derive(Debug, Clone, PartialEq, Eq)]
419pub struct Caller {
420    /// Its address, where replies go.
421    pub address: MailAddress,
422    /// The name it is known by.
423    pub name: String,
424}
425
426/// Why no session could be found behind a process.
427pub const CALLER_UNRESOLVED: &str = "Can't tell which session is running this command, so \
428    replies would have nowhere to go. Nothing was sent. Run it from your agent session's own \
429    shell tool.";
430
431/// The session behind a process, from its ancestry `pids` (nearest first):
432/// the first that owns a hosted runtime, a Claude session or a Codex
433/// conversation. Never declared by the caller; an environment id only
434/// corroborates, and a mismatch refuses.
435pub fn resolve_caller(homes: &HarnessHomes, pids: &[u32]) -> Result<Caller, String> {
436    let machine = local_machine_name();
437    let registry = read_registry(&registry_dir(homes));
438    let hosted = crate::runtime_mail::controlled_runtimes();
439    for &pid in pids {
440        if let Some(record) = hosted.iter().find(|record| record.pid == pid) {
441            let short: String = record.source.session_id.chars().take(8).collect();
442            return Ok(Caller {
443                address: MailAddress::new(
444                    &machine,
445                    &record.source.harness,
446                    &record.source.session_id,
447                )
448                .map_err(|error| error.to_string())?,
449                name: format!("{}-{short}@{machine}", record.source.harness),
450            });
451        }
452        if let Some(session) = registry.iter().find(|session| session.pid == pid) {
453            if let Ok(claimed) = std::env::var("CLAUDE_CODE_SESSION_ID") {
454                if !claimed.is_empty() && claimed != session.session_id {
455                    return Err(format!(
456                        "{CALLER_UNRESOLVED} (CLAUDE_CODE_SESSION_ID names {claimed}, but the Claude \
457                         process {pid} above this command is session {})",
458                        session.session_id
459                    ));
460                }
461            }
462            return Ok(Caller {
463                address: MailAddress::new(&machine, "claude-code", &session.session_id)
464                    .map_err(|error| error.to_string())?,
465                name: format!("{}@{machine}", session.name),
466            });
467        }
468        // Codex names the thread a command runs for (`CODEX_THREAD_ID`); the
469        // claim stands when this process holds that thread's rollout, as
470        // Codex's shared app-server daemon does for every thread it runs.
471        if let Some(thread) = std::env::var("CODEX_THREAD_ID")
472            .ok()
473            .filter(|thread| !thread.is_empty())
474            .filter(|thread| crate::codex_peer::holds_session(pid, thread))
475        {
476            return Ok(Caller {
477                address: MailAddress::new(&machine, "codex", &thread)
478                    .map_err(|error| error.to_string())?,
479                name: format!("{}@{machine}", codex_name(&thread)),
480            });
481        }
482        if let Some((session_id, _)) = crate::codex_peer::session_of_process(pid) {
483            return Ok(Caller {
484                address: MailAddress::new(&machine, "codex", &session_id)
485                    .map_err(|error| error.to_string())?,
486                name: format!("{}@{machine}", codex_name(&session_id)),
487            });
488        }
489    }
490    #[cfg(windows)]
491    if let Some(session) = msys_cut_claim(&registry, pids) {
492        return Ok(Caller {
493            address: MailAddress::new(&machine, "claude-code", &session.session_id)
494                .map_err(|error| error.to_string())?,
495            name: format!("{}@{machine}", session.name),
496        });
497    }
498    Err(format!(
499        "{CALLER_UNRESOLVED} (looked for this command's processes {pids:?} among {} Claude \
500         sessions in {} and {} hosted runtimes)",
501        registry.len(),
502        registry_dir(homes).display(),
503        hosted.len()
504    ))
505}
506
507/// Git Bash (MSYS) runs a command in a forked process that hands over to it. When the command is
508/// itself an MSYS program (a `sh` script, as npm's command shims are), the forked process exits once
509/// it has handed over, so the Windows parent chain is cut just above that shell and never reaches
510/// the Claude session running it. When the ancestry ends at exactly such a cut (its last pid is gone,
511/// the one before it is an MSYS shell), the session Claude names in `CLAUDE_CODE_SESSION_ID` stands
512/// if it is a live Claude session on this machine.
513#[cfg(windows)]
514fn msys_cut_claim<'a>(
515    registry: &'a [ClaudePeerSession],
516    pids: &[u32],
517) -> Option<&'a ClaudePeerSession> {
518    let table = process_table();
519    let [.., shell, cut] = pids else {
520        return None;
521    };
522    if table.contains_key(cut) {
523        return None;
524    }
525    let name = table.get(shell)?.1.to_ascii_lowercase();
526    if !matches!(name.as_str(), "sh.exe" | "bash.exe" | "dash.exe") {
527        return None;
528    }
529    let claimed = std::env::var("CLAUDE_CODE_SESSION_ID").ok()?;
530    registry
531        .iter()
532        .find(|session| !claimed.is_empty() && session.session_id == claimed)
533}
534
535/// This process's ancestry, nearest first (itself included).
536pub fn process_ancestry() -> Vec<u32> {
537    ancestry_of(std::process::id())
538}
539
540/// A process's ancestry, nearest first (itself included).
541pub fn ancestry_of(pid: u32) -> Vec<u32> {
542    let parents = parent_pids();
543    let mut chain = vec![pid];
544    let mut current = pid;
545    while let Some(&parent) = parents.get(&current) {
546        if parent <= 1 || chain.contains(&parent) {
547            break;
548        }
549        chain.push(parent);
550        current = parent;
551    }
552    chain
553}
554
555/// Every process's parent: pid → parent pid, from `ps`.
556#[cfg(not(windows))]
557fn parent_pids() -> std::collections::HashMap<u32, u32> {
558    let Ok(output) = std::process::Command::new("ps")
559        .args(["-axo", "pid=,ppid="])
560        .output()
561    else {
562        return Default::default();
563    };
564    String::from_utf8_lossy(&output.stdout)
565        .lines()
566        .filter_map(|line| {
567            let mut fields = line.split_whitespace();
568            Some((fields.next()?.parse().ok()?, fields.next()?.parse().ok()?))
569        })
570        .collect()
571}
572
573/// Every process's parent: pid → parent pid, from a process snapshot.
574#[cfg(windows)]
575fn parent_pids() -> std::collections::HashMap<u32, u32> {
576    process_table()
577        .into_iter()
578        .map(|(pid, (parent, _))| (pid, parent))
579        .collect()
580}
581
582/// Every process: pid → (parent pid, executable file name), from a process snapshot.
583#[cfg(windows)]
584fn process_table() -> std::collections::HashMap<u32, (u32, String)> {
585    use windows_sys::Win32::Foundation::{CloseHandle, INVALID_HANDLE_VALUE};
586    use windows_sys::Win32::System::Diagnostics::ToolHelp::{
587        CreateToolhelp32Snapshot, Process32FirstW, Process32NextW, PROCESSENTRY32W,
588        TH32CS_SNAPPROCESS,
589    };
590
591    let mut parents = std::collections::HashMap::new();
592    let snapshot = unsafe { CreateToolhelp32Snapshot(TH32CS_SNAPPROCESS, 0) };
593    if snapshot == INVALID_HANDLE_VALUE {
594        return parents;
595    }
596    let mut entry: PROCESSENTRY32W = unsafe { std::mem::zeroed() };
597    entry.dwSize = std::mem::size_of::<PROCESSENTRY32W>() as u32;
598    let mut has_entry = unsafe { Process32FirstW(snapshot, &mut entry) } != 0;
599    while has_entry {
600        let length = entry
601            .szExeFile
602            .iter()
603            .position(|&unit| unit == 0)
604            .unwrap_or(entry.szExeFile.len());
605        let name = String::from_utf16_lossy(&entry.szExeFile[..length]);
606        parents.insert(entry.th32ProcessID, (entry.th32ParentProcessID, name));
607        has_entry = unsafe { Process32NextW(snapshot, &mut entry) } != 0;
608    }
609    unsafe {
610        CloseHandle(snapshot);
611    }
612    parents
613}
614
615/// Codex's user-level hooks file.
616pub fn codex_hooks_path() -> std::path::PathBuf {
617    std::env::var_os("CODEX_HOME")
618        .map(std::path::PathBuf::from)
619        .or_else(|| {
620            supercode_interchange::user_home()
621                .map(std::path::PathBuf::into_os_string)
622                .map(|home| std::path::PathBuf::from(home).join(".codex"))
623        })
624        .unwrap_or_else(|| std::path::PathBuf::from(".codex"))
625        .join("hooks.json")
626}
627
628/// Whether supercode's mail hook is in Codex's user hooks file. (Codex runs
629/// it only once its user has trusted it.)
630pub fn codex_user_hook_installed() -> bool {
631    std::fs::read_to_string(codex_hooks_path())
632        .is_ok_and(|text| text.contains(CODEX_HOOK_ARGUMENTS))
633}
634
635/// Whether supercode's mail hook is in a project's Codex hooks file, in the
636/// session's directory or one above it.
637pub fn codex_project_hook_installed(cwd: &Path) -> bool {
638    cwd.ancestors().any(|directory| {
639        std::fs::read_to_string(directory.join(".codex").join("hooks.json"))
640            .is_ok_and(|text| text.contains(CODEX_HOOK_ARGUMENTS))
641    })
642}
643
644/// How a delivery ended.
645#[derive(Debug, Clone, PartialEq, Eq)]
646pub enum Delivered {
647    /// The receiver has it now: steered into a running turn.
648    Steered,
649    /// The receiver has it now: it started a turn in an idle session.
650    Started,
651    /// Claude reported it in the receiver's inbox; `busy` says whether the
652    /// receiver reads it at its next tool call (true) or it starts a turn.
653    Native {
654        /// True when the receiver was in a turn.
655        busy: bool,
656    },
657    /// Filed in the receiver's mailbox, which a hook points it at.
658    Hooked,
659    /// Filed in the receiver's mailbox, waiting to be read.
660    Queued,
661    /// Filed with no door to show it.
662    Stored,
663    /// Filed in an operator's mailbox.
664    Operator,
665}
666
667/// A delivery refused before anything was sent.
668#[derive(Debug, Clone, PartialEq, Eq)]
669pub enum Refused {
670    /// `--queue` cannot hold a message for an idle Claude session: Claude
671    /// starts a turn for every message it delivers.
672    CannotQueueNative,
673    /// The relayed text would not fit one relay turn.
674    TooLong(usize),
675}
676
677/// Largest message relayed into a Claude session. The relay copies it into a
678/// model turn byte for byte, so it must fit comfortably in one.
679pub const MAX_RELAYED_BYTES: usize = 100_000;
680
681/// Deliver `envelope` to `to` through `door`. With `wake` false an idle
682/// receiver is not started. With `notify_when_idle` the sender gets one idle
683/// notice after the receiver's next turn ends.
684pub async fn deliver(
685    envelope: &Envelope,
686    to: &MailAddress,
687    door: &Door,
688    wake: bool,
689    notify_when_idle: bool,
690) -> Result<Result<Delivered, Refused>, String> {
691    let mailbox = Mailbox::open(&mail_root(), to).map_err(|error| error.to_string())?;
692    // A runtime answers with its turn's final message, which the watcher
693    // sends back: not every runtime can run a command (a hosted agent may
694    // have no shell), and each can end a turn.
695    let mut final_reply = false;
696    let delivered = match door {
697        Door::Runtime(record) => {
698            let mut sent = envelope.clone();
699            let answers = matches!(envelope.reply_via, ReplyVia::Command);
700            if answers {
701                sent.reply_via = ReplyVia::FinalMessage {
702                    destination: envelope.from_name.clone(),
703                };
704            }
705            match deliver_to_runtime(record, sent.render(), wake).await? {
706                RuntimeDelivery::Steered => {
707                    mailbox
708                        .deliver_read(&sent)
709                        .map_err(|error| error.to_string())?;
710                    final_reply = answers;
711                    Delivered::Steered
712                }
713                RuntimeDelivery::Started => {
714                    mailbox
715                        .deliver_read(&sent)
716                        .map_err(|error| error.to_string())?;
717                    final_reply = answers;
718                    Delivered::Started
719                }
720                RuntimeDelivery::NotWoken => {
721                    mailbox
722                        .deliver(envelope)
723                        .map_err(|error| error.to_string())?;
724                    Delivered::Queued
725                }
726            }
727        }
728        Door::Native(session) => {
729            let busy = session.status != Some(ClaudePeerStatus::Idle);
730            if !wake && !busy {
731                return Ok(Err(Refused::CannotQueueNative));
732            }
733            // The same envelope every door delivers; Claude wraps it in its
734            // own, which names the relay. Its reply line names the door this
735            // session really has: its tools when it runs supercode's MCP server.
736            let mut envelope = envelope.clone();
737            if envelope.reply_via == ReplyVia::Command && has_message_tools(session.pid) {
738                envelope.reply_via = ReplyVia::Tool;
739            }
740            let envelope = &envelope;
741            let text = envelope.render();
742            if text.len() > MAX_RELAYED_BYTES {
743                return Ok(Err(Refused::TooLong(text.len())));
744            }
745            match send_through_relay(
746                &envelope.from,
747                &envelope.from_name,
748                &session.name,
749                text,
750                &envelope.id,
751            )
752            .await
753            {
754                RelayReceipt::Delivered { .. } => {
755                    mailbox
756                        .deliver_read(envelope)
757                        .map_err(|error| error.to_string())?;
758                    Delivered::Native { busy }
759                }
760                RelayReceipt::Failed { detail } => return Err(detail),
761            }
762        }
763        Door::Hook => {
764            mailbox
765                .deliver(envelope)
766                .map_err(|error| error.to_string())?;
767            Delivered::Hooked
768        }
769        Door::Stored => {
770            mailbox
771                .deliver(envelope)
772                .map_err(|error| error.to_string())?;
773            Delivered::Stored
774        }
775        Door::Operator => {
776            mailbox
777                .deliver(envelope)
778                .map_err(|error| error.to_string())?;
779            Delivered::Operator
780        }
781    };
782    let notice = notify_when_idle && !matches!(door, Door::Operator);
783    if notice || final_reply {
784        let mut subscription = IdleSubscription::new(envelope.id.clone(), envelope.from.clone());
785        subscription.notice = notice;
786        subscription.final_reply = final_reply;
787        mailbox
788            .subscribe_idle(&subscription)
789            .map_err(|error| error.to_string())?;
790        // The watcher that settles it runs beside the machine daemon.
791        if let Ok(program) = supercode_program() {
792            crate::claude_relay::ensure_machine_daemon(&program)
793                .await
794                .ok();
795        }
796    }
797    Ok(Ok(delivered))
798}
799
800/// How the user's own turn reached its session.
801#[derive(Debug, Clone, Copy, PartialEq, Eq)]
802pub enum UserTurn {
803    /// A hosted runtime was in a turn; the words were steered into it.
804    Steered,
805    /// A hosted runtime was idle; the words started a turn.
806    Started,
807    /// Typed into the session's pane as its own submitted turn.
808    Typed,
809    /// The pane's composer holds a draft, or its program is not at its
810    /// composer: the turn waits in the session's mailbox and is typed once
811    /// the composer is empty.
812    Waiting,
813}
814
815impl UserTurn {
816    /// Stable wire spelling.
817    pub const fn as_str(self) -> &'static str {
818        match self {
819            Self::Steered => "steered",
820            Self::Started => "started",
821            Self::Typed => "typed",
822            Self::Waiting => "waiting",
823        }
824    }
825}
826
827/// The daemon pane a session runs in, when it runs in one.
828pub fn daemon_pane(session: &LiveSession) -> Option<String> {
829    let name = session.tmux.as_deref()?.split(':').next()?;
830    let rest = name.strip_prefix("p_")?;
831    (!rest.is_empty()
832        && rest
833            .chars()
834            .all(|c| c.is_ascii_alphanumeric() || c == '_' || c == '-'))
835    .then(|| name.to_string())
836}
837
838/// Deliver the user's own turn to `to` through the one door that carries the
839/// user's authority: a hosted runtime's own input, or the session's pane.
840/// A session with neither (running outside supercode) is refused, never
841/// reached as a peer instead. Callers hold the owner's authority already:
842/// this is reached only through owner doors (the machine's own harness.v1).
843pub async fn deliver_user_turn(
844    homes: &HarnessHomes,
845    envelope: &Envelope,
846    to: &MailAddress,
847) -> Result<UserTurn, String> {
848    let session = LiveSessions::read(homes)
849        .sessions
850        .into_iter()
851        .find(|session| &session.address == to)
852        .ok_or_else(|| format!("{to} is not running; nothing was sent"))?;
853    let mailbox = Mailbox::open(&mail_root(), to).map_err(|error| error.to_string())?;
854    if let Door::Runtime(record) = &session.door {
855        let delivered = deliver_to_runtime(record, envelope.body.clone(), true).await?;
856        mailbox
857            .deliver_read(envelope)
858            .map_err(|error| error.to_string())?;
859        return Ok(match delivered {
860            RuntimeDelivery::Steered => UserTurn::Steered,
861            _ => UserTurn::Started,
862        });
863    }
864    let Some(pane) = daemon_pane(&session) else {
865        let name = session.name.split('@').next().unwrap_or(&session.name);
866        return Err(format!(
867            "{name} runs outside supercode, where nothing can speak as its user. Open it in a \
868             pane with `supercode open {name}`; nothing was sent."
869        ));
870    };
871    mailbox
872        .deliver(envelope)
873        .map_err(|error| error.to_string())?;
874    let typed = type_user_turns(&mailbox, &pane).await;
875    Ok(if typed.contains(&envelope.id) {
876        UserTurn::Typed
877    } else {
878        UserTurn::Waiting
879    })
880}
881
882/// Type the user's waiting turns into `pane`, oldest first, each only when
883/// the pane's composer is empty. Stops at the first that has to wait.
884/// Returns the ids typed.
885pub async fn type_user_turns(mailbox: &Mailbox, pane: &str) -> Vec<String> {
886    let mut typed = Vec::new();
887    for waiting in mailbox.user_turns().unwrap_or_default() {
888        // Claimed before it is typed: the machine's watcher and a direct delivery both type waiting turns, and
889        // two typers filling one composer sent a turn's text twice in one prompt. A turn another typer holds is
890        // that typer's, and so is every turn after it.
891        let Ok(Some(stored)) = mailbox.claim_user_turn(&waiting) else {
892            break;
893        };
894        match submit_when_composer_empty(pane, &stored.envelope.body).await {
895            Ok(true) => {
896                mailbox.acknowledge(&stored).ok();
897                typed.push(stored.envelope.id.clone());
898            }
899            Ok(false) => {
900                mailbox.release(&stored).ok();
901                break;
902            }
903            Err(error) => {
904                mailbox.release(&stored).ok();
905                eprintln!(
906                    "supercode: the user's turn {} for {} waits: {error}",
907                    stored.envelope.id,
908                    mailbox.address()
909                );
910                break;
911            }
912        }
913    }
914    typed
915}
916
917/// The daemon's pane door: type `text` as a submitted turn if the composer
918/// is empty now. `Ok(false)` when it holds a draft or is not on screen.
919async fn submit_when_composer_empty(pane: &str, text: &str) -> Result<bool, String> {
920    let entry = crate::teams_entry().map_err(|error| error.to_string())?;
921    let node = std::env::var(crate::orchestrator_door::NODE_BIN_ENV)
922        .ok()
923        .filter(|value| !value.trim().is_empty())
924        .unwrap_or_else(|| "node".into());
925    let output = tokio::process::Command::new(node)
926        .arg(entry)
927        .args(["input", pane, text, "--when-composer-empty"])
928        .stdin(std::process::Stdio::null())
929        .output()
930        .await
931        .map_err(|error| error.to_string())?;
932    if !output.status.success() {
933        return Err(crate::mailbox::error_line(&String::from_utf8_lossy(
934            &output.stderr,
935        )));
936    }
937    let answer: serde_json::Value =
938        serde_json::from_slice(&output.stdout).map_err(|error| error.to_string())?;
939    Ok(answer["delivered"] == true)
940}