B4a claude-code-backend: Claude backend driving the unmodified claude binary on the operator's subscription; tool-less turns, per-turn usage, process-group kill (squash of eureka-body/claude-code-backend e0c0832..341be82; Soul s1, s2)

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
Eureka 2026-10-29 11:03:03 +01:00
commit 200933f7cf
11 changed files with 3158 additions and 2 deletions

14
Cargo.lock generated
View file

@ -134,6 +134,20 @@ dependencies = [
"inout",
]
[[package]]
name = "claude-connector"
version = "0.1.0"
dependencies = [
"libc",
"model-connector",
"rmp-serde",
"serde",
"serde_json",
"sha2",
"thiserror",
"zeroize",
]
[[package]]
name = "const-oid"
version = "0.9.6"

View file

@ -4,8 +4,8 @@ The connector kit for GameCult minds: one typed, encrypted, replay-safe contract
between a caller that needs a model turn and a daemon that holds the provider
credential, so no project wires its own provider calls.
Status: the `model-connector` crate is live. No connector binary exists yet
(the Claude connector is first, then an OpenAI-compatible one).
Status: the `model-connector` crate and the `claude-connector` backend crate are live. No
connector binary exists yet (the Claude connector binary is first, then an OpenAI-compatible one).
## Entry points
@ -32,6 +32,19 @@ Crate `crates/model-connector`:
`ProviderBackend` trait. A connector binary implements `ProviderBackend` for its
provider and calls `serve`; the kit names no provider.
Crate `crates/claude-connector` (library only; the binary that wires it to `serve` is not written):
- `ClaudeCodeBackend` implements `ProviderBackend` for provider `claude-code` by driving the
unmodified `claude` binary, one child per call, in print stream-json mode with no tools, no MCP
servers, no settings sources and the system prompt replaced. Tool-less turns only: a request
with tools, an output schema or a final tool result is refused before any spawn.
- Reasoning state is a session handle (`ClaudeCodeHandle`); a handle that cannot be proven to
match the request re-seeds from the caller's transcript, and every resume forks.
- `ProcessPort` is the one OS seam (`StdProcessPort`, unix, kills the child's whole process
group); `SessionStore` and `StderrSink` are the other ports. `ClaudeCodeConfig` is what the
binary hands in; the credential is read by name from the connector's own environment
(`CLAUDE_CODE_OAUTH_TOKEN`) and copied only into the child.
Frame arithmetic is public in `src/envelope.rs`, measured on the carriage frame: `frame_bytes_for_payload(n)` is the frame a
caller payload bound of `n` needs, `payload_bytes_for_frame(frame)` is the largest payload bound
a frame allows, and `MAX_IDENTITY_BYTES` (256) bounds caller ids, request ids and model names.

View file

@ -0,0 +1,20 @@
[package]
name = "claude-connector"
description = "Claude backend for the connector kit: drives the unmodified claude binary in print stream-json mode"
version.workspace = true
edition.workspace = true
license.workspace = true
repository.workspace = true
publish = false
[dependencies]
model-connector = { path = "../model-connector", features = ["daemon"] }
rmp-serde = "1"
serde = { version = "1", features = ["derive"] }
serde_json = "1"
sha2 = "0.10"
thiserror = "2"
zeroize = "1"
[target.'cfg(unix)'.dependencies]
libc = "0.2"

View file

