supercode_harness/

claude_relay.rs

//! Claude relays: the degraded tier's door into a Claude Code session that
//! supercode does not control.
//!
//! A Claude session answers a message by copying its `from` into
//! `SendMessage`. For that answer to come back, the message must come FROM a
//! live Claude peer. A relay is that peer: a Claude Code runtime that
//! supercode hosts like any other (`runtimes.start` in the machine daemon's
//! `harness serve`, registered in the live-runtime registry), named
//! `sc-<name>-on-<machine>`, sending on one session's behalf. Nothing here
//! runs its own process supervisor or writes Claude's private socket frame.
//!
//! Three rules earn their place here:
//!
//! 1. **No model runs.** A relay's API is supercode's own deterministic
//!    endpoint ([`crate::relay_endpoint`]), which answers a send turn with a
//!    `SendMessage` of the queued pair. Every send is queued first, and a
//!    PreToolUse hook (`supercode message gate`) still denies any tool call
//!    but a `SendMessage` of exactly that pair.
//! 2. **Receipts come from Claude, through hooks.** A PostToolUse hook writes
//!    Claude's own `SendMessage` result beside the queue; a denied call and a
//!    turn that ends without sending write theirs too. The sender waits on
//!    that file, not on the model's words.
//! 3. **Inbound mail never reaches the model.** Claude hands every inbound
//!    peer envelope and idle or delivery notice to the relay's
//!    `UserPromptSubmit` hook byte-exact; the hook files it in the represented
//!    session's mailbox and blocks the prompt, so no model turn runs for it
//!    (measured: `duration_api_ms: 0`).
//!
//! A relay's busy/idle is its own; its name carries the `sc-` prefix and
//! every turn ends with [`RELAY_STATUS_LINE`].

use std::collections::{BTreeMap, HashMap};
use std::path::{Path, PathBuf};
use std::time::{Duration, Instant};

use serde::{Deserialize, Serialize};
use serde_json::{json, Value};

use crate::claude_peer::{read_registry, registry_dir, ClaudePeerSession};
use crate::mailbox::{mail_root, Envelope, MailAddress, MailKind, ReplyVia};
use crate::HarnessHomes;

/// Model name a relay's requests carry. The relay endpoint answers them
/// without any model.
pub const RELAY_MODEL: &str = "haiku";

/// Prefix marking a registry name as a relay rather than a session.
pub const RELAY_NAME_PREFIX: &str = "sc-";

/// Text every relay turn ends with. Claude quotes a session's last line in
/// its idle notices, so this is where the correction reaches the agent.
pub const RELAY_STATUS_LINE: &str = "relay, not the session's status";

/// How long a send waits for Claude's own receipt.
pub const SEND_TIMEOUT: Duration = Duration::from_secs(90);

/// Tools a relay may touch.
const RELAY_TOOLS: &str = "ListAgents,SendMessage";

/// Line Claude opens a delivered peer message with in a turn. The hook sees
/// the envelope itself; one of the two starts is required, so text that
/// merely quotes an envelope further in is never read as mail.
const NATIVE_PEER_PREAMBLE: &str = "Another Claude session sent a message:";
/// Tags of Claude's peer envelope.
const NATIVE_PEER_OPENING: &str = "<cross-session-message ";
const NATIVE_PEER_CLOSING: &str = "</cross-session-message>";
const NATIVE_IDLE_NOTICE: &str = "[Cross-session idle notice]";
const NATIVE_DELIVERY_NOTICE: &str = "[Cross-session delivery notice]";

/// Registry name of the relay that represents `name` on `machine`.
///
/// Not `name@machine`: Claude accepts `@` in `--name`, but its `SendMessage`
/// refuses a `to` containing `@` ("must be a bare teammate name").
pub fn relay_name(name: &str, machine: &str) -> String {
    let base = name.strip_prefix(RELAY_NAME_PREFIX).unwrap_or(name);
    format!("{RELAY_NAME_PREFIX}{base}-on-{machine}")
}

/// The one send a relay is allowed to make next.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct QueuedSend {
    /// Registry name of the receiving Claude session.
    pub to: String,
    /// Exact message text.
    pub message: String,
}

