supercode_harness/

relay_endpoint.rs

//! The Messages API a Claude relay talks to: deterministic, no model.
//!
//! A relay ([`crate::claude_relay`]) is a real Claude Code process, because
//! only a live Claude peer's `SendMessage` gets a Claude session's reply back.
//! What it asks its model is always the same: send the queued message. So a
//! relay's `ANTHROPIC_BASE_URL` points here, and this endpoint answers as the
//! model would, from the relay's own queue:
//!
//! - a send turn gets one `SendMessage` tool call carrying the queued `to`
//!   and `message`, byte for byte;
//! - the turn after a tool result, or with nothing queued, gets
//!   [`RELAY_STATUS_LINE`] and ends.
//!
//! No model runs and nothing is billed. The gate hook still checks every tool
//! call against the queue, and the receipt is still Claude's own
//! `SendMessage` result. A relay's API key names its directory under
//! `<mail root>/relays/`.
//!
//! The machine daemon's `supercode message watch` child serves it on
//! loopback, on the port recorded in `<mail root>/relay-endpoint.json` (a
//! restarted server binds the same port again, so running relays keep it).

use std::path::PathBuf;
use std::time::Duration;

use serde::{Deserialize, Serialize};
use serde_json::{json, Value};
use tokio::io::BufReader;
use tokio::net::TcpListener;

use crate::claude_relay::{QueuedSend, RelayPaths, RELAY_STATUS_LINE};
use crate::mailbox::mail_root;
use crate::server::{read_http_request, write_http_response, HttpRequest};

/// Where the serving port is recorded.
fn record_path() -> PathBuf {
    mail_root().join("relay-endpoint.json")
}

#[derive(Serialize, Deserialize)]
struct Record {
    port: u16,
}

/// The endpoint's base URL, when it is answering.
pub fn relay_endpoint_url() -> Option<String> {
    let record: Record = serde_json::from_slice(&std::fs::read(record_path()).ok()?).ok()?;
    let address = std::net::SocketAddr::from(([127, 0, 0, 1], record.port));
    std::net::TcpStream::connect_timeout(&address, Duration::from_millis(300)).ok()?;
    Some(format!("http://127.0.0.1:{}", record.port))
}

fn write_record(port: u16) -> std::io::Result<()> {
    let staging = record_path().with_extension("json.tmp");
    std::fs::write(&staging, serde_json::to_vec(&Record { port })?)?;
    std::fs::rename(&staging, record_path())
}

/// Serve the endpoint until the process ends.
pub async fn serve_relay_endpoint() -> std::io::Result<()> {
    let recorded = std::fs::read(record_path())
        .ok()
        .and_then(|bytes| serde_json::from_slice::<Record>(&bytes).ok())
        .map(|record| record.port);
    let listener = match recorded {
        Some(port) => match TcpListener::bind(("127.0.0.1", port)).await {
            Ok(listener) => listener,
            Err(_) => TcpListener::bind(("127.0.0.1", 0)).await?,
        },
        None => TcpListener::bind(("127.0.0.1", 0)).await?,
    };
    let port = listener.local_addr()?.port();
    std::fs::create_dir_all(mail_root())?;
    write_record(port)?;
    // The record is how senders find this endpoint. Another watcher (a daemon
    // started for a moment by another command) can overwrite it and then exit,
    // leaving it naming a dead port while this one listens unseen: keep it ours.
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(Duration::from_secs(5)).await;
            let current = std::fs::read(record_path())
                .ok()
                .and_then(|bytes| serde_json::from_slice::<Record>(&bytes).ok())
                .map(|record| record.port);
            if current != Some(port) {
                write_record(port).ok();
            }
        }
    });
    loop {
        let (stream, _) = listener.accept().await?;
        tokio::spawn(async move {
            let (read, mut write) = stream.into_split();
            let mut reader = BufReader::new(read);
            let Ok(Some(request)) = read_http_request(&mut reader).await else {
                return;
            };
            let (status, content_type, body) = answer(&request);
            let reason = if status == 200 { "OK" } else { "Error" };
            write_http_response(&mut write, status, reason, content_type, &body)
                .await
                .ok();
        });
    }
}

/// The relay directory an API key names, if it is one.
fn relay_paths(request: &HttpRequest) -> Option<RelayPaths> {
    let key = request.headers.get("x-api-key")?;
    if key.is_empty() || !key.chars().all(|c| c.is_ascii_hexdigit()) {
        return None;
    }
    let directory = mail_root().join("relays").join(key);
    directory
        .is_dir()
        .then(|| RelayPaths::in_directory(directory))
}