@ -0,0 +1,263 @@
use std::fs::File;
use std::io;
use std::time::Instant;
use model_connector::{
FailureKind, InputItem, ProviderBackend, ProviderCall, ProviderRequest, TransportEvent,
TransportEventPayload, TransportOutcome, TransportResult, Usage,
};
use sha2::{Digest, Sha256};
use zeroize::Zeroizing;
use crate::config::{ClaudeCodeConfig, PROVIDER};
use crate::handle::{ClaudeCodeHandle, SessionStore, plan, prefix_digest, stdin_bytes};
use crate::launch::{PromptFile, TOKEN_VAR, argv, child_env};
use crate::process::{ProcessPort, RunningChild, StdoutRead};
use crate::stream::{Failure, Finished, Step, StreamState};
/// How the backend reads its own environment.
type EnvLookup = Box<dyn Fn(&str) -> Option<String> + Send + Sync>;
/// Where the tail of a child's stderr goes. The backend never puts it in a
/// result; the connector binary wires this to its log.
pub trait StderrSink: Send + Sync {
fn stderr_tail(&self, tail: &[u8]);
}
/// Answers each call with one `claude` child process: never more than one per
/// call, and a failed call is never run again.
pub struct ClaudeCodeBackend {
config: ClaudeCodeConfig,
port: Box<dyn ProcessPort>,
sessions: Box<dyn SessionStore>,
sink: Box<dyn StderrSink>,
env_lookup: EnvLookup,
}
impl ClaudeCodeBackend {
/// The backend over real processes and the CLI's real session store.
#[cfg(unix)]
pub fn new(config: ClaudeCodeConfig, sink: Box<dyn StderrSink>) -> Self {
let port = Box::new(crate::process::StdProcessPort::new(config.limits));
let sessions = Box::new(crate::handle::FsSessionStore::new(
config.config_dir.clone(),
));
Self::with_ports(config, port, sessions, sink)
}
pub fn with_ports(
config: ClaudeCodeConfig,
port: Box<dyn ProcessPort>,
sessions: Box<dyn SessionStore>,
sink: Box<dyn StderrSink>,
) -> Self {
Self {
config,
port,
sessions,
sink,
env_lookup: Box::new(|name| std::env::var(name).ok()),
}
}
fn token(&self) -> Option<Zeroizing<String>> {
(self.env_lookup)(TOKEN_VAR)
.filter(|value| !value.is_empty())
.map(Zeroizing::new)
}
fn binary_matches_pin(&self) -> bool {
let Ok(mut file) = File::open(&self.config.claude_binary) else {
return false;
};
let mut hasher = Sha256::new();
io::copy(&mut file, &mut hasher).is_ok()
&& <[u8; 32]>::from(hasher.finalize()) == self.config.claude_binary_sha256
}
fn run(&self, request: &ProviderRequest) -> (Vec<TransportEvent>, TransportOutcome) {
let plan = match plan(request, &self.config, self.sessions.as_ref()) {
Ok(plan) => plan,
Err(error) => {
return failed(
FailureKind::InvalidRequest,
error.to_string(),
Some(zero_usage()),
);
}
};
let Some(token) = self.token() else {
return failed(
FailureKind::Authentication,
"claude-code: the connector has no credential to give the child".to_string(),
Some(zero_usage()),
);
};
let Ok(prompt) = PromptFile::create(&request.instructions) else {
return failure(Failure::unbilled(
FailureKind::Provider,
"claude-code: the system prompt could not be staged",
));
};
let argv = argv(&self.config, request, &plan, prompt.path());
let env = child_env(&self.config, request, token);
let deadline = Instant::now() + self.config.limits.deadline;
let spawned = self.port.spawn(
&argv,
&env,
&self.config.work_dir,
stdin_bytes(plan.blocks()),
);
let Ok(mut child) = spawned else {
return failure(Failure::unbilled(
FailureKind::Provider,
"claude-code: the child could not be spawned",
));
};
let finished = read_stream(child.as_mut(), deadline, &request.model);
child.kill();
child.wait();
let tail = child.stderr_tail();
if !tail.is_empty() {
self.sink.stderr_tail(&tail);
}
match finished {
Ok(finished) => self.completed(request, finished),
Err(failure_) => failure(failure_),
}
}
fn completed(
&self,
request: &ProviderRequest,
finished: Finished,
) -> (Vec<TransportEvent>, TransportOutcome) {
let mut payloads: Vec<TransportEventPayload> = finished
.texts
.iter()
.map(|text| TransportEventPayload::TextDelta { text: text.clone() })
.collect();
let mut prefix = request.input.clone();
if !finished.texts.is_empty() {
prefix.push(InputItem::AssistantText {
text: finished.texts.concat(),
});
}
let handle = ClaudeCodeHandle::Ended {
session_id: finished.session_id,
instance: self.config.instance.clone(),
prefix_sha256: prefix_digest(&request.instructions, &request.model, &prefix),
};
payloads.push(TransportEventPayload::ReasoningState {
state: handle.encode(),
});
let events = payloads
.into_iter()
.enumerate()
.map(|(sequence, payload)| TransportEvent {
sequence: sequence as u64,
payload,
})
.collect();
(
events,
TransportOutcome::Completed {
provider_response_id: finished.message_id,
stop_reason: finished.stop_reason,
usage: finished.usage,
},
)
}
}
impl ProviderBackend for ClaudeCodeBackend {
fn ready(&self) -> bool {
self.config.work_dir.is_dir()
&& self.config.config_dir.is_dir()
&& self.token().is_some()
&& self.binary_matches_pin()
}
fn provider(&self) -> &str {
PROVIDER
}
fn execute(&self, call: &ProviderCall) -> TransportResult {
let (events, outcome) = self.run(call.request());
call.transported(events, outcome)
}
}
fn read_stream(
child: &mut dyn RunningChild,
deadline: Instant,
model: &str,
) -> Result<Finished, Failure> {
let mut state = StreamState::new(model);
let finished = loop {
match child.read_line(deadline) {
StdoutRead::Line(line) => {
if let Step::Done(done) = state.feed(&line) {
break *done;
}
}
StdoutRead::Eof => {
return Err(Failure::unbilled(
FailureKind::Provider,
"claude-code: the stream ended without a result",
));
}
StdoutRead::Timeout => {
return Err(Failure::unbilled(
FailureKind::Timeout,
"claude-code: the deadline passed before a result",
));
}
StdoutRead::Breach => {
return Err(Failure::unbilled(
FailureKind::Provider,
"claude-code: the output passed its size cap",
));
}
}
};
// The result is in; give the child until its deadline to close stdout.
while matches!(child.read_line(deadline), StdoutRead::Line(_)) {}
finished
}
fn zero_usage() -> Usage {
Usage {
input_tokens: 0,
cache_read_tokens: 0,
cache_write_tokens: 0,
output_tokens: 0,
reasoning_tokens: None,
}
}
fn failed(
kind: FailureKind,
message: String,
usage: Option<Usage>,
) -> (Vec<TransportEvent>, TransportOutcome) {
(
Vec::new(),
TransportOutcome::Failed {
kind,
message,
usage,
},
)
}
fn failure(failure: Failure) -> (Vec<TransportEvent>, TransportOutcome) {
failed(failure.kind, failure.message.to_string(), failure.usage)
}
#[cfg(test)]
mod tests;