/// Decide one PreToolUse hook call. `None` allows it; `Some(reason)` denies.
///
/// Only a `SendMessage` whose `to` and `message` equal the queued pair is
/// allowed; `ListAgents` is read-only and always allowed; anything else, or a
/// send with nothing queued, is denied.
pub fn gate_decision(hook_input: &Value, queued: Option<&QueuedSend>) -> Option<String> {
    let tool = hook_input
        .get("tool_name")
        .and_then(Value::as_str)
        .unwrap_or_default();
    if tool == "ListAgents" {
        return None;
    }
    if tool != "SendMessage" {
        return Some(format!("a relay may not use {tool}"));
    }
    let Some(queued) = queued else {
        return Some(
            "nothing is queued to send; mail for the host is not answered by the relay".into(),
        );
    };
    let input = hook_input.get("tool_input").cloned().unwrap_or(Value::Null);
    let to = input.get("to").and_then(Value::as_str).unwrap_or_default();
    let message = input
        .get("message")
        .and_then(Value::as_str)
        .unwrap_or_default();
    if to != queued.to {
        return Some(format!(
            "not the queued outbound message: `to` is {to:?}, the queue says {:?}",
            queued.to
        ));
    }
    if message != queued.message {
        let at = message
            .char_indices()
            .zip(queued.message.chars())
            .find(|((_, sent), queued)| sent != queued)
            .map(|((index, _), _)| index)
            .unwrap_or_else(|| message.len().min(queued.message.len()));
        return Some(format!(
            "not the queued outbound message: `message` ({} bytes) differs from the queue ({} \
             bytes) at byte {at}: sent {:?}, queued {:?}",
            message.len(),
            queued.message.len(),
            message
                .get(at..)
                .unwrap_or_default()
                .chars()
                .take(40)
                .collect::<String>(),
            queued
                .message
                .get(at..)
                .unwrap_or_default()
                .chars()
                .take(40)
                .collect::<String>(),
        ));
    }
    None
}

/// The hook's JSON answer for a denial.
pub fn gate_denial(reason: &str) -> Value {
    json!({
        "hookSpecificOutput": {
            "hookEventName": "PreToolUse",
            "permissionDecision": "deny",
            "permissionDecisionReason": reason,
        }
    })
}

/// Files of one relay, all in its own directory (also its working directory,
/// which holds no project settings).
#[derive(Debug, Clone)]
pub struct RelayPaths {
    /// The relay's directory.
    pub directory: PathBuf,
    /// The single queued send the gate compares against.
    pub queue: PathBuf,
    /// Claude's answer to the queued send, written by the relay's hooks.
    pub receipt: PathBuf,
    /// The prompt that began the relay's current turn, written as the turn starts: a turn's end
    /// answers the queued send only when that send began it.
    pub turn: PathBuf,
    /// Settings passed with `--settings`, carrying the hooks.
    pub settings: PathBuf,
    /// The hosted runtime this relay runs as.
    pub record: PathBuf,
    /// Newest message id sent to each Claude session name, for threading.
    pub sent: PathBuf,
    /// Serializes sends through this relay.
    pub lock: PathBuf,
}

impl RelayPaths {
    /// Paths of the relay named `relay_name` under `root`.
    pub fn new(root: &Path, relay_name: &str) -> Self {
        let hash = blake3::hash(relay_name.as_bytes()).to_hex();
        Self::in_directory(root.join("relays").join(&hash[..24]))
    }

    /// Paths of the relay whose directory is `directory`.
    pub fn in_directory(directory: PathBuf) -> Self {
        Self {
            queue: directory.join("queue.json"),
            receipt: directory.join("receipt.json"),
            turn: directory.join("turn.txt"),
            settings: directory.join("settings.json"),
            record: directory.join("relay.json"),
            sent: directory.join("sent.json"),
            lock: directory.join("send.lock"),
            directory,
        }
    }

    /// Newest message id sent to each Claude session name.
    pub fn last_sent(&self) -> HashMap<String, String> {
        std::fs::read(&self.sent)
            .ok()
            .and_then(|bytes| serde_json::from_slice(&bytes).ok())
            .unwrap_or_default()
    }
}

/// Everything needed to start one relay.
#[derive(Debug, Clone)]
pub struct RelaySpec {
    /// Session the relay speaks for.
    pub represented: MailAddress,
    /// Name the represented session is known by.
    pub represented_name: String,
    /// Registry name the relay registers under.
    pub name: String,
    /// Relay files.
    pub paths: RelayPaths,
    /// Program run by the hooks (this supercode binary).
    pub program: PathBuf,
}

impl RelaySpec {
    /// The relay that speaks for `represented`.
    pub fn for_sender(represented: &MailAddress, represented_name: &str) -> std::io::Result<Self> {
        let base = represented_name
            .split('@')
            .next()
            .unwrap_or(represented_name);
        let name = relay_name(base, &represented.machine);
        Ok(Self {
            represented: represented.clone(),
            represented_name: represented_name.to_string(),
            paths: RelayPaths::new(&mail_root(), &name),
            name,
            program: supercode_program()?,
        })
    }
}

