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/// Serve the endpoint until the process ends.
54pub async fn serve_relay_endpoint() -> std::io::Result<()> {
55    let recorded = std::fs::read(record_path())
56        .ok()
57        .and_then(|bytes| serde_json::from_slice::<Record>(&bytes).ok())
58        .map(|record| record.port);
59    let listener = match recorded {
60        Some(port) => match TcpListener::bind(("127.0.0.1", port)).await {
61            Ok(listener) => listener,
62            Err(_) => TcpListener::bind(("127.0.0.1", 0)).await?,
63        },
64        None => TcpListener::bind(("127.0.0.1", 0)).await?,
65    };
66    let port = listener.local_addr()?.port();
67    std::fs::create_dir_all(mail_root())?;
68    let staging = record_path().with_extension("json.tmp");
69    std::fs::write(&staging, serde_json::to_vec(&Record { port })?)?;
70    std::fs::rename(&staging, record_path())?;
71    loop {
72        let (stream, _) = listener.accept().await?;
73        tokio::spawn(async move {
74            let (read, mut write) = stream.into_split();
75            let mut reader = BufReader::new(read);
76            let Ok(Some(request)) = read_http_request(&mut reader).await else {
77                return;
78            };
79            let (status, content_type, body) = answer(&request);
80            let reason = if status == 200 { "OK" } else { "Error" };
81            write_http_response(&mut write, status, reason, content_type, &body)
82                .await
83                .ok();
84        });
85    }
86}
87
88/// The relay directory an API key names, if it is one.
89fn relay_paths(request: &HttpRequest) -> Option<RelayPaths> {
90    let key = request.headers.get("x-api-key")?;
91    if key.is_empty() || !key.chars().all(|c| c.is_ascii_hexdigit()) {
92        return None;
93    }
94    let directory = mail_root().join("relays").join(key);
95    directory
96        .is_dir()
97        .then(|| RelayPaths::in_directory(directory))
98}
99
100fn answer(request: &HttpRequest) -> (u16, &'static str, Vec<u8>) {
101    let error = |status: u16, message: &str| {
102        let body = json!({
103            "type": "error",
104            "error": {"type": "invalid_request_error", "message": message},
105        });
106        (status, "application/json", body.to_string().into_bytes())
107    };
108    if request.method != "POST" {
109        return error(404, "not found");
110    }
111    let Some(paths) = relay_paths(request) else {
112        return error(401, "not a supercode relay");
113    };
114    if request.path.ends_with("/v1/messages/count_tokens") {
115        return (
116            200,
117            "application/json",
118            json!({"input_tokens": 1}).to_string().into_bytes(),
119        );
120    }
121    if !request.path.ends_with("/v1/messages") {
122        return error(404, "not found");
123    }
124    let Ok(body) = serde_json::from_slice::<Value>(&request.body) else {
125        return error(400, "unreadable request");
126    };
127    let queued = std::fs::read(&paths.queue)
128        .ok()
129        .and_then(|bytes| serde_json::from_slice::<QueuedSend>(&bytes).ok());
130    let content = match queued {
131        Some(send) if !after_tool_result(&body) => json!({
132            "type": "tool_use",
133            "id": format!("toolu_{}", crate::mailbox::new_message_id().unwrap_or_default()),
134            "name": "SendMessage",
135            "input": {"to": send.to, "message": send.message},
136        }),
137        _ => json!({"type": "text", "text": RELAY_STATUS_LINE}),
138    };
139    let model = body.get("model").cloned().unwrap_or(Value::Null);
140    if body.get("stream").and_then(Value::as_bool) == Some(true) {
141        (
142            200,
143            "text/event-stream",
144            stream(&model, &content).into_bytes(),
145        )
146    } else {
147        (
148            200,
149            "application/json",
150            message(&model, &content).to_string().into_bytes(),
151        )
152    }
153}
154
155/// Whether the conversation's last turn carries a tool result: the send
156/// happened (or was denied), and the turn should end.
157fn after_tool_result(body: &Value) -> bool {
158    body.get("messages")
159        .and_then(Value::as_array)
160        .and_then(|messages| messages.last())
161        .and_then(|last| last.get("content"))
162        .and_then(Value::as_array)
163        .is_some_and(|content| {
164            content
165                .iter()
166                .any(|block| block.get("type").and_then(Value::as_str) == Some("tool_result"))
167        })
168}
169
170fn stop_reason(content: &Value) -> &'static str {
171    if content["type"] == "tool_use" {
172        "tool_use"
173    } else {
174        "end_turn"
175    }
176}
177
178fn message(model: &Value, content: &Value) -> Value {
179    json!({
180        "id": format!("msg_{}", crate::mailbox::new_message_id().unwrap_or_default()),
181        "type": "message",
182        "role": "assistant",
183        "model": model,
184        "content": [content],
185        "stop_reason": stop_reason(content),
186        "stop_sequence": null,
187        "usage": {"input_tokens": 1, "output_tokens": 1},
188    })
189}
190
191/// The same message as the server-sent events a streaming request expects.
192fn stream(model: &Value, content: &Value) -> String {
193    let mut start = message(model, content);
194    start["content"] = json!([]);
195    start["stop_reason"] = Value::Null;
196    let (opening, delta) = if content["type"] == "tool_use" {
197        let mut opening = content.clone();
198        opening["input"] = json!({});
199        let delta =
200            json!({"type": "input_json_delta", "partial_json": content["input"].to_string()});
201        (opening, delta)
202    } else {
203        let delta = json!({"type": "text_delta", "text": content["text"]});
204        (json!({"type": "text", "text": ""}), delta)
205    };
206    let events = [
207        (
208            "message_start",
209            json!({"type": "message_start", "message": start}),
210        ),
211        (
212            "content_block_start",
213            json!({"type": "content_block_start", "index": 0, "content_block": opening}),
214        ),
215        (
216            "content_block_delta",
217            json!({"type": "content_block_delta", "index": 0, "delta": delta}),
218        ),
219        (
220            "content_block_stop",
221            json!({"type": "content_block_stop", "index": 0}),
222        ),
223        (
224            "message_delta",
225            json!({
226                "type": "message_delta",
227                "delta": {"stop_reason": stop_reason(content), "stop_sequence": null},
228                "usage": {"output_tokens": 1},
229            }),
230        ),
231        ("message_stop", json!({"type": "message_stop"})),
232    ];
233    events
234        .iter()
235        .map(|(name, data)| format!("event: {name}\ndata: {data}\n\n"))
236        .collect()
237}
supercode_harness/relay_endpoint.rs

supercode_harness/
relay_endpoint.rs