File diff suppressed because it is too large Load diff

View file

@ -0,0 +1,35 @@
use std::path::PathBuf;
use std::time::Duration;
/// The provider tag the kit puts on every receipt and on the reasoning state
/// of this backend. It is distinct from any Messages-API backend's tag, so the
/// kit treats each backend's state as foreign to the other.
pub(crate) const PROVIDER: &str = "claude-code";
/// What the connector's binary hands the backend. Every path is absolute.
#[derive(Debug, Clone)]
pub struct ClaudeCodeConfig {
/// The `claude` executable.
pub claude_binary: PathBuf,
/// The SHA-256 the executable must have for [`ready`](model_connector::ProviderBackend::ready).
pub claude_binary_sha256: [u8; 32],
/// An empty directory the connector owns: the CLI writes its cwd, git state
/// and platform into the system prompt.
pub work_dir: PathBuf,
/// Passed to the child as `CLAUDE_CONFIG_DIR` and `HOME`; it holds the
/// CLI's session store.
pub config_dir: PathBuf,
/// The connector instance id written into every handle.
pub instance: String,
pub limits: ClaudeCodeLimits,
}
/// The bounds of one child process.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct ClaudeCodeLimits {
/// The whole call, spawn to result.
pub deadline: Duration,
pub max_line_bytes: usize,
pub max_total_bytes: usize,
pub stderr_tail_bytes: usize,
}

View file