/// Arguments of the Claude Code runtime a relay runs as.
///
/// `--setting-sources project` in an empty directory keeps user hooks and
/// settings out; `--settings` then adds only the relay's hooks. The API is
/// the relay endpoint, through the environment ([`relay_environment`]).
pub fn relay_arguments(spec: &RelaySpec) -> Vec<String> {
    vec![
        "--print".into(),
        "--input-format".into(),
        "stream-json".into(),
        "--output-format".into(),
        "stream-json".into(),
        "--verbose".into(),
        "--permission-prompt-tool".into(),
        "stdio".into(),
        "--model".into(),
        RELAY_MODEL.into(),
        "--name".into(),
        spec.name.clone(),
        "--permission-mode".into(),
        "bypassPermissions".into(),
        "--setting-sources".into(),
        "project".into(),
        "--settings".into(),
        spec.paths.settings.to_string_lossy().into_owned(),
        "--tools".into(),
        RELAY_TOOLS.into(),
        "--no-session-persistence".into(),
    ]
}

/// Environment of a relay: its API is the relay endpoint at `endpoint`, and
/// its API key names its directory there. The key also keeps account
/// connectors (MCP servers) out of the relay.
pub fn relay_environment(spec: &RelaySpec, endpoint: &str) -> BTreeMap<String, String> {
    let key = spec
        .paths
        .directory
        .file_name()
        .map(|name| name.to_string_lossy().into_owned())
        .unwrap_or_default();
    BTreeMap::from([
        ("ANTHROPIC_BASE_URL".to_string(), endpoint.to_string()),
        ("ANTHROPIC_API_KEY".to_string(), key),
        (
            "CLAUDE_CODE_DISABLE_NONESSENTIAL_TRAFFIC".to_string(),
            "1".to_string(),
        ),
    ])
}

/// Settings document installing a relay's hooks.
pub fn relay_settings(spec: &RelaySpec) -> Value {
    let program = shell_quote(&spec.program.to_string_lossy());
    let directory = shell_quote(&spec.paths.directory.to_string_lossy());
    let represented = shell_quote(&spec.represented.to_string());
    let command = |verb: &str| format!("{program} message {verb} {directory}");
    json!({
        "hooks": {
            // Every tool, not only the two named in --tools.
            "PreToolUse": [{
                "matcher": ".*",
                "hooks": [{"type": "command", "command": command("gate")}],
            }],
            "PostToolUse": [{
                "matcher": "SendMessage",
                "hooks": [{"type": "command", "command": command("relay-receipt")}],
            }],
            // A SendMessage that fails fires this, with Claude Code's own error.
            "PostToolUseFailure": [{
                "matcher": "SendMessage",
                "hooks": [{"type": "command", "command": command("relay-receipt")}],
            }],
            "Stop": [{
                "hooks": [{"type": "command", "command": command("relay-receipt")}],
            }],
            // A turn an API error ends (a usage limit, an overload) fires this, not Stop.
            "StopFailure": [{
                "hooks": [{"type": "command", "command": command("relay-receipt")}],
            }],
            "UserPromptSubmit": [{
                "hooks": [{
                    "type": "command",
                    "command": format!("{program} message relay-inbound {represented} {directory}"),
                }],
            }],
        }
    })
}

fn shell_quote(value: &str) -> String {
    format!("'{}'", value.replace('\'', "'\\''"))
}

/// Whether a prompt Claude handed the relay is inbound mail (a peer envelope
/// or a notice) rather than its host's own send turn.
pub fn is_inbound_prompt(prompt: &str) -> bool {
    let trimmed = prompt.trim_start();
    trimmed.starts_with(NATIVE_PEER_OPENING)
        || trimmed.starts_with(NATIVE_IDLE_NOTICE)
        || trimmed.starts_with(NATIVE_DELIVERY_NOTICE)
}

/// The UserPromptSubmit answer that keeps a prompt from the model.
pub fn inbound_block() -> Value {
    json!({
        "decision": "block",
        "reason": "filed in the mailbox of the session this relay speaks for",
    })
}

/// The host turn asking a relay to make one send.
pub fn send_turn(send: &QueuedSend) -> String {
    format!(
        "Send one message with SendMessage.\n\
         to: {}\n\
         message: the exact text between the markers, without the markers\n\
         ---BEGIN MESSAGE---\n{}\n---END MESSAGE---",
        send.to, send.message
    )
}

/// A native peer envelope as Claude delivered it to a relay.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct NativeEnvelope {
    /// Claude's `from` (the sender's socket, `uds:/tmp/cc-socks/<pid>.sock`).
    pub from: String,
    /// Claude's `from-name`.
    pub from_name: Option<String>,
    /// Message body, unescaped as Claude delivered it.
    pub body: String,
}

/// One inbound prompt, as Claude handed it to a relay's hook.
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum RelayEvent {
    /// A peer message arrived.
    Peer(NativeEnvelope),
    /// A `[Cross-session idle notice]`.
    IdleNotice(String),
    /// A `[Cross-session delivery notice]`.
    DeliveryNotice(String),
}

