supercode_harness/
relay_endpoint.rs

1//! The Messages API a Claude relay talks to: deterministic, no model.
2//!
3//! A relay ([`crate::claude_relay`]) is a real Claude Code process, because
4//! only a live Claude peer's `SendMessage` gets a Claude session's reply back.
5//! What it asks its model is always the same: send the queued message. So a
6//! relay's `ANTHROPIC_BASE_URL` points here, and this endpoint answers as the
7//! model would, from the relay's own queue:
8//!
9//! - a send turn gets one `SendMessage` tool call carrying the queued `to`
10//!   and `message`, byte for byte;
11//! - the turn after a tool result, or with nothing queued, gets
12//!   [`RELAY_STATUS_LINE`] and ends.
13//!
14//! No model runs and nothing is billed. The gate hook still checks every tool
15//! call against the queue, and the receipt is still Claude's own
16//! `SendMessage` result. A relay's API key names its directory under
17//! `<mail root>/relays/`.
18//!
19//! The machine daemon's `supercode message watch` child serves it on
20//! loopback, on the port recorded in `<mail root>/relay-endpoint.json` (a
21//! restarted server binds the same port again, so running relays keep it).
22
23use std::path::PathBuf;
24use std::time::Duration;
25
26use serde::{Deserialize, Serialize};
27use serde_json::{json, Value};
28use tokio::io::BufReader;
29use tokio::net::TcpListener;
30
31use crate::claude_relay::{QueuedSend, RelayPaths, RELAY_STATUS_LINE};
32use crate::mailbox::mail_root;
33use crate::server::{read_http_request, write_http_response, HttpRequest};
34
35/// Where the serving port is recorded.
36fn record_path() -> PathBuf {
37    mail_root().join("relay-endpoint.json")
38}
39
40#[derive(Serialize, Deserialize)]
41struct Record {
42    port: u16,
43}
44
45/// The endpoint's base URL, when it is answering.
46pub fn relay_endpoint_url() -> Option<String> {
47    let record: Record = serde_json::from_slice(&std::fs::read(record_path()).ok()?).ok()?;
48    let address = std::net::SocketAddr::from(([127, 0, 0, 1], record.port));
49    std::net::TcpStream::connect_timeout(&address, Duration::from_millis(300)).ok()?;
50    Some(format!("http://127.0.0.1:{}", record.port))
51}
52
53/// Whether the relay endpoint at `url` (`http://127.0.0.1:<port>`) answers.
54pub fn endpoint_answers(url: &str) -> bool {
55    let Some(port) = url
56        .strip_prefix("http://127.0.0.1:")
57        .and_then(|port| port.trim_end_matches('/').parse::<u16>().ok())
58    else {
59        return false;
60    };
61    let address = std::net::SocketAddr::from(([127, 0, 0, 1], port));
62    std::net::TcpStream::connect_timeout(&address, Duration::from_millis(300)).is_ok()
63}
64
65fn write_record(port: u16) -> std::io::Result<()> {
66    let staging = record_path().with_extension("json.tmp");
67    std::fs::write(&staging, serde_json::to_vec(&Record { port })?)?;
68    std::fs::rename(&staging, record_path())
69}
70
71/// Serve the endpoint until the process ends.
72pub async fn serve_relay_endpoint() -> std::io::Result<()> {
73    let recorded = std::fs::read(record_path())
74        .ok()
75        .and_then(|bytes| serde_json::from_slice::<Record>(&bytes).ok())
76        .map(|record| record.port);
77    let listener = match recorded {
78        Some(port) => match TcpListener::bind(("127.0.0.1", port)).await {
79            Ok(listener) => listener,
80            Err(_) => TcpListener::bind(("127.0.0.1", 0)).await?,
81        },
82        None => TcpListener::bind(("127.0.0.1", 0)).await?,
83    };
84    let port = listener.local_addr()?.port();
85    std::fs::create_dir_all(mail_root())?;
86    write_record(port)?;
87    // The record is how senders find this endpoint. Another watcher (a daemon
88    // started for a moment by another command) can overwrite it and then exit,
89    // leaving it naming a dead port while this one listens unseen: keep it ours.
90    tokio::spawn(async move {
91        loop {
92            tokio::time::sleep(Duration::from_secs(5)).await;
93            let current = std::fs::read(record_path())
94                .ok()
95                .and_then(|bytes| serde_json::from_slice::<Record>(&bytes).ok())
96                .map(|record| record.port);
97            if current != Some(port) {
98                write_record(port).ok();
99            }
100        }
101    });
102    loop {
103        let (stream, _) = listener.accept().await?;
104        tokio::spawn(async move {
105            let (read, mut write) = stream.into_split();
106            let mut reader = BufReader::new(read);
107            let Ok(Some(request)) = read_http_request(&mut reader).await else {
108                return;
109            };
110            let (status, content_type, body) = answer(&request);
111            let reason = if status == 200 { "OK" } else { "Error" };
112            write_http_response(&mut write, status, reason, content_type, &body)
113                .await
114                .ok();
115        });
116    }
117}
118
119/// The relay directory an API key names, if it is one.
120fn relay_paths(request: &HttpRequest) -> Option<RelayPaths> {
121    let key = request.headers.get("x-api-key")?;
122    if key.is_empty() || !key.chars().all(|c| c.is_ascii_hexdigit()) {
123        return None;
124    }
125    let directory = mail_root().join("relays").join(key);
126    directory
127        .is_dir()
128        .then(|| RelayPaths::in_directory(directory))
129}
130
131fn answer(request: &HttpRequest) -> (u16, &'static str, Vec<u8>) {
132    let error = |status: u16, message: &str| {
133        let body = json!({
134            "type": "error",
135            "error": {"type": "invalid_request_error", "message": message},
136        });
137        (status, "application/json", body.to_string().into_bytes())
138    };
139    if request.method != "POST" {
140        return error(404, "not found");
141    }
142    let Some(paths) = relay_paths(request) else {
143        return error(401, "not a supercode relay");
144    };
145    if request.path.ends_with("/v1/messages/count_tokens") {
146        return (
147            200,
148            "application/json",
149            json!({"input_tokens": 1}).to_string().into_bytes(),
150        );
151    }
152    if !request.path.ends_with("/v1/messages") {
153        return error(404, "not found");
154    }
155    let Ok(body) = serde_json::from_slice::<Value>(&request.body) else {
156        return error(400, "unreadable request");
157    };
158    let queued = std::fs::read(&paths.queue)
159        .ok()
160        .and_then(|bytes| serde_json::from_slice::<QueuedSend>(&bytes).ok());
161    let content = match queued {
162        Some(send) if !after_tool_result(&body) => json!({
163            "type": "tool_use",
164            "id": format!("toolu_{}", crate::mailbox::new_message_id().unwrap_or_default()),
165            "name": "SendMessage",
166            "input": {"to": send.to, "message": send.message},
167        }),
168        _ => json!({"type": "text", "text": RELAY_STATUS_LINE}),
169    };
170    let model = body.get("model").cloned().unwrap_or(Value::Null);
171    if body.get("stream").and_then(Value::as_bool) == Some(true) {
172        (
173            200,
174            "text/event-stream",
175            stream(&model, &content).into_bytes(),
176        )
177    } else {
178        (
179            200,
180            "application/json",
181            message(&model, &content).to_string().into_bytes(),
182        )
183    }
184}
185
186/// Whether the conversation's last turn carries a tool result: the send
187/// happened (or was denied), and the turn should end.
188fn after_tool_result(body: &Value) -> bool {
189    body.get("messages")
190        .and_then(Value::as_array)
191        .and_then(|messages| messages.last())
192        .and_then(|last| last.get("content"))
193        .and_then(Value::as_array)
194        .is_some_and(|content| {
195            content
196                .iter()
197                .any(|block| block.get("type").and_then(Value::as_str) == Some("tool_result"))
198        })
199}
200
201fn stop_reason(content: &Value) -> &'static str {
202    if content["type"] == "tool_use" {
203        "tool_use"
204    } else {
205        "end_turn"
206    }
207}
208
209fn message(model: &Value, content: &Value) -> Value {
210    json!({
211        "id": format!("msg_{}", crate::mailbox::new_message_id().unwrap_or_default()),
212        "type": "message",
213        "role": "assistant",
214        "model": model,
215        "content": [content],
216        "stop_reason": stop_reason(content),
217        "stop_sequence": null,
218        "usage": {"input_tokens": 1, "output_tokens": 1},
219    })
220}
221
222/// The same message as the server-sent events a streaming request expects.
223fn stream(model: &Value, content: &Value) -> String {
224    let mut start = message(model, content);
225    start["content"] = json!([]);
226    start["stop_reason"] = Value::Null;
227    let (opening, delta) = if content["type"] == "tool_use" {
228        let mut opening = content.clone();
229        opening["input"] = json!({});
230        let delta =
231            json!({"type": "input_json_delta", "partial_json": content["input"].to_string()});
232        (opening, delta)
233    } else {
234        let delta = json!({"type": "text_delta", "text": content["text"]});
235        (json!({"type": "text", "text": ""}), delta)
236    };
237    let events = [
238        (
239            "message_start",
240            json!({"type": "message_start", "message": start}),
241        ),
242        (
243            "content_block_start",
244            json!({"type": "content_block_start", "index": 0, "content_block": opening}),
245        ),
246        (
247            "content_block_delta",
248            json!({"type": "content_block_delta", "index": 0, "delta": delta}),
249        ),
250        (
251            "content_block_stop",
252            json!({"type": "content_block_stop", "index": 0}),
253        ),
254        (
255            "message_delta",
256            json!({
257                "type": "message_delta",
258                "delta": {"stop_reason": stop_reason(content), "stop_sequence": null},
259                "usage": {"output_tokens": 1},
260            }),
261        ),
262        ("message_stop", json!({"type": "message_stop"})),
263    ];
264    events
265        .iter()
266        .map(|(name, data)| format!("event: {name}\ndata: {data}\n\n"))
267        .collect()
268}
supercode_harness/relay_endpoint.rs

supercode_harness/
relay_endpoint.rs