@ -0,0 +1,266 @@
//! The reasoning state of this backend and the decision it drives: resume the
//! CLI session a handle names, or start a fresh one seeded with the caller's
//! transcript. The session is a cache of the caller's transcript; the
//! transcript is the truth.
use std::path::PathBuf;
use model_connector::{InputItem, ProviderRequest};
use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256};
use crate::config::{ClaudeCodeConfig, PROVIDER};
/// The state this backend hands a caller inside `ReasoningState`. A new
/// variant is a new version.
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
pub(crate) enum ClaudeCodeHandle {
/// The session a finished turn left behind. `prefix_sha256` is the digest
/// of the instructions, the model and every transcript item that precedes
/// this handle.
Ended {
session_id: String,
instance: String,
prefix_sha256: [u8; 32],
},
}
impl ClaudeCodeHandle {
pub(crate) fn encode(&self) -> Vec<u8> {
rmp_serde::to_vec_named(self).expect("a handle of strings and bytes encodes")
}
fn decode(state: &[u8]) -> Option<Self> {
rmp_serde::from_slice(state).ok()
}
}
/// The digest a handle binds: named MessagePack of the instructions, the model
/// and the transcript items before the handle.
pub(crate) fn prefix_digest(instructions: &str, model: &str, items: &[InputItem]) -> [u8; 32] {
let bytes =
rmp_serde::to_vec_named(&(instructions, model, items)).expect("strings and bytes encode");
Sha256::digest(bytes).into()
}
/// Whether the CLI's session store still holds a session.
pub trait SessionStore: Send + Sync {
fn exists(&self, session_id: &str) -> bool;
}
/// Looks for `<config_dir>/projects/*/<session_id>.jsonl`. A session id that
/// is not a canonical UUID never names a path.
pub struct FsSessionStore {
config_dir: PathBuf,
}
impl FsSessionStore {
pub fn new(config_dir: PathBuf) -> Self {
Self { config_dir }
}
}
fn is_canonical_uuid(id: &str) -> bool {
id.len() == 36
&& id.char_indices().all(|(at, c)| match at {
8 | 13 | 18 | 23 => c == '-',
_ => c.is_ascii_digit() || ('a'..='f').contains(&c),
})
}
impl SessionStore for FsSessionStore {
fn exists(&self, session_id: &str) -> bool {
if !is_canonical_uuid(session_id) {
return false;
}
let Ok(projects) = std::fs::read_dir(self.config_dir.join("projects")) else {
return false;
};
projects
.flatten()
.any(|project| project.path().join(format!("{session_id}.jsonl")).is_file())
}
}
/// Why a request cannot be answered by this backend; the text is the failure
/// message and names what lifts it.
#[derive(Debug, thiserror::Error)]
pub(crate) enum PlanError {
#[error("claude-code: tools are not supported by this backend yet")]
Tools,
#[error("claude-code: structured output is not supported by this backend yet")]
OutputSchema,
#[error("claude-code: tool results are not supported by this backend yet")]
ToolResult,
#[error("claude-code: the request ends without user text")]
NoUserText,
#[error("claude-code: the request carries a reasoning state this backend cannot read")]
UndecodableHandle,
}
/// How a call reaches the CLI. `blocks` are the text blocks of the one user
/// message the child is given.
#[derive(Debug, PartialEq, Eq)]
pub(crate) enum Plan {
Resume {
session_id: String,
blocks: Vec<String>,
},
Seed {
blocks: Vec<String>,
},
}
impl Plan {
pub(crate) fn blocks(&self) -> &[String] {
match self {
Plan::Resume { blocks, .. } | Plan::Seed { blocks } => blocks,
}
}
}
pub(crate) fn plan(
request: &ProviderRequest,
config: &ClaudeCodeConfig,
sessions: &dyn SessionStore,
) -> Result<Plan, PlanError> {
if !request.tools.is_empty() {
return Err(PlanError::Tools);
}
if request.output_schema_json.is_some() {
return Err(PlanError::OutputSchema);
}
let input = &request.input;
let run_start = input
.iter()
.rposition(|item| {
!matches!(
item,
InputItem::UserText { .. } | InputItem::ToolResult { .. }
)
})
.map_or(0, |at| at + 1);
let run = &input[run_start..];
if run
.iter()
.any(|item| matches!(item, InputItem::ToolResult { .. }))
{
return Err(PlanError::ToolResult);
}
let user_blocks: Vec<String> = run
.iter()
.filter_map(|item| match item {
InputItem::UserText { text } => Some(text.clone()),
_ => None,
})
.collect();
if user_blocks.is_empty() {
return Err(PlanError::NoUserText);
}
let last_own = input.iter().rposition(
|item| matches!(item, InputItem::ReasoningState { provider, .. } if provider == PROVIDER),
);
if let Some(at) = last_own {
let InputItem::ReasoningState { state, .. } = &input[at] else {
unreachable!("rposition matched a reasoning state");
};
let Some(ClaudeCodeHandle::Ended {
session_id,
instance,
prefix_sha256,
}) = ClaudeCodeHandle::decode(state)
else {
return Err(PlanError::UndecodableHandle);
};
let tail_is_user_text = input[at + 1..]
.iter()
.all(|item| matches!(item, InputItem::UserText { .. }));
let resumable = instance == config.instance
&& tail_is_user_text
&& prefix_sha256 == prefix_digest(&request.instructions, &request.model, &input[..at])
&& sessions.exists(&session_id);
if resumable {
return Ok(Plan::Resume {
session_id,
blocks: user_blocks,
});
}
}
let mut blocks = Vec::new();
if let Some(transcript) = render_transcript(&input[..run_start]) {
blocks.push(transcript);
}
blocks.extend(user_blocks);
Ok(Plan::Seed { blocks })
}
/// The earlier conversation as one deterministic text, or `None` when nothing
/// of it is renderable. Reasoning states are omitted.
fn render_transcript(items: &[InputItem]) -> Option<String> {
let groups: Vec<String> = items
.iter()
.filter_map(|item| match item {
InputItem::UserText { text } => Some(format!("[user]\n{text}")),
InputItem::AssistantText { text } => Some(format!("[assistant]\n{text}")),
InputItem::ToolCall {
call_id,
name,
arguments,
} => Some(format!("[tool_call name={name} id={call_id}]\n{arguments}")),
InputItem::ToolResult { call_id, output } => {
Some(format!("[tool_result id={call_id}]\n{output}"))
}
InputItem::ReasoningState { .. } => None,
})
.collect();
if groups.is_empty() {
return None;
}
Some(format!(
"The conversation so far, replayed as text because the earlier session cannot be continued:\n\n{}",
groups.join("\n\n")
))
}
#[derive(Serialize)]
struct UserLine<'a> {
#[serde(rename = "type")]
kind: &'static str,
message: UserMessage<'a>,
parent_tool_use_id: Option<&'a str>,
}
#[derive(Serialize)]
struct UserMessage<'a> {
role: &'static str,
content: Vec<TextBlock<'a>>,
}
#[derive(Serialize)]
struct TextBlock<'a> {
#[serde(rename = "type")]
kind: &'static str,
text: &'a str,
}
/// The child's whole stdin: exactly one stream-json user line. Several lines
/// would queue several turns.
pub(crate) fn stdin_bytes(blocks: &[String]) -> Vec<u8> {
let line = UserLine {
kind: "user",
message: UserMessage {
role: "user",
content: blocks
.iter()
.map(|text| TextBlock { kind: "text", text })
.collect(),
},
parent_tool_use_id: None,
};
let mut bytes = serde_json::to_vec(&line).expect("typed strings encode");
bytes.push(b'\n');
bytes
}

View file