/// The live Claude session behind a native `uds:` sender address.
pub fn resolve_native_sender(
    registry: &[ClaudePeerSession],
    native_from: &str,
) -> Option<ClaudePeerSession> {
    let socket = native_from.strip_prefix("uds:").unwrap_or(native_from);
    registry
        .iter()
        .find(|session| session.socket_path.to_string_lossy() == socket)
        .cloned()
}

/// Outcome of one relayed send, as the sender reads it.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(rename_all = "snake_case", tag = "outcome")]
pub enum RelayReceipt {
    /// Claude reported the message queued in the receiver's inbox.
    Delivered {
        /// Claude's own result text.
        detail: String,
        /// Claude's message id, when it reported one.
        native_msg_id: Option<String>,
    },
    /// The send did not happen.
    Failed {
        /// Why, in words the sending agent can act on.
        detail: String,
    },
}

/// Read a receipt the relay's hooks wrote: Claude's own `SendMessage`
/// result, a gate denial, or a turn that ended without sending.
pub fn read_receipt(value: &Value) -> RelayReceipt {
    if let Some(reason) = value.get("denied").and_then(Value::as_str) {
        return RelayReceipt::Failed {
            detail: format!("the relay's gate refused the send: {reason}"),
        };
    }
    if let Some(error) = value.get("send_failed").and_then(Value::as_str) {
        return RelayReceipt::Failed {
            detail: format!("Claude Code refused the relay's send: {}", error.trim()),
        };
    }
    if value.get("turn_ended").and_then(Value::as_bool) == Some(true) {
        let said = value
            .get("said")
            .and_then(Value::as_str)
            .filter(|said| !said.trim().is_empty())
            .map(|said| format!("; it said: {}", said.trim()))
            .unwrap_or_default();
        return RelayReceipt::Failed {
            detail: format!(
                "the Claude relay ended its turn without sending; nothing was sent{said}"
            ),
        };
    }
    let response = value.get("tool_response").cloned().unwrap_or(Value::Null);
    let parsed = match &response {
        Value::String(text) => serde_json::from_str::<Value>(text).unwrap_or(response.clone()),
        Value::Array(blocks) => blocks
            .iter()
            .find_map(|block| block.get("text").and_then(Value::as_str))
            .and_then(|text| serde_json::from_str::<Value>(text).ok())
            .unwrap_or(response.clone()),
        _ => response.clone(),
    };
    if parsed.get("success").and_then(Value::as_bool) != Some(true) {
        return RelayReceipt::Failed {
            detail: format!("Claude did not report the send as successful: {parsed}"),
        };
    }
    RelayReceipt::Delivered {
        detail: parsed
            .get("message")
            .and_then(Value::as_str)
            .unwrap_or_default()
            .to_string(),
        native_msg_id: parsed
            .get("msg_id")
            .and_then(Value::as_str)
            .map(str::to_string),
    }
}

/// Send `message` to the Claude session named `to`, from `sender` through
/// its relay: start the relay as a hosted runtime when it is not running,
/// queue the send, hand it the send turn through the runtime's own door, and
/// wait for Claude's receipt.
pub async fn send_through_relay(
    sender: &MailAddress,
    sender_name: &str,
    to: &str,
    message: String,
    message_id: &str,
) -> RelayReceipt {
    let failed = |detail: String| RelayReceipt::Failed { detail };
    let spec = match RelaySpec::for_sender(sender, sender_name) {
        Ok(spec) => spec,
        Err(error) => return failed(error.to_string()),
    };
    if let Err(error) = std::fs::create_dir_all(&spec.paths.directory) {
        return failed(error.to_string());
    }
    // One send at a time through one relay: the queue file is the gate.
    let lock = match tokio::task::spawn_blocking({
        let path = spec.paths.lock.clone();
        move || SendLock::acquire(&path)
    })
    .await
    {
        Ok(Ok(lock)) => lock,
        Ok(Err(error)) => return failed(format!("could not take the relay's send lock: {error}")),
        Err(error) => return failed(error.to_string()),
    };
    let runtime = match ensure_relay_runtime(&spec).await {
        Ok(runtime) => runtime,
        Err(detail) => return failed(detail),
    };
    let queued = QueuedSend {
        to: to.to_string(),
        message,
    };
    std::fs::remove_file(&spec.paths.receipt).ok();
    if let Err(error) = std::fs::write(
        &spec.paths.queue,
        serde_json::to_vec(&queued).unwrap_or_default(),
    ) {
        return failed(error.to_string());
    }
    let delivered =
        crate::runtime_mail::deliver_to_runtime(&runtime, send_turn(&queued), true).await;
    let receipt = match delivered {
        Err(detail) => failed(format!("could not reach the Claude relay: {detail}")),
        Ok(_) => wait_for_receipt(&spec.paths.receipt).await,
    };
    std::fs::remove_file(&spec.paths.queue).ok();
    if matches!(receipt, RelayReceipt::Delivered { .. }) {
        let mut sent = spec.paths.last_sent();
        sent.insert(to.to_string(), message_id.to_string());
        std::fs::write(
            &spec.paths.sent,
            serde_json::to_vec(&sent).unwrap_or_default(),
        )
        .ok();
    }
    drop(lock);
    receipt
}

