diff --git a/Cargo.lock b/Cargo.lock index a3cbdf7..34dec06 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -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" diff --git a/README.md b/README.md index 596f571..339e793 100644 --- a/README.md +++ b/README.md @@ -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. diff --git a/crates/claude-connector/Cargo.toml b/crates/claude-connector/Cargo.toml new file mode 100644 index 0000000..a35bb4a --- /dev/null +++ b/crates/claude-connector/Cargo.toml @@ -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" diff --git a/crates/claude-connector/src/backend.rs b/crates/claude-connector/src/backend.rs new file mode 100644 index 0000000..ad3fb8c --- /dev/null +++ b/crates/claude-connector/src/backend.rs @@ -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 Option + 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, + sessions: Box, + sink: Box, + 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) -> 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, + sessions: Box, + sink: Box, + ) -> Self { + Self { + config, + port, + sessions, + sink, + env_lookup: Box::new(|name| std::env::var(name).ok()), + } + } + + fn token(&self) -> Option> { + (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, 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, TransportOutcome) { + let mut payloads: Vec = 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 { + 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, +) -> (Vec, TransportOutcome) { + ( + Vec::new(), + TransportOutcome::Failed { + kind, + message, + usage, + }, + ) +} + +fn failure(failure: Failure) -> (Vec, TransportOutcome) { + failed(failure.kind, failure.message.to_string(), failure.usage) +} + +#[cfg(test)] +mod tests; diff --git a/crates/claude-connector/src/backend/tests.rs b/crates/claude-connector/src/backend/tests.rs new file mode 100644 index 0000000..309e2d1 --- /dev/null +++ b/crates/claude-connector/src/backend/tests.rs @@ -0,0 +1,1851 @@ +use std::collections::VecDeque; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::sync::{Arc, Mutex}; +use std::time::Duration; + +use model_connector::{ + CallerAdmission, ModelPrice, PriceTable, Rates, ReasoningEffort, SpendCeiling, SpendWindowMs, + ToolDefinition, TransportAdmission, TransportDisposition, TransportInvocation, TransportKey, + TransportReceipt, TransportService, decrypt_result, encrypt_invocation, +}; + +use super::*; +use crate::config::ClaudeCodeLimits; +use crate::handle::{FsSessionStore, SessionStore}; +use crate::launch::ChildEnv; + +const CANARY_TOKEN: &str = "canary-oauth-token-value"; +const MODEL: &str = "claude-haiku-5-5"; +const INSTANCE: &str = "persona-a"; +const SESSION: &str = "11111111-2222-3333-4444-555555555555"; +const HEX_SESSION: &str = "aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee"; +const OTHER_MODEL: &str = "claude-opus-5"; +const FORK: &str = "66666666-7777-8888-9999-000000000000"; +const FRAME: usize = 1 << 20; + +// ---- fixtures, written by hand from the fields the probes observed ---- + +fn init_line(model: &str, session: &str) -> String { + format!( + r#"{{"type":"system","subtype":"init","session_id":"{session}","model":"{model}","tools":[],"apiKeySource":"none"}}"# + ) +} + +fn text_line(id: &str, text: &str) -> String { + format!( + r#"{{"type":"assistant","message":{{"id":"{id}","role":"assistant","content":[{{"type":"text","text":"{text}"}}],"stop_reason":null}},"session_id":"{SESSION}"}}"# + ) +} + +fn thinking_line(id: &str) -> String { + format!( + r#"{{"type":"assistant","message":{{"id":"{id}","role":"assistant","content":[{{"type":"thinking","thinking":"","signature":"sig"}}]}}}}"# + ) +} + +fn tool_use_line() -> String { + r#"{"type":"assistant","message":{"id":"msg_t","content":[{"type":"tool_use","id":"toolu_1","name":"x","input":{}}]}}"#.to_string() +} + +fn rate_limit_line(status: &str) -> String { + format!( + r#"{{"type":"rate_limit_event","rate_limit_info":{{"status":"{status}","rateLimitType":"five_hour"}}}}"# + ) +} + +const USAGE: &str = r#"{"claude-haiku-5-5":{"inputTokens":2,"outputTokens":4,"cacheReadInputTokens":7,"cacheCreationInputTokens":1043,"thinkingTokens":3,"costUSD":0.5}}"#; + +/// What a resumed session's `modelUsage` says: every turn so far. Every result +/// fixture carries it beside the turn's own `model_usage_since_last_result`, +/// so reading the wrong field shows in a price. +const CUMULATIVE: &str = r#"{"claude-haiku-5-5":{"inputTokens":900000,"outputTokens":900000,"cacheReadInputTokens":900000,"cacheCreationInputTokens":900000,"thinkingTokens":900000,"costUSD":99.0}}"#; + +fn result_line(stop_reason: &str, since_last: &str) -> String { + format!( + r#"{{"type":"result","subtype":"success","is_error":false,"stop_reason":"{stop_reason}","session_id":"{SESSION}","total_cost_usd":99.0,"modelUsage":{CUMULATIVE},"model_usage_since_last_result":{since_last}}}"# + ) +} + +fn error_result_line(extra: &str) -> String { + format!( + r#"{{"type":"result","subtype":"error_during_execution","is_error":true,"session_id":"{SESSION}","total_cost_usd":99.0,"modelUsage":{CUMULATIVE},"model_usage_since_last_result":{USAGE}{extra}}}"# + ) +} + +fn ok_script(session: &str, text: &str) -> Script { + Script::lines(vec![ + init_line(MODEL, session), + text_line("msg_1", text), + rate_limit_line("allowed"), + result_line("end_turn", USAGE), + ]) +} + +fn usage( + input: u64, + cache_read: u64, + cache_write: u64, + output: u64, + reasoning: Option, +) -> Usage { + Usage { + input_tokens: input, + cache_read_tokens: cache_read, + cache_write_tokens: cache_write, + output_tokens: output, + reasoning_tokens: reasoning, + } +} + +fn user(text: &str) -> InputItem { + InputItem::UserText { text: text.into() } +} + +fn assistant(text: &str) -> InputItem { + InputItem::AssistantText { text: text.into() } +} + +fn request(input: Vec) -> ProviderRequest { + let mut request = ProviderRequest::new("req", "conv", MODEL, "Be terse.", 1024); + request.input = input; + request +} + +// ---- the scripted port: replays fixture lines and records what it was given ---- + +#[derive(Clone, Copy)] +enum Terminal { + Eof, + Timeout, + Breach, +} + +struct Script { + lines: Vec, + terminal: Terminal, +} + +impl Script { + fn lines(lines: Vec) -> Self { + Self { + lines, + terminal: Terminal::Eof, + } + } + + fn ending(mut self, terminal: Terminal) -> Self { + self.terminal = terminal; + self + } +} + +#[derive(Clone, Default)] +struct Seen { + argv: Vec, + env: Vec<(String, String)>, + cwd: PathBuf, + stdin: Vec, + /// The file `--system-prompt-file` named, as it was when the child started. + prompt_file: Option<(PathBuf, String)>, + /// Every deadline the backend passed to a read of this child. + deadlines: Arc>>, +} + +struct ScriptedPort { + scripts: Mutex>, + spawns: Arc, + seen: Arc>, + stderr: Vec, +} + +struct ScriptedChild { + lines: VecDeque, + terminal: Terminal, + stderr: Vec, + deadlines: Arc>>, +} + +impl ProcessPort for ScriptedPort { + fn spawn( + &self, + argv: &[String], + env: &ChildEnv, + cwd: &Path, + stdin: Vec, + ) -> io::Result> { + self.spawns.fetch_add(1, Ordering::SeqCst); + let deadlines = Arc::new(Mutex::new(Vec::new())); + *self.seen.lock().unwrap() = Seen { + prompt_file: arg_after(argv, "--system-prompt-file").map(|path| { + ( + PathBuf::from(path), + std::fs::read_to_string(path).expect("the prompt file exists at spawn"), + ) + }), + deadlines: deadlines.clone(), + argv: argv.to_vec(), + env: env + .iter() + .map(|(k, v)| (k.to_string(), v.to_string())) + .collect(), + cwd: cwd.to_path_buf(), + stdin, + }; + let script = self + .scripts + .lock() + .unwrap() + .pop_front() + .expect("the test scripted no further spawn"); + Ok(Box::new(ScriptedChild { + lines: script.lines.into(), + terminal: script.terminal, + stderr: self.stderr.clone(), + deadlines, + })) + } +} + +impl RunningChild for ScriptedChild { + fn read_line(&mut self, deadline: Instant) -> StdoutRead { + self.deadlines.lock().unwrap().push(deadline); + match self.lines.pop_front() { + Some(line) => StdoutRead::Line(line.into_bytes()), + None => match self.terminal { + Terminal::Eof => StdoutRead::Eof, + Terminal::Timeout => StdoutRead::Timeout, + Terminal::Breach => StdoutRead::Breach, + }, + } + } + fn stderr_tail(&self) -> Vec { + self.stderr.clone() + } + fn kill(&mut self) {} + fn wait(&mut self) {} +} + +struct Sessions(Vec); + +impl SessionStore for Sessions { + fn exists(&self, session_id: &str) -> bool { + self.0.iter().any(|known| known == session_id) + } +} + +#[derive(Clone, Default)] +struct Sink(Arc>>); + +impl StderrSink for Sink { + fn stderr_tail(&self, tail: &[u8]) { + self.0.lock().unwrap().extend_from_slice(tail); + } +} + +fn config(instance: &str, limits: ClaudeCodeLimits) -> ClaudeCodeConfig { + ClaudeCodeConfig { + claude_binary: PathBuf::from("/opt/claude"), + claude_binary_sha256: [9; 32], + work_dir: PathBuf::from("/work"), + config_dir: PathBuf::from("/cfg"), + instance: instance.to_string(), + limits, + } +} + +fn limits() -> ClaudeCodeLimits { + ClaudeCodeLimits { + deadline: Duration::from_secs(30), + max_line_bytes: 1 << 20, + max_total_bytes: 1 << 24, + stderr_tail_bytes: 4096, + } +} + +fn with_canary_token(mut backend: ClaudeCodeBackend) -> ClaudeCodeBackend { + backend.env_lookup = Box::new(|name| (name == TOKEN_VAR).then(|| CANARY_TOKEN.to_string())); + backend +} + +struct Rig { + backend: ClaudeCodeBackend, + spawns: Arc, + seen: Arc>, + sink: Sink, +} + +fn rig_for(instance: &str, scripts: Vec