@ -0,0 +1,148 @@
//! The claude argv and the child environment: every flag and variable name
//! this backend sends is a constant or built here, so a CLI upgrade is one
//! place to re-probe.
use std::fmt;
use std::fs::OpenOptions;
use std::io::{self, Write};
use std::path::{Path, PathBuf};
use std::sync::atomic::{AtomicU64, Ordering};
use model_connector::{ProviderRequest, ReasoningEffort};
use zeroize::Zeroizing;
use crate::config::ClaudeCodeConfig;
use crate::handle::Plan;
/// The one credential variable, copied by name from the connector's own
/// environment into the child and nowhere else.
pub(crate) const TOKEN_VAR: &str = "CLAUDE_CODE_OAUTH_TOKEN";
const CHILD_PATH: &str = "/usr/local/bin:/usr/bin:/bin";
/// The environment of the child: exactly these names and nothing inherited.
/// Values are never formatted; `Debug` lists names only.
pub struct ChildEnv(Vec<(&'static str, Zeroizing<String>)>);
impl ChildEnv {
pub fn iter(&self) -> impl Iterator<Item = (&str, &str)> {
self.0.iter().map(|(name, value)| (*name, value.as_str()))
}
}
impl fmt::Debug for ChildEnv {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_list()
.entries(self.0.iter().map(|(name, _)| name))
.finish()
}
}
pub(crate) fn child_env(
config: &ClaudeCodeConfig,
request: &ProviderRequest,
token: Zeroizing<String>,
) -> ChildEnv {
let config_dir = config.config_dir.to_string_lossy().into_owned();
ChildEnv(vec![
("PATH", Zeroizing::new(CHILD_PATH.to_string())),
("HOME", Zeroizing::new(config_dir.clone())),
("CLAUDE_CONFIG_DIR", Zeroizing::new(config_dir)),
(
"CLAUDE_CODE_MAX_OUTPUT_TOKENS",
Zeroizing::new(request.max_output_tokens.to_string()),
),
(TOKEN_VAR, token),
])
}
/// The system prompt staged for one child: an exclusively created file only
/// its owner can read, removed when this value drops, however the call ends.
pub(crate) struct PromptFile(PathBuf);
static PROMPT_FILES: AtomicU64 = AtomicU64::new(0);
impl PromptFile {
pub(crate) fn create(instructions: &str) -> io::Result<Self> {
let name = format!(
"claude-connector-prompt-{}-{}",
std::process::id(),
PROMPT_FILES.fetch_add(1, Ordering::Relaxed)
);
let path = std::env::temp_dir().join(name);
let mut options = OpenOptions::new();
options.write(true).create_new(true);
#[cfg(unix)]
std::os::unix::fs::OpenOptionsExt::mode(&mut options, 0o600);
let mut file = options.open(&path)?;
let owned = Self(path);
file.write_all(instructions.as_bytes())?;
Ok(owned)
}
pub(crate) fn path(&self) -> &Path {
&self.0
}
}
impl Drop for PromptFile {
fn drop(&mut self) {
let _ = std::fs::remove_file(&self.0);
}
}
fn effort(effort: ReasoningEffort) -> &'static str {
match effort {
ReasoningEffort::Low => "low",
ReasoningEffort::Medium => "medium",
ReasoningEffort::High => "high",
ReasoningEffort::Max => "max",
}
}
/// The program and its arguments. The system prompt is replaced, by a file
/// rather than an argument because one argument is capped near 128 KiB
/// (E2BIG) and instructions are not, the child has
/// no tools, no MCP servers, no settings, no slash commands and nobody to ask
/// for permission, and a resume always forks.
pub(crate) fn argv(
config: &ClaudeCodeConfig,
request: &ProviderRequest,
plan: &Plan,
system_prompt_file: &Path,
) -> Vec<String> {
let mut argv: Vec<String> = [
config.claude_binary.to_string_lossy().as_ref(),
"-p",
"--output-format",
"stream-json",
"--verbose",
"--input-format",
"stream-json",
"--model",
&request.model,
"--system-prompt-file",
&system_prompt_file.to_string_lossy(),
"--tools",
"",
"--strict-mcp-config",
"--setting-sources",
"",
"--disable-slash-commands",
"--permission-prompts",
"none",
]
.into_iter()
.map(str::to_string)
.collect();
if let Some(level) = request.reasoning_effort {
argv.extend(["--effort".to_string(), effort(level).to_string()]);
}
if let Plan::Resume { session_id, .. } = plan {
argv.extend([
"--resume".to_string(),
session_id.clone(),
"--fork-session".to_string(),
]);
}
argv
}

View file

@ -0,0 +1,19 @@
//! The Claude backend of the connector kit. [`ClaudeCodeBackend`] answers each
//! call by driving the unmodified `claude` binary once, in print mode with
//! stream-json in and out, with every tool, MCP server, settings source and
//! plugin switched off. A binary wires it to [`model_connector::serve`].
mod backend;
mod config;
mod handle;
mod launch;
mod process;
mod stream;
pub use backend::{ClaudeCodeBackend, StderrSink};
pub use config::{ClaudeCodeConfig, ClaudeCodeLimits};
pub use handle::{FsSessionStore, SessionStore};
pub use launch::ChildEnv;
#[cfg(unix)]
pub use process::StdProcessPort;
pub use process::{ProcessPort, RunningChild, StdoutRead};