async fn wait_for_receipt(path: &Path) -> RelayReceipt {
    let started = Instant::now();
    while started.elapsed() < SEND_TIMEOUT {
        if let Some(value) = std::fs::read(path)
            .ok()
            .and_then(|bytes| serde_json::from_slice::<Value>(&bytes).ok())
        {
            return read_receipt(&value);
        }
        tokio::time::sleep(Duration::from_millis(200)).await;
    }
    RelayReceipt::Failed {
        detail: format!(
            "the Claude relay did not confirm the send within {} seconds; it may still arrive",
            SEND_TIMEOUT.as_secs()
        ),
    }
}

/// The relay's hosted runtime, started in the machine daemon when it is not
/// running.
async fn ensure_relay_runtime(
    spec: &RelaySpec,
) -> Result<crate::live_runtime::LiveRuntimeRecord, String> {
    #[derive(Serialize, Deserialize)]
    struct Record {
        runtime_id: String,
        #[serde(default)]
        endpoint: String,
        /// The machine daemon's connection to the runtime: the door that closes it.
        #[serde(default)]
        connection: String,
    }
    let endpoint = relay_endpoint(&spec.program).await?;
    // Written on every send, not only at start: Claude Code reloads a changed settings file, so
    // a running relay takes the hooks this supercode installs.
    std::fs::write(
        &spec.paths.settings,
        serde_json::to_vec_pretty(&relay_settings(spec)).unwrap_or_default(),
    )
    .map_err(|error| error.to_string())?;
    if let Some(record) = std::fs::read(&spec.paths.record)
        .ok()
        .and_then(|bytes| serde_json::from_slice::<Record>(&bytes).ok())
    {
        if let Some(runtime) =
            crate::runtime_mail::controlled_runtime("claude-code", &record.runtime_id)
        {
            // Every relay endpoint answers from the relay's own queue, so one the relay was started
            // against serves it as long as it answers, whichever watcher now owns the record (two
            // watchers on one mail root swapped it per send, and each swap started a relay).
            if record.endpoint == endpoint
                || crate::relay_endpoint::endpoint_answers(&record.endpoint)
            {
                return Ok(runtime);
            }
            // Started against an endpoint no longer serving, its requests
            // would go nowhere: it ends, and a new one starts.
            end_relay_process(&spec.name);
        }
        // The replaced runtime leaves the daemon's table too: ending only its process left its
        // listener and live receipt behind for every client to poll (t_f3f102db t_4f29b43e).
        if !record.connection.is_empty() {
            machine_rpc(
                &spec.program,
                "runtimes.close",
                &json!({ "connection": record.connection, "runtime_id": record.runtime_id }),
            )
            .await
            .ok();
        }
    }
    let params = json!({
        "harness": "claude-code",
        "launch": {
            "program": "claude",
            "arguments": relay_arguments(spec),
            "env": relay_environment(spec, &endpoint),
        },
        "cwd": spec.paths.directory,
    });
    let result = machine_rpc(&spec.program, "runtimes.start", &params).await?;
    let runtime_id = result
        .pointer("/handle/runtime_id")
        .and_then(Value::as_str)
        .ok_or_else(|| format!("the machine daemon did not start the relay: {result}"))?
        .to_string();
    let connection = result
        .get("connection")
        .and_then(Value::as_str)
        .unwrap_or_default()
        .to_string();
    std::fs::write(
        &spec.paths.record,
        serde_json::to_vec(&Record {
            runtime_id: runtime_id.clone(),
            endpoint,
            connection,
        })
        .unwrap_or_default(),
    )
    .map_err(|error| error.to_string())?;
    crate::runtime_mail::controlled_runtime("claude-code", &runtime_id)
        .ok_or_else(|| "the relay started but registered no live runtime".to_string())
}