fn answer(request: &HttpRequest) -> (u16, &'static str, Vec<u8>) {
    let error = |status: u16, message: &str| {
        let body = json!({
            "type": "error",
            "error": {"type": "invalid_request_error", "message": message},
        });
        (status, "application/json", body.to_string().into_bytes())
    };
    if request.method != "POST" {
        return error(404, "not found");
    }
    let Some(paths) = relay_paths(request) else {
        return error(401, "not a supercode relay");
    };
    if request.path.ends_with("/v1/messages/count_tokens") {
        return (
            200,
            "application/json",
            json!({"input_tokens": 1}).to_string().into_bytes(),
        );
    }
    if !request.path.ends_with("/v1/messages") {
        return error(404, "not found");
    }
    let Ok(body) = serde_json::from_slice::<Value>(&request.body) else {
        return error(400, "unreadable request");
    };
    let queued = std::fs::read(&paths.queue)
        .ok()
        .and_then(|bytes| serde_json::from_slice::<QueuedSend>(&bytes).ok());
    let content = match queued {
        Some(send) if !after_tool_result(&body) => json!({
            "type": "tool_use",
            "id": format!("toolu_{}", crate::mailbox::new_message_id().unwrap_or_default()),
            "name": "SendMessage",
            "input": {"to": send.to, "message": send.message},
        }),
        _ => json!({"type": "text", "text": RELAY_STATUS_LINE}),
    };
    let model = body.get("model").cloned().unwrap_or(Value::Null);
    if body.get("stream").and_then(Value::as_bool) == Some(true) {
        (
            200,
            "text/event-stream",
            stream(&model, &content).into_bytes(),
        )
    } else {
        (
            200,
            "application/json",
            message(&model, &content).to_string().into_bytes(),
        )
    }
}

/// Whether the conversation's last turn carries a tool result: the send
/// happened (or was denied), and the turn should end.
fn after_tool_result(body: &Value) -> bool {
    body.get("messages")
        .and_then(Value::as_array)
        .and_then(|messages| messages.last())
        .and_then(|last| last.get("content"))
        .and_then(Value::as_array)
        .is_some_and(|content| {
            content
                .iter()
                .any(|block| block.get("type").and_then(Value::as_str) == Some("tool_result"))
        })
}

fn stop_reason(content: &Value) -> &'static str {
    if content["type"] == "tool_use" {
        "tool_use"
    } else {
        "end_turn"
    }
}

fn message(model: &Value, content: &Value) -> Value {
    json!({
        "id": format!("msg_{}", crate::mailbox::new_message_id().unwrap_or_default()),
        "type": "message",
        "role": "assistant",
        "model": model,
        "content": [content],
        "stop_reason": stop_reason(content),
        "stop_sequence": null,
        "usage": {"input_tokens": 1, "output_tokens": 1},
    })
}

/// The same message as the server-sent events a streaming request expects.
fn stream(model: &Value, content: &Value) -> String {
    let mut start = message(model, content);
    start["content"] = json!([]);
    start["stop_reason"] = Value::Null;
    let (opening, delta) = if content["type"] == "tool_use" {
        let mut opening = content.clone();
        opening["input"] = json!({});
        let delta =
            json!({"type": "input_json_delta", "partial_json": content["input"].to_string()});
        (opening, delta)
    } else {
        let delta = json!({"type": "text_delta", "text": content["text"]});
        (json!({"type": "text", "text": ""}), delta)
    };
    let events = [
        (
            "message_start",
            json!({"type": "message_start", "message": start}),
        ),
        (
            "content_block_start",
            json!({"type": "content_block_start", "index": 0, "content_block": opening}),
        ),
        (
            "content_block_delta",
            json!({"type": "content_block_delta", "index": 0, "delta": delta}),
        ),
        (
            "content_block_stop",
            json!({"type": "content_block_stop", "index": 0}),
        ),
        (
            "message_delta",
            json!({
                "type": "message_delta",
                "delta": {"stop_reason": stop_reason(content), "stop_sequence": null},
                "usage": {"output_tokens": 1},
            }),
        ),
        ("message_stop", json!({"type": "message_stop"})),
    ];
    events
        .iter()
        .map(|(name, data)| format!("event: {name}\ndata: {data}\n\n"))
        .collect()
}