View file

@ -0,0 +1,232 @@
//! The one OS seam: starting the child and reading its output.
use std::io;
use std::path::Path;
use std::time::Instant;
use crate::launch::ChildEnv;
/// What one read of the child's stdout produced.
#[derive(Debug, PartialEq, Eq)]
pub enum StdoutRead {
/// One non-empty line, without its terminator.
Line(Vec<u8>),
/// The child closed its stdout.
Eof,
/// The deadline passed first.
Timeout,
/// A line or the total output passed its cap.
Breach,
}
/// Starts a `claude` child. `argv[0]` is the program; the environment is
/// exactly `env`; `stdin` is written and closed.
pub trait ProcessPort: Send + Sync {
fn spawn(
&self,
argv: &[String],
env: &ChildEnv,
cwd: &Path,
stdin: Vec<u8>,
) -> io::Result<Box<dyn RunningChild>>;
}
/// A started child. The backend ends every child with `kill` then `wait`.
pub trait RunningChild: Send {
fn read_line(&mut self, deadline: Instant) -> StdoutRead;
/// The last bytes the child wrote to stderr.
fn stderr_tail(&self) -> Vec<u8>;
/// Kills the child and everything it started.
fn kill(&mut self);
fn wait(&mut self);
}
#[cfg(unix)]
pub use unix::StdProcessPort;
#[cfg(unix)]
mod unix {
use std::io::{self, BufRead, BufReader, Read, Write};
use std::os::unix::process::CommandExt;
use std::path::Path;
use std::process::{Child, Command, Stdio};
use std::sync::mpsc::{Receiver, RecvTimeoutError, channel};
use std::sync::{Arc, Mutex};
use std::thread;
use std::time::{Duration, Instant};
use super::{ProcessPort, RunningChild, StdoutRead};
use crate::config::ClaudeCodeLimits;
use crate::launch::ChildEnv;
/// Spawns each child in its own process group, so that killing the call
/// kills what the child started too.
pub struct StdProcessPort {
limits: ClaudeCodeLimits,
}
impl StdProcessPort {
pub fn new(limits: ClaudeCodeLimits) -> Self {
Self { limits }
}
}
impl ProcessPort for StdProcessPort {
fn spawn(
&self,
argv: &[String],
env: &ChildEnv,
cwd: &Path,
stdin: Vec<u8>,
) -> io::Result<Box<dyn RunningChild>> {
let (program, args) = argv
.split_first()
.ok_or_else(|| io::Error::from(io::ErrorKind::InvalidInput))?;
let mut command = Command::new(program);
command
.args(args)
.env_clear()
.envs(env.iter())
.current_dir(cwd)
.stdin(Stdio::piped())
.stdout(Stdio::piped())
.stderr(Stdio::piped())
.process_group(0);
let mut child = command.spawn()?;
let (stdin_pipe, stdout, stderr) =
(child.stdin.take(), child.stdout.take(), child.stderr.take());
if let Some(mut pipe) = stdin_pipe {
thread::spawn(move || {
let _ = pipe.write_all(&stdin);
});
}
let (lines, receiver) = channel();
if let Some(stdout) = stdout {
let limits = self.limits;
thread::spawn(move || {
let mut reader = BufReader::new(stdout);
let mut total = 0usize;
loop {
let mut line = Vec::new();
let read = reader
.by_ref()
.take(limits.max_line_bytes as u64 + 1)
.read_until(b'\n', &mut line);
let Ok(read) = read else { break };
if read == 0 {
break;
}
total = total.saturating_add(read);
let terminated = line.last() == Some(&b'\n');
if terminated {
line.pop();
}
if line.last() == Some(&b'\r') {
line.pop();
}
if line.len() > limits.max_line_bytes || total > limits.max_total_bytes {
let _ = lines.send(Piece::Breach);
return;
}
if !line.is_empty() && lines.send(Piece::Line(line)).is_err() {
return;
}
}
});
}
let tail = Arc::new(Mutex::new(Vec::new()));
let stderr_reader = stderr.map(|mut stderr| {
let (tail, keep) = (tail.clone(), self.limits.stderr_tail_bytes);
thread::spawn(move || {
let mut chunk = [0u8; 4096];
while let Ok(read) = stderr.read(&mut chunk) {
if read == 0 {
break;
}
let mut held = tail.lock().unwrap_or_else(|e| e.into_inner());
held.extend_from_slice(&chunk[..read]);
let excess = held.len().saturating_sub(keep);
held.drain(..excess);
}
})
});
Ok(Box::new(StdChild {
child,
receiver,
tail,
stderr_reader,
reaped: false,
}))
}
}
enum Piece {
Line(Vec<u8>),
Breach,
}
struct StdChild {
child: Child,
receiver: Receiver<Piece>,
tail: Arc<Mutex<Vec<u8>>>,
stderr_reader: Option<thread::JoinHandle<()>>,
reaped: bool,
}
impl RunningChild for StdChild {
fn read_line(&mut self, deadline: Instant) -> StdoutRead {
match self
.receiver
.recv_timeout(deadline.saturating_duration_since(Instant::now()))
{
Ok(Piece::Line(line)) => StdoutRead::Line(line),
Ok(Piece::Breach) => StdoutRead::Breach,
Err(RecvTimeoutError::Timeout) => StdoutRead::Timeout,
Err(RecvTimeoutError::Disconnected) => StdoutRead::Eof,
}
}
fn stderr_tail(&self) -> Vec<u8> {
self.tail.lock().unwrap_or_else(|e| e.into_inner()).clone()
}
fn kill(&mut self) {
// The group leader stays a zombie until `wait`, so its pid is
// still ours to signal.
if !self.reaped {
// SAFETY: killpg takes a process group id and no pointers.
unsafe {
libc::killpg(self.child.id() as libc::pid_t, libc::SIGKILL);
}
}
}
fn wait(&mut self) {
let _ = self.child.wait();
self.reaped = true;
// Every writer to stderr died with the group, so the tail is
// complete once its reader sees the end. A process that escaped
// the group is not waited for.
if let Some(reader) = self.stderr_reader.take() {
for _ in 0..100 {
if reader.is_finished() {
let _ = reader.join();
break;
}
thread::sleep(Duration::from_millis(2));
}
}
}
}
impl Drop for StdChild {
fn drop(&mut self) {
self.kill();
self.wait();
}
}
}