/// The relays of sessions no longer running, retired: each relay's runtime (a `claude --print` under
/// `supercode harness serve`, some hundreds of MB) is closed through the machine daemon that
/// started it (`runtimes.close`, so the daemon's own table forgets it: a relay process ended behind
/// its back left sends failing until the daemon restarted), and its folder removed. A relay speaks
/// for one session and receives the replies to what that session sent; once the session has ended,
/// nothing reaches it. A relay for a board or an operator, which is not a session, is kept; so is a
/// live relay recorded before its connection was kept, which no door can close cleanly.
/// Answers the represented addresses whose relays were retired.
pub async fn retire_relays_of_ended_sessions(homes: &HarnessHomes) -> Vec<String> {
    #[derive(Deserialize)]
    struct Record {
        runtime_id: String,
        #[serde(default)]
        connection: String,
    }
    let Ok(entries) = std::fs::read_dir(mail_root().join("relays")) else {
        return Vec::new();
    };
    let running: std::collections::HashSet<String> = crate::mail_route::LiveSessions::read(homes)
        .all()
        .iter()
        .map(|session| session.address.to_string())
        .collect();
    let Ok(program) = supercode_program() else {
        return Vec::new();
    };
    let mut retired = Vec::new();
    for entry in entries.flatten() {
        let paths = RelayPaths::in_directory(entry.path());
        // the address the relay speaks for, as its inbound hook names it
        let Some(represented) = std::fs::read_to_string(&paths.settings)
            .ok()
            .and_then(|text| {
                let from = text.find("relay-inbound '")? + "relay-inbound '".len();
                let len = text[from..].find('\'')?;
                MailAddress::parse(&text[from..from + len]).ok()
            })
        else {
            continue;
        };
        if matches!(represented.harness.as_str(), "board" | "operator")
            || running.contains(&represented.to_string())
        {
            continue;
        }
        let record = std::fs::read(&paths.record)
            .ok()
            .and_then(|bytes| serde_json::from_slice::<Record>(&bytes).ok());
        let alive = record.as_ref().is_some_and(|record| {
            crate::runtime_mail::controlled_runtime("claude-code", &record.runtime_id).is_some()
        });
        if alive {
            let (connection, runtime_id) = record
                .map(|record| (record.connection, record.runtime_id))
                .unwrap_or_default();
            if connection.is_empty()
                || machine_rpc(
                    &program,
                    "runtimes.close",
                    &json!({ "connection": connection, "runtime_id": runtime_id }),
                )
                .await
                .is_err()
            {
                continue;
            }
        }
        if std::fs::remove_dir_all(&paths.directory).is_ok() {
            retired.push(represented.to_string());
        }
    }
    retired
}

/// End the Claude process registered under the relay name `name`.
fn end_relay_process(name: &str) {
    let registry = registry_dir(&HarnessHomes::default());
    for session in read_registry(&registry) {
        #[cfg(unix)]
        if session.name == name {
            // SAFETY: `kill` only sends a signal to the relay's own process.
            unsafe {
                libc::kill(session.pid as libc::pid_t, libc::SIGTERM);
            }
        }
    }
}

/// The relay endpoint's base URL. The machine daemon's watcher serves it;
/// the daemon is started when it is not running.
async fn relay_endpoint(program: &Path) -> Result<String, String> {
    if let Some(url) = crate::relay_endpoint::relay_endpoint_url() {
        return Ok(url);
    }
    ensure_machine_daemon(program).await?;
    let deadline = Instant::now() + Duration::from_secs(10);
    while Instant::now() < deadline {
        if let Some(url) = crate::relay_endpoint::relay_endpoint_url() {
            return Ok(url);
        }
        tokio::time::sleep(Duration::from_millis(200)).await;
    }
    Err(
        "the relay endpoint is not answering; the machine daemon's `supercode message watch` \
         serves it (see mail/machine-daemon.log)"
            .into(),
    )
}

/// One `harness.v1` call into this machine's daemon (`supercode teams rpc`),
/// starting the daemon when it is not running.
pub async fn machine_rpc(program: &Path, method: &str, params: &Value) -> Result<Value, String> {
    let call = || {
        let mut command = tokio::process::Command::new(program);
        command
            .args(["teams", "rpc", method, &params.to_string()])
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::piped())
            .stderr(std::process::Stdio::piped());
        command.output()
    };
    let output = call().await.map_err(|error| error.to_string())?;
    if output.status.success() {
        return serde_json::from_slice(&output.stdout)
            .map_err(|error| format!("unreadable answer from the machine daemon: {error}"));
    }
    // Not running: start it (this OS user's own daemon, no Teams context
    // needed) and ask once more.
    ensure_machine_daemon(program).await?;
    let output = call().await.map_err(|error| error.to_string())?;
    if output.status.success() {
        return serde_json::from_slice(&output.stdout)
            .map_err(|error| format!("unreadable answer from the machine daemon: {error}"));
    }
    Err(error_line(&String::from_utf8_lossy(&output.stderr)))
}