View file

@ -0,0 +1,295 @@
//! Reading the CLI's stream-json output. The format is foreign and moves with
//! the CLI, so every struct here ignores fields it does not name.
use std::collections::BTreeMap;
use model_connector::{FailureKind, StopReason, Usage};
use serde::Deserialize;
use serde_json::Value;
/// A call that cannot complete, with the stage that failed. `usage` is `None`
/// unless a result event reported it.
#[derive(Debug, PartialEq, Eq)]
pub(crate) struct Failure {
pub kind: FailureKind,
pub message: &'static str,
pub usage: Option<Usage>,
}
impl Failure {
pub(crate) fn unbilled(kind: FailureKind, message: &'static str) -> Self {
Self {
kind,
message,
usage: None,
}
}
}
/// What the stream said, once its result event arrived.
#[derive(Debug, PartialEq, Eq)]
pub(crate) struct Finished {
pub session_id: String,
/// The non-empty text blocks, in order.
pub texts: Vec<String>,
/// The id of the last assistant message.
pub message_id: Option<String>,
pub stop_reason: StopReason,
pub usage: Usage,
}
pub(crate) enum Step {
More,
Done(Box<Result<Finished, Failure>>),
}
#[derive(Deserialize)]
struct Line {
#[serde(rename = "type")]
kind: String,
subtype: Option<String>,
session_id: Option<String>,
model: Option<String>,
message: Option<Message>,
rate_limit_info: Option<RateLimitInfo>,
is_error: Option<bool>,
stop_reason: Option<String>,
api_error_status: Option<Value>,
result: Option<Value>,
errors: Option<Value>,
/// What this turn alone consumed. `modelUsage` is deliberately not read:
/// on a resumed session it carries every earlier turn too.
model_usage_since_last_result: Option<Value>,
}
#[derive(Deserialize)]
struct Message {
id: Option<String>,
content: Option<Vec<Block>>,
}
#[derive(Deserialize)]
struct Block {
#[serde(rename = "type")]
kind: String,
text: Option<String>,
}
#[derive(Deserialize)]
struct RateLimitInfo {
status: Option<String>,
}
/// One model's entry. The four counts are required: an entry that lacks one
/// is a shape this reader does not know, and unknown usage is never zero.
#[derive(Deserialize)]
#[serde(rename_all = "camelCase")]
struct ModelUsage {
input_tokens: u64,
output_tokens: u64,
cache_read_input_tokens: u64,
cache_creation_input_tokens: u64,
thinking_tokens: Option<u64>,
}
/// One call's stream, line by line.
pub(crate) struct StreamState<'a> {
model: &'a str,
session_id: Option<String>,
texts: Vec<String>,
message_id: Option<String>,
plan_rejected: bool,
}
impl<'a> StreamState<'a> {
pub(crate) fn new(model: &'a str) -> Self {
Self {
model,
session_id: None,
texts: Vec::new(),
message_id: None,
plan_rejected: false,
}
}
pub(crate) fn feed(&mut self, raw: &[u8]) -> Step {
let Ok(line) = serde_json::from_slice::<Line>(raw) else {
return fail(
FailureKind::Provider,
"claude-code: a stream line did not parse",
);
};
match line.kind.as_str() {
"system" => match line.subtype.as_deref() {
Some("init") => {
if line.model.as_deref() != Some(self.model) {
return fail(
FailureKind::Provider,
"claude-code: the session model differs from the request",
);
}
self.session_id = line.session_id;
Step::More
}
Some("compact_boundary") => fail(
FailureKind::Provider,
"claude-code: the session was compacted",
),
_ => Step::More,
},
"assistant" => self.assistant(line.message),
"rate_limit_event" => {
self.plan_rejected |= line
.rate_limit_info
.and_then(|info| info.status)
.is_some_and(|status| status == "rejected");
Step::More
}
"result" => Step::Done(Box::new(self.result(line))),
_ => Step::More,
}
}
fn assistant(&mut self, message: Option<Message>) -> Step {
let Some(message) = message else {
return Step::More;
};
if message.id.is_some() {
self.message_id = message.id;
}
for block in message.content.unwrap_or_default() {
match block.kind.as_str() {
"text" => {
if let Some(text) = block.text.filter(|text| !text.is_empty()) {
self.texts.push(text);
}
}
"tool_use" => {
return fail(
FailureKind::Provider,
"claude-code: the model called a tool in a tool-less turn",
);
}
_ => {}
}
}
Step::More
}
fn result(&mut self, line: Line) -> Result<Finished, Failure> {
let success = line.subtype.as_deref() == Some("success") && line.is_error == Some(false);
let door_refusal = !success
&& matches!(
line.api_error_status.as_ref().and_then(Value::as_u64),
Some(401 | 403)
);
let usage = read_usage(line.model_usage_since_last_result, door_refusal);
if !success {
return Err(Failure {
kind: self.error_kind(&line.api_error_status, &line.result, &line.errors),
message: "claude-code: the result reported an error",
usage,
});
}
let Some(usage) = usage else {
return Err(Failure {
kind: FailureKind::Provider,
message: "claude-code: the result did not report usable usage",
usage: None,
});
};
let stop_reason = match line.stop_reason.as_deref() {
Some("end_turn") => StopReason::EndTurn,
Some("max_tokens") => StopReason::MaxOutputTokens,
Some("stop_sequence") => StopReason::StopSequence,
Some("refusal") => StopReason::Refusal,
Some("model_context_window_exceeded") => StopReason::ContextWindow,
_ => {
return Err(Failure {
kind: FailureKind::Provider,
message: "claude-code: the result has an unknown stop reason",
usage: Some(usage),
});
}
};
let Some(session_id) = self.session_id.take() else {
return Err(Failure {
kind: FailureKind::Provider,
message: "claude-code: the stream never announced its session",
usage: Some(usage),
});
};
Ok(Finished {
session_id,
texts: std::mem::take(&mut self.texts),
message_id: self.message_id.take(),
stop_reason,
usage,
})
}
fn error_kind(
&self,
status: &Option<Value>,
result: &Option<Value>,
errors: &Option<Value>,
) -> FailureKind {
let status = status.as_ref().and_then(Value::as_u64);
let names_rate_limit = [result, errors].into_iter().flatten().any(|value| {
let text = value.to_string().to_lowercase();
text.contains("rate limit") || text.contains("rate_limit")
});
match status {
_ if self.plan_rejected || names_rate_limit || status == Some(429) => {
FailureKind::RateLimited
}
Some(401 | 403) => FailureKind::Authentication,
Some(400) => FailureKind::InvalidRequest,
Some(500..=599) => FailureKind::Unavailable,
_ => FailureKind::Provider,
}
}
}
fn fail(kind: FailureKind, message: &'static str) -> Step {
Step::Done(Box::new(Err(Failure::unbilled(kind, message))))
}
/// What this turn consumed, summed over every model it used (harness side
/// calls included), or `None` when the result does not say: the field absent,
/// of another shape, or empty. An empty report is believed only for a refusal
/// at the door (401 or 403), which the provider cannot have billed; elsewhere
/// the CLI also zeroes usage on crashes after real work. `None` is billed the
/// reservation by the kit.
fn read_usage(reported: Option<Value>, door_refusal: bool) -> Option<Usage> {
let models: BTreeMap<String, ModelUsage> = serde_json::from_value(reported?).ok()?;
if models.is_empty() {
return door_refusal.then(|| sum_usage(std::iter::empty()));
}
Some(sum_usage(models.into_values()))
}
fn sum_usage(models: impl Iterator<Item = ModelUsage>) -> Usage {
let mut usage = Usage {
input_tokens: 0,
cache_read_tokens: 0,
cache_write_tokens: 0,
output_tokens: 0,
reasoning_tokens: None,
};
for model in models {
usage.input_tokens = usage.input_tokens.saturating_add(model.input_tokens);
usage.cache_read_tokens = usage
.cache_read_tokens
.saturating_add(model.cache_read_input_tokens);
usage.cache_write_tokens = usage
.cache_write_tokens
.saturating_add(model.cache_creation_input_tokens);
usage.output_tokens = usage.output_tokens.saturating_add(model.output_tokens);
if let Some(thinking) = model.thinking_tokens {
usage.reasoning_tokens =
Some(usage.reasoning_tokens.unwrap_or(0).saturating_add(thinking));
}
}
usage
}