/// Start this OS user's machine daemon when it is not answering. It hosts
/// relays (in its `harness serve`) and the idle watcher.
pub async fn ensure_machine_daemon(program: &Path) -> Result<(), String> {
    let describe = || {
        tokio::process::Command::new(program)
            .args(["teams", "describe"])
            .stdin(std::process::Stdio::null())
            .stdout(std::process::Stdio::null())
            .stderr(std::process::Stdio::null())
            .status()
    };
    if describe().await.is_ok_and(|status| status.success()) {
        return Ok(());
    }
    // A service manager runs this user's daemon (a connector, or the machine
    // unit): it is between restarts, and a daemon started here would take the
    // socket and lock the service's daemon out. Wait for the service instead.
    if crate::teams::service_owns_daemon() {
        for _ in 0..100 {
            tokio::time::sleep(Duration::from_millis(200)).await;
            if describe().await.is_ok_and(|status| status.success()) {
                return Ok(());
            }
        }
        return Err(
            "this machine's daemon is run by its service manager and did not answer within 20 seconds; \
             see `supercode teams status`"
                .into(),
        );
    }
    let root = mail_root();
    std::fs::create_dir_all(&root).map_err(|error| error.to_string())?;
    let log = std::fs::OpenOptions::new()
        .create(true)
        .append(true)
        .open(root.join("machine-daemon.log"))
        .map_err(|error| error.to_string())?;
    let mut command = std::process::Command::new(program);
    command
        .args(["teams", "machine", "start", "--supercode"])
        .arg(program)
        .stdin(std::process::Stdio::null())
        .stdout(log.try_clone().map_err(|error| error.to_string())?)
        .stderr(log);
    #[cfg(unix)]
    {
        use std::os::unix::process::CommandExt;
        // Its own process group, so the caller's exit does not take the
        // daemon (and every relay) with it.
        command.process_group(0);
    }
    command
        .spawn()
        .map_err(|error| format!("could not start the machine daemon: {error}"))?;
    for _ in 0..50 {
        tokio::time::sleep(Duration::from_millis(200)).await;
        if describe().await.is_ok_and(|status| status.success()) {
            return Ok(());
        }
    }
    Err(format!(
        "the machine daemon did not start within 10 seconds; see {}",
        root.join("machine-daemon.log").display()
    ))
}

/// The one line of a failed command's stderr an agent can act on.
fn error_line(stderr: &str) -> String {
    let lines: Vec<&str> = stderr
        .lines()
        .map(str::trim)
        .filter(|line| !line.is_empty())
        .collect();
    lines
        .iter()
        .find(|line| line.starts_with("Error") || line.starts_with("error"))
        .or(lines.last())
        .copied()
        .unwrap_or("the machine daemon failed without saying why")
        .chars()
        .take(300)
        .collect()
}

/// An exclusive lock on a relay's sends, released on drop.
struct SendLock {
    _file: std::fs::File,
}

impl SendLock {
    fn acquire(path: &Path) -> std::io::Result<Self> {
        let file = std::fs::OpenOptions::new()
            .create(true)
            .truncate(false)
            .write(true)
            .open(path)?;
        #[cfg(unix)]
        {
            use std::os::unix::io::AsRawFd;
            // SAFETY: flock on a descriptor this function owns; the lock is
            // released when the file is dropped.
            if unsafe { libc::flock(file.as_raw_fd(), libc::LOCK_EX) } != 0 {
                return Err(std::io::Error::last_os_error());
            }
        }
        Ok(Self { _file: file })
    }
}

/// File one inbound prompt in `represented`'s mailbox. `last_sent` maps a
/// Claude session name to the newest message this relay sent it, for
pub fn file_inbound_prompt(
    homes: &HarnessHomes,
    represented: &MailAddress,
    prompt: &str,
    last_sent: Option<&HashMap<String, String>>,
) {
    match parse_inbound_text(prompt) {
        Some(RelayEvent::Peer(native)) => {
            let registry = read_registry(&registry_dir(homes));
            let sender = resolve_native_sender(&registry, &native.from);
            let in_reply_to = sender
                .as_ref()
                .and_then(|session| last_sent.and_then(|sent| sent.get(&session.name).cloned()));
            file_inbound(represented, &native, sender.as_ref(), in_reply_to);
        }
        Some(RelayEvent::IdleNotice(text) | RelayEvent::DeliveryNotice(text)) => {
            file_notice(represented, &text);
        }
        _ => {}
    }
}
/// Recognize an inbound peer envelope or notice in the text Claude handed a
/// relay.
pub fn parse_inbound_text(text: &str) -> Option<RelayEvent> {
    let trimmed = text.trim_start();
    if trimmed.starts_with(NATIVE_IDLE_NOTICE) {
        return Some(RelayEvent::IdleNotice(trimmed.to_string()));
    }
    if trimmed.starts_with(NATIVE_DELIVERY_NOTICE) {
        return Some(RelayEvent::DeliveryNotice(trimmed.to_string()));
    }
    if !trimmed.starts_with(NATIVE_PEER_PREAMBLE) && !trimmed.starts_with(NATIVE_PEER_OPENING) {
        return None;
    }
    let start = text.find(NATIVE_PEER_OPENING)?;
    let header_end = start + text[start..].find(">\n")?;
    let header = &text[start + NATIVE_PEER_OPENING.len()..header_end];
    let body_start = header_end + 2;
    // The body is not escaped by Claude; the closing tag is the LAST one, and
    // the fixed paragraph after it never contains the tag.
    let body_end = text.rfind(&format!("\n{NATIVE_PEER_CLOSING}"))?;
    if body_end < body_start {
        return None;
    }
    let attributes = parse_attributes(header);
    Some(RelayEvent::Peer(NativeEnvelope {
        from: attributes.get("from")?.clone(),
        from_name: attributes.get("from-name").cloned(),
        body: text[body_start..body_end].to_string(),
    }))
}
fn parse_attributes(header: &str) -> BTreeMap<String, String> {
    let mut attributes = BTreeMap::new();
    let mut rest = header;
    while let Some(equals) = rest.find("=\"") {
        let key = rest[..equals].trim().to_string();
        let value_start = equals + 2;
        let Some(length) = rest[value_start..].find('"') else {
            break;
        };
        attributes.insert(key, rest[value_start..value_start + length].to_string());
        rest = &rest[value_start + length + 1..];
    }
    attributes
}
/// Deliver mail to the session a relay speaks for. When it lives on another
/// machine and Teams cannot take the message there, it is kept in this
/// machine's copy of that mailbox (never dropped) and the failure is written
fn deliver_home(represented: &MailAddress, envelope: &Envelope) {
    if let Err(error) = crate::mailbox::deliver_to(represented, envelope) {
        eprintln!(
            "could not deliver {} to {represented}: {error}; kept in this machine's mailbox for it",
            envelope.id
        );
        if let Ok(mailbox) = crate::mailbox::Mailbox::open(&mail_root(), represented) {
            mailbox.deliver(envelope).ok();
        }
        return;
    }
    // Filed; now hand it to its session by the session's own door, as any
    // sender's message is (a Claude session gets it at its next tool round).
    // That can take as long as a relay send, which this hook must not wait
    // on, so a detached `message push` does it.
    if represented.machine == crate::mailbox::local_machine_name() {
        if let Ok(program) = supercode_program() {
            std::process::Command::new(program)
                .args(["message", "push", &represented.to_string(), &envelope.id])
                .stdin(std::process::Stdio::null())
                .stdout(std::process::Stdio::null())
                .stderr(std::process::Stdio::null())
                .spawn()
                .ok();
        }
    }
}
fn file_inbound(
    represented: &MailAddress,
    native: &NativeEnvelope,
    sender: Option<&ClaudePeerSession>,
    in_reply_to: Option<String>,
) {
    // The replying Claude session runs beside the relay, on this machine —
    // not on the machine of the session the relay speaks for.
    let machine = crate::mailbox::local_machine_name();
    let (from, from_name) = match sender {
        Some(session) => (
            MailAddress::new(&machine, "claude-code", &session.session_id),
            format!("{}@{machine}", session.name),
        ),
        None => (
            MailAddress::new(&machine, "claude-code", "unknown"),
            native
                .from_name
                .clone()
                .unwrap_or_else(|| "an unknown Claude session".into()),
        ),
    };
    let Ok(from) = from else { return };
    let Ok(mut envelope) = Envelope::new(
        from,
        from_name,
        MailKind::Peer,
        ReplyVia::Command,
        native.body.clone(),
    ) else {
        return;
    };
    envelope.native_from = Some(native.from.clone());
    if let Some(in_reply_to) = in_reply_to {
        envelope.thread = crate::mailbox::thread_of_reply(Some(&in_reply_to));
        envelope.in_reply_to = Some(in_reply_to);
        envelope.in_reply_to_inferred = true;
    }
    // The session a relay speaks for may be on another machine; its mail goes
    // home through Teams.
    deliver_home(represented, &envelope);
}
fn file_notice(represented: &MailAddress, text: &str) {
    let Ok(from) = MailAddress::new(
        crate::mailbox::local_machine_name(),
        "claude-code",
        "notice",
    ) else {
        return;
    };
    let Ok(envelope) = Envelope::new(
        from,
        "Claude Code",
        MailKind::Notice,
        ReplyVia::None,
        text.to_string(),
    ) else {
        return;
    };
    // The session a relay speaks for may be on another machine; its mail goes
    // home through Teams.
    deliver_home(represented, &envelope);
}
/// The `supercode` program that the relay hooks and the machine daemon run:
/// this process when it is `supercode` (the CLI, `supercode harness serve`),
pub fn supercode_program() -> std::io::Result<PathBuf> {
    let current = std::env::current_exe()?;
    if current.file_stem().and_then(|stem| stem.to_str()) == Some("supercode") {
        return Ok(current);
    }
    std::env::var_os("PATH")
        .iter()
        .flat_map(std::env::split_paths)
        .map(|directory| directory.join("supercode"))
        .find(|candidate| candidate.is_file())
        .ok_or_else(|| {
            std::io::Error::new(
                std::io::ErrorKind::NotFound,
                "the supercode program is not on PATH; Claude relays and the machine daemon need it",
            )
        })
}