diff --git a/Cargo.lock b/Cargo.lock index aa377dd..79bc767 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -666,6 +666,7 @@ dependencies = [ "codex-connector", "cultcache-rs", "ghostlight-persona-projection", + "reqwest", "rmp-serde", "serde", "serde_bytes", diff --git a/crates/ghostlight-dungeon/src/runtime.rs b/crates/ghostlight-dungeon/src/runtime.rs index 26c7518..e134dcc 100644 --- a/crates/ghostlight-dungeon/src/runtime.rs +++ b/crates/ghostlight-dungeon/src/runtime.rs @@ -13,11 +13,12 @@ use crate::{ use ghostlight::{ AffordanceId, CONSUMER_BODY_LIMIT, CommandBody, CommandId, ConnectorBinding, ConsumerPort, ConsumerRegistry, ControllerModels, ControllerRunner, ControllerWorkCustody, - CreateJurisdictionIntent, CreateWorldIntent, DEFAULT_SDK_MODEL_PREFIX, DecisionInvocation, - DecisionOpportunity, KernelError, Lens, LensWeights, MailboxError, PrincipalCommandIntent, - PrincipalId, SdkBinding, SeedOutcome, SeedPort, Statement, SubjectKind, SubmitReceipt, - TickMinutes, VaultEvidenceSource, VerifiedPrincipalEvidence, WorldMailbox, WorldPhase, - WorldSnapshot, open_controller_work, open_inference, + CreateJurisdictionIntent, CreateWorldIntent, DEFAULT_LOCAL_MODEL_PREFIX, + DEFAULT_SDK_MODEL_PREFIX, DecisionInvocation, DecisionOpportunity, KernelError, Lens, + LensWeights, LocalBinding, MailboxError, PrincipalCommandIntent, PrincipalId, SdkBinding, + SeedOutcome, SeedPort, Statement, SubjectKind, SubmitReceipt, TickMinutes, VaultEvidenceSource, + VerifiedPrincipalEvidence, WorldMailbox, WorldPhase, WorldSnapshot, open_controller_work, + open_inference, }; use anyhow::{Context, bail, ensure}; use axum::{ @@ -390,7 +391,21 @@ fn open_controller( model_prefix: std::env::var("GHOSTLIGHT_SDK_MODEL_PREFIX") .unwrap_or_else(|_| DEFAULT_SDK_MODEL_PREFIX.into()), }); - let inference = open_inference(connector, sdk, &models)?; + // Ghostlight reads no credential for this transport either: a loopback + // local model server needs none, and `open_inference` refuses anything + // else at open. + let local = match std::env::var("GHOSTLIGHT_LOCAL_ENDPOINT") { + Ok(value) => Some(LocalBinding { + endpoint: value.parse().with_context(|| { + format!("GHOSTLIGHT_LOCAL_ENDPOINT `{value}` is not a socket address") + })?, + model_prefix: std::env::var("GHOSTLIGHT_LOCAL_MODEL_PREFIX") + .unwrap_or_else(|_| DEFAULT_LOCAL_MODEL_PREFIX.into()), + caller_runtime_id: runtime_id.to_owned(), + }), + Err(_) => None, + }; + let inference = open_inference(connector, sdk, local, &models.each())?; let work = open_controller_work(service_root.join("controller-work.cc"))?; ControllerRunner::open(world.clone(), inference, work, models).map_err(Into::into) } @@ -1781,7 +1796,7 @@ mod tests { key_path: controller_key, caller_runtime_id: "ghostlight-runtime-test".into(), }; - let inference = open_inference(Some(connector), None, &models).unwrap(); + let inference = open_inference(Some(connector), None, None, &models.each()).unwrap(); let work = open_controller_work(directory.path().join("controller-work.cc")).unwrap(); let controllers = ControllerRunner::open(world.clone(), inference, work, models).unwrap(); let mesh = MeshPublisher::open( diff --git a/crates/ghostlight/Cargo.toml b/crates/ghostlight/Cargo.toml index 1641bfb..1d66013 100644 --- a/crates/ghostlight/Cargo.toml +++ b/crates/ghostlight/Cargo.toml @@ -10,6 +10,7 @@ chrono.workspace = true codex-connector.workspace = true cultcache-rs.workspace = true ghostlight-persona-projection = { path = "../ghostlight-persona-projection" } +reqwest.workspace = true rmp-serde.workspace = true serde.workspace = true serde_bytes = "0.11" diff --git a/crates/ghostlight/src/controllers.rs b/crates/ghostlight/src/controllers.rs index 74e867b..4493090 100644 --- a/crates/ghostlight/src/controllers.rs +++ b/crates/ghostlight/src/controllers.rs @@ -11,6 +11,7 @@ use super::elaboration::{ NullEvidenceSource, SeedCheckpoint, SeedRunner, valid_elaboration_progression, valid_seed_progression, }; +use super::local_inference::{LocalBinding, open_local_port}; use super::sdk_inference::{ ChildProcessLink, DEFAULT_SDK_MODEL_PREFIX, RoutedInferencePort, SdkBinding, SdkInferencePort, }; @@ -2293,14 +2294,16 @@ pub struct ControllerModels { impl ControllerModels { /// Every model a configured backend must claim, in one place, so the /// open-time routing check and the canonical-identifier gate read the same - /// list. - pub(crate) fn each(&self) -> [&String; 5] { + /// list. `pub` so a caller can extend it with a model this struct does not + /// carry (the play lane's model, Cut 8) before passing the combined slice + /// to `open_inference`. + pub fn each(&self) -> [&str; 5] { [ - &self.projector, - &self.persona, - &self.interpreter, - &self.operational_agent, - &self.elaborator, + self.projector.as_str(), + self.persona.as_str(), + self.interpreter.as_str(), + self.operational_agent.as_str(), + self.elaborator.as_str(), ] } @@ -2321,6 +2324,10 @@ pub enum ControllerOpenError { UnroutableModel { model: String }, #[error("the SDK sidecar entry `{path}` is not a file")] SdkSidecarMissing { path: String }, + #[error("the local inference endpoint `{endpoint}` is not loopback")] + LocalEndpointNotLoopback { endpoint: SocketAddr }, + #[error("the local and SDK inference transports both claim model prefix `{prefix}`")] + SharedModelPrefix { prefix: String }, } /// Everything the CodexConnector transport needs to open, gathered so @@ -2332,11 +2339,15 @@ pub struct ConnectorBinding { } /// Builds the one port every lane shares. A lane whose model no configured -/// backend claims fails here, at open, rather than at its first tick. +/// backend claims fails here, at open, rather than at its first tick. `models` +/// is every model that must route, not just `ControllerModels`'s own fields, +/// so a caller can pass `&models.each()` extended with a model that struct +/// does not carry (the play lane's, Cut 8). pub fn open_inference( connector: Option, sdk: Option, - models: &ControllerModels, + local: Option, + models: &[&str], ) -> Result, ControllerOpenError> { let connector: Option> = match connector { Some(binding) => Some(Arc::new(CodexConnectorInferencePort::from_secret_file( @@ -2362,11 +2373,34 @@ pub fn open_inference( } None => None, }; - let routed = RoutedInferencePort::new(connector, sdk, sdk_model_prefix); - for model in models.each() { + // Ghostlight never holds a credential for this transport: a loopback + // local model server needs none, and refusing anything else here is what + // keeps that true by construction rather than by convention. + let local: Option<(String, Arc)> = match local { + Some(binding) => { + if !binding.endpoint.ip().is_loopback() { + return Err(ControllerOpenError::LocalEndpointNotLoopback { + endpoint: binding.endpoint, + }); + } + let prefix = binding.model_prefix.clone(); + Some((prefix, open_local_port(binding))) + } + None => None, + }; + if let Some((prefix, _)) = &local + && sdk.is_some() + && *prefix == sdk_model_prefix + { + return Err(ControllerOpenError::SharedModelPrefix { + prefix: prefix.clone(), + }); + } + let routed = RoutedInferencePort::new(connector, sdk, sdk_model_prefix, local); + for &model in models { if routed.route(model).is_none() { return Err(ControllerOpenError::UnroutableModel { - model: model.clone(), + model: model.to_owned(), }); } } @@ -9224,7 +9258,8 @@ mod tests { caller_runtime_id: runtime_id, }), None, - &models, + None, + &models.each(), ) .unwrap(); let work = open_controller_work(directory.path().join("controller-work.cc")).unwrap(); diff --git a/crates/ghostlight/src/lib.rs b/crates/ghostlight/src/lib.rs index 38e4145..242e952 100644 --- a/crates/ghostlight/src/lib.rs +++ b/crates/ghostlight/src/lib.rs @@ -63,6 +63,7 @@ mod cover; mod elaboration; mod journal; mod lens; +mod local_inference; mod mailbox; mod patch; mod sdk_inference; @@ -92,6 +93,7 @@ pub use elaboration::{ EvidenceReceipt, EvidenceSource, SeedCheckpoint, SeedOutcome, SeedRunner, select_row, }; pub use lens::{Lens, LensWeights}; +pub use local_inference::{DEFAULT_LOCAL_MODEL_PREFIX, LocalBinding}; pub(crate) use mailbox::ElaborationPort; pub use mailbox::{ConsumerPort, ControllerPort, MailboxError, SeedPort, WorldMailbox}; pub use patch::{ diff --git a/crates/ghostlight/src/local_inference.rs b/crates/ghostlight/src/local_inference.rs new file mode 100644 index 0000000..0c1d9aa --- /dev/null +++ b/crates/ghostlight/src/local_inference.rs @@ -0,0 +1,860 @@ +//! A local OpenAI-compatible inference port, for a loopback model server this +//! process does not authenticate to and does not need a credential for. +//! +//! This module owns one authority: lowering a prepared request to one +//! `POST /v1/chat/completions` call and lifting the reply back into the same +//! `InferenceOutput` every other port produces. It runs no tool loop of its +//! own — a returned tool call is inert, exactly as the connector lane's is — +//! and it holds no mutex across a request, so several requests may be in +//! flight through the same port at once. + +use super::controllers::{ + InferenceEvent, InferenceFault, InferenceOutput, InferencePort, InferenceRequest, + PreparedInference, REQUEST_EXPIRY, RESPONSE_TIMEOUT, unix_ms, +}; +use async_trait::async_trait; +use codex_connector::CodexInputItem; +use serde::{Deserialize, Serialize}; +use serde_json::{Value, json}; +use sha2::{Digest, Sha256}; +use std::net::SocketAddr; +use std::sync::Arc; + +/// The model-name prefix that names the local transport when nothing +/// configures one. +pub const DEFAULT_LOCAL_MODEL_PREFIX: &str = "local/"; + +/// Everything the local port needs to open, gathered so `open_inference` +/// takes a binding rather than reading the environment itself. +pub struct LocalBinding { + pub endpoint: SocketAddr, + pub model_prefix: String, + pub caller_runtime_id: String, +} + +const LOCAL_RECEIPT_SCHEMA: &str = "ghostlight.local_inference_receipt.v1"; + +/// The identity half is computed here from the invocation, exactly as the SDK +/// receipt's is; the provenance half is the endpoint's own reply, carried +/// verbatim. +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +struct LocalInferenceReceipt { + schema_id: String, + request_id: String, + conversation_id: String, + caller_runtime_id: String, + native_request_sha256: [u8; 32], + provider_request_sha256: [u8; 32], + model: String, + response_id: String, + finish_reason: Option, + prompt_tokens: u64, + completion_tokens: u64, +} + +/// One OpenAI-compatible chat-completions reply, only as much of the shape as +/// this port needs. A reply that does not decode into this shape is an +/// integrity violation, not a best-effort read. +#[derive(Debug, Deserialize)] +struct LocalChatResponse { + id: String, + choices: Vec, + #[serde(default)] + usage: Option, +} + +#[derive(Debug, Deserialize)] +struct LocalChatChoice { + message: LocalChatMessage, + #[serde(default)] + finish_reason: Option, +} + +#[derive(Debug, Deserialize)] +struct LocalChatMessage { + #[serde(default)] + content: Option, + #[serde(default)] + tool_calls: Vec, +} + +#[derive(Debug, Deserialize)] +struct LocalChatToolCall { + id: String, + function: LocalChatFunctionCall, +} + +#[derive(Debug, Deserialize)] +struct LocalChatFunctionCall { + name: String, + arguments: String, +} + +#[derive(Debug, Deserialize)] +struct LocalChatUsage { + #[serde(default)] + prompt_tokens: u64, + #[serde(default)] + completion_tokens: u64, +} + +/// A call id the connector's own validators would refuse cannot be persisted, +/// same rule the SDK port applies to its sidecar's reports. +fn call_id_is_valid(call_id: &str) -> bool { + !call_id.is_empty() && call_id.len() <= 64 && call_id.is_ascii() +} + +fn tool_name_is_valid(name: &str) -> bool { + !name.is_empty() + && name.len() <= 64 + && name + .bytes() + .all(|byte| byte.is_ascii_alphanumeric() || byte == b'_' || byte == b'-') +} + +/// The local OpenAI-compatible backend behind Ghostlight's inference seam. +/// Unlike the SDK port, it runs no query loop: one prepared request lowers to +/// one HTTP call, and any tool call in the reply comes back inert for Rust's +/// own evaluator to drive the next round, exactly as the connector lane +/// already does. +pub(super) struct LocalInferencePort { + client: reqwest::Client, + endpoint: SocketAddr, + prefix: String, + caller_runtime_id: String, +} + +impl LocalInferencePort { + fn new(endpoint: SocketAddr, prefix: impl Into, caller_runtime_id: impl Into) -> Self { + Self { + client: reqwest::Client::builder() + .timeout(RESPONSE_TIMEOUT) + .build() + .expect("the local inference HTTP client builds with no custom TLS material"), + endpoint, + prefix: prefix.into(), + caller_runtime_id: caller_runtime_id.into(), + } + } + + async fn send(&self, body: Value) -> Result { + let response = self + .client + .post(format!("http://{}/v1/chat/completions", self.endpoint)) + .json(&body) + .send() + .await + .map_err(|error| { + if error.is_connect() { + InferenceFault::retryable(format!( + "the local inference endpoint refused the connection: {error}" + )) + } else { + InferenceFault::new(format!("the local inference request failed: {error}")) + } + })?; + let status = response.status(); + if !status.is_success() { + let detail = format!("the local inference endpoint returned {status}"); + return Err(if status.as_u16() == 429 || status.as_u16() == 503 { + InferenceFault::retryable(detail) + } else { + InferenceFault::new(detail) + }); + } + response.json::().await.map_err(|error| { + InferenceFault::integrity_violation(format!( + "the local inference endpoint's reply was not the declared shape: {error}" + )) + }) + } +} + +/// Lowers one prepared request to the OpenAI chat-completions body: the +/// system message is the request's instructions, every input item maps to +/// one message in order, and every offered tool is declared `strict`. No +/// header but content type is ever set on the request this builds. +fn lower_request(prepared: &PreparedInference, prefix: &str) -> Result { + let request = &prepared.invocation.request; + let model = request.model.strip_prefix(prefix).unwrap_or(&request.model); + let mut messages = vec![json!({"role": "system", "content": request.instructions})]; + for item in &request.input { + messages.push(match item { + CodexInputItem::UserText { text } => json!({"role": "user", "content": text}), + CodexInputItem::AssistantText { text } => json!({"role": "assistant", "content": text}), + CodexInputItem::ToolCall { + call_id, + name, + arguments, + } => json!({ + "role": "assistant", + "content": Value::Null, + "tool_calls": [{ + "id": call_id, + "type": "function", + "function": {"name": name, "arguments": arguments}, + }], + }), + CodexInputItem::ToolResult { call_id, output } => json!({ + "role": "tool", + "tool_call_id": call_id, + "content": output, + }), + }); + } + let mut tools = Vec::with_capacity(request.tools.len()); + for tool in &request.tools { + let parameters: Value = serde_json::from_str(&tool.parameters_json).map_err(|error| { + InferenceFault::integrity_violation(format!( + "tool `{}` parameters are not valid JSON: {error}", + tool.name + )) + })?; + tools.push(json!({ + "type": "function", + "function": { + "name": tool.name, + "description": tool.description, + "parameters": parameters, + "strict": true, + }, + })); + } + let mut body = json!({ + "model": model, + "messages": messages, + "tools": tools, + "parallel_tool_calls": request.parallel_tool_calls, + }); + if let Some(max_tokens) = request.max_output_tokens { + body["max_tokens"] = json!(max_tokens); + } + Ok(body) +} + +/// Lifts one chat-completions reply into the port's output. A tool call +/// naming a tool this request never offered, or carrying a call id or tool +/// name the provider contract refuses, is the endpoint lying about the +/// request it was given, not a gap — so it is an integrity violation, not a +/// dropped call. +fn assemble_output( + prepared: &PreparedInference, + response: LocalChatResponse, +) -> Result { + let request = &prepared.invocation.request; + let mut choices = response.choices; + if choices.is_empty() { + return Err(InferenceFault::integrity_violation( + "the local inference endpoint's reply carried no choices", + )); + } + let choice = choices.remove(0); + let mut events = Vec::new(); + if let Some(content) = choice.message.content + && !content.is_empty() + { + events.push(InferenceEvent::Text(content)); + } + for call in choice.message.tool_calls { + if !call_id_is_valid(&call.id) || !tool_name_is_valid(&call.function.name) { + return Err(InferenceFault::integrity_violation( + "the local inference endpoint reported a call id or tool name the provider contract refuses", + )); + } + if !request.tools.iter().any(|tool| tool.name == call.function.name) { + return Err(InferenceFault::integrity_violation( + "the local inference endpoint called a tool this request did not offer", + )); + } + events.push(InferenceEvent::ToolCall { + call_id: call.id, + name: call.function.name, + arguments: call.function.arguments, + }); + } + let receipt = LocalInferenceReceipt { + schema_id: LOCAL_RECEIPT_SCHEMA.to_owned(), + request_id: request.request_id.clone(), + conversation_id: request.conversation_id.clone(), + caller_runtime_id: prepared.invocation.caller_runtime_id.clone(), + native_request_sha256: prepared.invocation.native_request_sha256, + provider_request_sha256: prepared.invocation.provider_request_sha256, + model: request.model.clone(), + response_id: response.id, + finish_reason: choice.finish_reason, + prompt_tokens: response + .usage + .as_ref() + .map(|usage| usage.prompt_tokens) + .unwrap_or_default(), + completion_tokens: response + .usage + .as_ref() + .map(|usage| usage.completion_tokens) + .unwrap_or_default(), + }; + let receipt_bytes = rmp_serde::to_vec_named(&receipt) + .map_err(|error| InferenceFault::new(error.to_string()))?; + Ok(InferenceOutput::new( + events, + format!("sha256:{:x}", Sha256::digest(&receipt_bytes)), + )) +} + +#[async_trait] +impl InferencePort for LocalInferencePort { + fn prepare(&self, request: InferenceRequest) -> Result { + PreparedInference::prepare( + &self.caller_runtime_id, + unix_ms()?.saturating_add(REQUEST_EXPIRY.as_millis() as u64), + request, + ) + } + + async fn infer(&self, request: PreparedInference) -> Result { + if request.invocation.caller_runtime_id != self.caller_runtime_id { + return Err(InferenceFault::integrity_violation( + "persisted inference caller does not match the configured runtime identity", + )); + } + if request.invocation.expires_at_unix_ms <= unix_ms()? { + return Err(InferenceFault::new( + "persisted inference invocation expired before it reached the local endpoint", + )); + } + let body = lower_request(&request, &self.prefix)?; + let response = self.send(body).await?; + assemble_output(&request, response) + } +} + +/// Builds the local port from its binding. The caller validates the endpoint +/// is loopback before this is reached: this constructor does no validation of +/// its own, so it cannot silently open a non-loopback port through a path +/// that skips the check `open_inference` runs. +pub(super) fn open_local_port(binding: LocalBinding) -> Arc { + Arc::new(LocalInferencePort::new( + binding.endpoint, + binding.model_prefix, + binding.caller_runtime_id, + )) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::controllers::{ + ControllerOpenError, InferencePurpose, RequestShape, open_inference, tool_request, + }; + use crate::elaboration::{ElaborationLoopEvaluation, SEED_ROUND_BUDGET, evaluate_elaboration_loop}; + use crate::patch::{RECORD_GAP_PATCH_TOOL, SUBMIT_PATCH_TOOL, patch_tools}; + use crate::sdk_inference::{DEFAULT_SDK_MODEL_PREFIX, RoutedInferencePort, SdkBinding}; + use crate::CommandId; + use codex_connector::CodexToolDefinition; + use std::collections::VecDeque; + use std::sync::Mutex; + use tokio::io::{AsyncReadExt, AsyncWriteExt}; + use tokio::net::{TcpListener, TcpStream}; + + const TEST_RUNTIME: &str = "ghostlight-local-test"; + const TEST_MODEL: &str = "local/llama-8b"; + + /// A raw HTTP/1.1 responder over a loopback `TcpListener`, so the driver + /// is exercised against real sockets without a mock HTTP crate. It plays + /// back one scripted `(status, body)` reply per accepted connection and + /// records each request's header block and body for inspection. + struct ScriptedResponder { + addr: SocketAddr, + state: Arc>, + } + + #[derive(Default)] + struct ScriptedResponderState { + replies: VecDeque<(u16, String)>, + headers: Vec, + bodies: Vec, + } + + impl ScriptedResponder { + async fn start(replies: Vec<(u16, String)>) -> Self { + let listener = TcpListener::bind("127.0.0.1:0").await.expect("loopback binds"); + let addr = listener.local_addr().expect("a bound listener has a local address"); + let state = Arc::new(Mutex::new(ScriptedResponderState { + replies: replies.into(), + ..Default::default() + })); + let worker_state = Arc::clone(&state); + tokio::spawn(async move { + loop { + let Ok((stream, _)) = listener.accept().await else { + return; + }; + serve_one(stream, &worker_state).await; + } + }); + Self { addr, state } + } + + fn endpoint(&self) -> SocketAddr { + self.addr + } + + fn headers(&self) -> Vec { + self.state.lock().expect("the responder mutex is never poisoned").headers.clone() + } + + fn bodies(&self) -> Vec { + self.state.lock().expect("the responder mutex is never poisoned").bodies.clone() + } + } + + async fn serve_one(mut stream: TcpStream, state: &Arc>) { + let mut buffer = Vec::new(); + let mut chunk = [0_u8; 4096]; + let header_end = loop { + let read = stream.read(&mut chunk).await.expect("the test socket reads"); + if read == 0 { + return; + } + buffer.extend_from_slice(&chunk[..read]); + if let Some(pos) = find_double_crlf(&buffer) { + break pos; + } + }; + let header_block = String::from_utf8_lossy(&buffer[..header_end]).into_owned(); + let content_length: usize = header_block + .lines() + .find_map(|line| { + let (name, value) = line.split_once(':')?; + name.eq_ignore_ascii_case("content-length") + .then(|| value.trim().parse().ok()) + .flatten() + }) + .unwrap_or(0); + let body_start = header_end + 4; + while buffer.len() < body_start + content_length { + let read = stream.read(&mut chunk).await.expect("the test socket reads"); + if read == 0 { + break; + } + buffer.extend_from_slice(&chunk[..read]); + } + let body = String::from_utf8_lossy(&buffer[body_start..body_start + content_length]).into_owned(); + + let (status, reply_body) = { + let mut guard = state.lock().expect("the responder mutex is never poisoned"); + guard.headers.push(header_block); + guard.bodies.push(body); + guard.replies.pop_front().unwrap_or((500, "{}".to_owned())) + }; + let reason = match status { + 200 => "OK", + 400 => "Bad Request", + 429 => "Too Many Requests", + 503 => "Service Unavailable", + _ => "Error", + }; + let response = format!( + "HTTP/1.1 {status} {reason}\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{reply_body}", + reply_body.len() + ); + let _ = stream.write_all(response.as_bytes()).await; + let _ = stream.shutdown().await; + } + + fn find_double_crlf(buffer: &[u8]) -> Option { + buffer.windows(4).position(|window| window == b"\r\n\r\n") + } + + fn tool_call_reply(id: &str, name: &str, arguments: &str) -> String { + json!({ + "id": format!("resp-{id}"), + "choices": [{ + "message": { + "role": "assistant", + "content": Value::Null, + "tool_calls": [{ + "id": id, + "type": "function", + "function": {"name": name, "arguments": arguments}, + }], + }, + "finish_reason": "tool_calls", + }], + "usage": {"prompt_tokens": 10, "completion_tokens": 5}, + }) + .to_string() + } + + fn two_tool_call_reply(first: (&str, &str, &str), second: (&str, &str, &str)) -> String { + json!({ + "id": "resp-double", + "choices": [{ + "message": { + "role": "assistant", + "content": Value::Null, + "tool_calls": [ + {"id": first.0, "type": "function", "function": {"name": first.1, "arguments": first.2}}, + {"id": second.0, "type": "function", "function": {"name": second.1, "arguments": second.2}}, + ], + }, + "finish_reason": "tool_calls", + }], + "usage": {"prompt_tokens": 10, "completion_tokens": 5}, + }) + .to_string() + } + + fn text_reply(text: &str) -> String { + json!({ + "id": "resp-text", + "choices": [{ + "message": {"role": "assistant", "content": text, "tool_calls": []}, + "finish_reason": "stop", + }], + "usage": {"prompt_tokens": 3, "completion_tokens": 2}, + }) + .to_string() + } + + fn seed_shape() -> RequestShape { + RequestShape { + max_output_tokens: 4_000, + parallel_tool_calls: false, + } + } + + fn port(endpoint: SocketAddr) -> LocalInferencePort { + LocalInferencePort::new(endpoint, DEFAULT_LOCAL_MODEL_PREFIX, TEST_RUNTIME) + } + + /// Spec test: pins invariant 7. Two rounds through the real seed + /// evaluator: round 0 carries two non-terminal tool calls, round 1 + /// submits, and the second HTTP request literally carries the first + /// round's tool calls and their derived results. + #[tokio::test] + async fn a_scripted_backend_drives_an_elaboration_round_through_the_existing_evaluator() { + let prompt = "Author the shortfall."; + let responder = ScriptedResponder::start(vec![ + ( + 200, + two_tool_call_reply( + ("call-0", RECORD_GAP_PATCH_TOOL, r#"{"detail":"no route"}"#), + ("call-1", RECORD_GAP_PATCH_TOOL, r#"{"detail":"still missing"}"#), + ), + ), + (200, tool_call_reply("call-2", SUBMIT_PATCH_TOOL, "{}")), + ]) + .await; + let port = port(responder.endpoint()); + let command_id = CommandId::new(); + + let round0 = tool_request( + command_id, + 0, + InferencePurpose::Elaboration, + TEST_MODEL, + "Author the shortfall.", + vec![CodexInputItem::UserText { + text: prompt.into(), + }], + patch_tools(), + seed_shape(), + ) + .expect("round 0 builds"); + let prepared0 = port.prepare(round0).expect("round 0 prepares"); + let output0 = port.infer(prepared0).await.expect("round 0 infers"); + assert_eq!( + output0.events.len(), + 2, + "both round-0 tool calls came back inert" + ); + + let ElaborationLoopEvaluation::Continue { conversation } = + evaluate_elaboration_loop(prompt, &[], std::slice::from_ref(&output0), SEED_ROUND_BUDGET) + .expect("round 0 re-derives") + else { + panic!("round 0 completed early"); + }; + + let round1 = tool_request( + command_id, + 1, + InferencePurpose::Elaboration, + TEST_MODEL, + "Author the shortfall.", + conversation, + patch_tools(), + seed_shape(), + ) + .expect("round 1 builds"); + let prepared1 = port.prepare(round1).expect("round 1 prepares"); + let output1 = port.infer(prepared1).await.expect("round 1 infers"); + + let evaluation = + evaluate_elaboration_loop(prompt, &[], &[output0, output1], SEED_ROUND_BUDGET) + .expect("the seed evaluator re-derives"); + let ElaborationLoopEvaluation::Complete { capture } = evaluation else { + panic!("two local inference rounds did not close the seed round") + }; + assert!(capture.submitted, "the re-derived capture did not submit"); + + let bodies = responder.bodies(); + assert_eq!(bodies.len(), 2, "the port made anything but two requests"); + // The second request is Rust's own loop replaying the first round's + // tool calls and their fold-derived results, not the port's. + assert!(bodies[1].contains("no route")); + assert!(bodies[1].contains("still missing")); + assert!(bodies[1].contains("call-0")); + assert!(bodies[1].contains("call-1")); + assert!(bodies[1].contains("gap recorded")); + } + + /// Spec test: pins the message mapping item by item. + #[tokio::test] + async fn the_request_lowers_every_input_item() { + let responder = ScriptedResponder::start(vec![(200, text_reply("acknowledged"))]).await; + let port = port(responder.endpoint()); + let request = tool_request( + CommandId::new(), + 0, + InferencePurpose::Elaboration, + TEST_MODEL, + "Only reply once every item is accounted for.", + vec![ + CodexInputItem::UserText { + text: "Open the gate.".into(), + }, + CodexInputItem::AssistantText { + text: "Checking the ledger.".into(), + }, + CodexInputItem::ToolCall { + call_id: "call-9".into(), + name: RECORD_GAP_PATCH_TOOL.into(), + arguments: r#"{"detail":"no ledger entry"}"#.into(), + }, + CodexInputItem::ToolResult { + call_id: "call-9".into(), + output: "gap recorded".into(), + }, + ], + patch_tools(), + seed_shape(), + ) + .expect("the request builds"); + let prepared = port.prepare(request).expect("the port prepares"); + let output = port.infer(prepared).await.expect("the port infers"); + assert_eq!( + output.events, + vec![InferenceEvent::Text("acknowledged".into())] + ); + + let sent: Value = serde_json::from_str(&responder.bodies()[0]).expect("the body is JSON"); + let messages = sent["messages"].as_array().expect("messages is an array"); + assert_eq!(messages.len(), 5, "instructions plus four input items"); + assert_eq!(messages[0]["role"], "system"); + assert_eq!(messages[1]["role"], "user"); + assert_eq!(messages[1]["content"], "Open the gate."); + assert_eq!(messages[2]["role"], "assistant"); + assert_eq!(messages[2]["content"], "Checking the ledger."); + assert_eq!(messages[3]["role"], "assistant"); + assert_eq!( + messages[3]["tool_calls"][0]["function"]["name"], + RECORD_GAP_PATCH_TOOL + ); + assert_eq!(messages[4]["role"], "tool"); + assert_eq!(messages[4]["tool_call_id"], "call-9"); + assert_eq!(messages[4]["content"], "gap recorded"); + } + + /// Spec test. + #[test] + fn a_non_loopback_endpoint_is_refused_at_open() { + let directory = tempfile::tempdir().unwrap(); + let sdk_entry = directory.path().join("main.js"); + std::fs::write(&sdk_entry, "// sidecar").unwrap(); + let error = open_inference( + None, + Some(SdkBinding { + sidecar_entry: sdk_entry, + caller_runtime_id: TEST_RUNTIME.into(), + model_prefix: DEFAULT_SDK_MODEL_PREFIX.into(), + }), + Some(LocalBinding { + endpoint: "93.184.216.34:443".parse().unwrap(), + model_prefix: DEFAULT_LOCAL_MODEL_PREFIX.into(), + caller_runtime_id: TEST_RUNTIME.into(), + }), + &[TEST_MODEL], + ) + .err() + .expect("a non-loopback local endpoint opened"); + assert!(matches!( + error, + ControllerOpenError::LocalEndpointNotLoopback { .. } + )); + } + + /// Spec test. + #[tokio::test] + async fn a_call_to_an_unoffered_tool_is_an_integrity_violation() { + let responder = + ScriptedResponder::start(vec![(200, tool_call_reply("call-0", "speek", "{}"))]).await; + let port = port(responder.endpoint()); + let request = tool_request( + CommandId::new(), + 0, + InferencePurpose::Elaboration, + TEST_MODEL, + "Author the shortfall.", + vec![CodexInputItem::UserText { + text: "Author the shortfall.".into(), + }], + patch_tools(), + seed_shape(), + ) + .expect("the request builds"); + let prepared = port.prepare(request).expect("the port prepares"); + let fault = port + .infer(prepared) + .await + .expect_err("an unoffered tool call produced an output"); + assert!(fault.integrity_was_violated(), "{fault:?}"); + } + + /// Spec test. `RoutedInferencePort` prefers the longest matching claim + /// among overlapping prefixes, and `open_inference` refuses two slots + /// that claim the identical prefix outright. + #[test] + fn routing_prefers_the_longest_claim_and_refuses_a_shared_prefix() { + let local_port = Arc::new(port("127.0.0.1:1".parse().unwrap())) as Arc; + let sdk_port = Arc::new(port("127.0.0.1:1".parse().unwrap())) as Arc; + let connector_port = Arc::new(port("127.0.0.1:1".parse().unwrap())) as Arc; + let routed = RoutedInferencePort::new( + Some(Arc::clone(&connector_port)), + Some(Arc::clone(&sdk_port)), + "claude", + Some(("claude-local/".to_owned(), Arc::clone(&local_port))), + ); + assert!(Arc::ptr_eq( + routed.route("claude-local/llama-8b").unwrap(), + &local_port + )); + assert!(Arc::ptr_eq(routed.route("claude-opus-5").unwrap(), &sdk_port)); + assert!(Arc::ptr_eq( + routed.route("gpt-5.6-terra").unwrap(), + &connector_port + )); + + let directory = tempfile::tempdir().unwrap(); + let sdk_entry = directory.path().join("main.js"); + std::fs::write(&sdk_entry, "// sidecar").unwrap(); + let error = open_inference( + None, + Some(SdkBinding { + sidecar_entry: sdk_entry, + caller_runtime_id: TEST_RUNTIME.into(), + model_prefix: "claude".into(), + }), + Some(LocalBinding { + endpoint: "127.0.0.1:1".parse().unwrap(), + model_prefix: "claude".into(), + caller_runtime_id: TEST_RUNTIME.into(), + }), + &["claude-opus-5"], + ) + .err() + .expect("two ports sharing one prefix opened"); + assert!(matches!( + error, + ControllerOpenError::SharedModelPrefix { .. } + )); + } + + /// Spec test. The responder asserts on the raw header block, not the + /// decoded request, so a header this port never intended to send cannot + /// hide behind a friendly wrapper's own parsing. + #[tokio::test] + async fn no_request_carries_an_authorization_header() { + let responder = ScriptedResponder::start(vec![(200, text_reply("ok"))]).await; + let port = port(responder.endpoint()); + let request = tool_request( + CommandId::new(), + 0, + InferencePurpose::Persona, + TEST_MODEL, + "Respond only in natural prose.", + vec![CodexInputItem::UserText { + text: "Say something true.".into(), + }], + Vec::::new(), + RequestShape { + max_output_tokens: 1_200, + parallel_tool_calls: false, + }, + ) + .expect("the request builds"); + let prepared = port.prepare(request).expect("the port prepares"); + let output = port.infer(prepared).await.expect("the port infers"); + assert_eq!(output.events, vec![InferenceEvent::Text("ok".into())]); + + let header_block = responder.headers()[0].to_ascii_lowercase(); + assert!(!header_block.contains("authorization")); + assert!(!header_block.contains("bearer")); + assert!(!header_block.contains("api-key")); + assert!(!header_block.contains("api_key")); + } + + /// Spec test (mutation M2.3). Connection refusal and the two status codes + /// the map names are retryable; any other unsuccessful status is + /// recovery-required, not a special case. + #[tokio::test] + async fn connection_refusal_and_429_503_are_retryable_other_statuses_are_recovery_required() { + let listener = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let refused_endpoint = listener.local_addr().unwrap(); + drop(listener); + let refused = port(refused_endpoint); + let request = tool_request( + CommandId::new(), + 0, + InferencePurpose::Persona, + TEST_MODEL, + "Respond only in natural prose.", + vec![CodexInputItem::UserText { + text: "Say something true.".into(), + }], + Vec::::new(), + RequestShape { + max_output_tokens: 1_200, + parallel_tool_calls: false, + }, + ) + .expect("the request builds"); + let prepared = refused.prepare(request.clone()).expect("the port prepares"); + let fault = refused + .infer(prepared) + .await + .expect_err("a closed port accepted a connection"); + assert!( + !fault.requires_recovery(), + "a refused connection was not retryable: {fault:?}" + ); + assert!(!fault.integrity_was_violated()); + + for (status, retryable) in [(429_u16, true), (503, true), (400, false)] { + let responder = ScriptedResponder::start(vec![(status, "{}".to_owned())]).await; + let statused = port(responder.endpoint()); + let prepared = statused.prepare(request.clone()).expect("the port prepares"); + let fault = statused + .infer(prepared) + .await + .expect_err("a non-2xx status produced an output"); + assert!(!fault.integrity_was_violated(), "{fault:?}"); + assert_eq!( + !fault.requires_recovery(), + retryable, + "status {status} disposition mismatch: {fault:?}" + ); + } + } +} diff --git a/crates/ghostlight/src/sdk_inference.rs b/crates/ghostlight/src/sdk_inference.rs index eb0a507..dd57bbb 100644 --- a/crates/ghostlight/src/sdk_inference.rs +++ b/crates/ghostlight/src/sdk_inference.rs @@ -716,6 +716,10 @@ pub(super) struct RoutedInferencePort { connector: Option>, sdk: Option>, sdk_model_prefix: String, + /// The local port's own claimed prefix, carried beside it rather than in a + /// third bare `String` field, so `route` can compare its length against + /// the SDK's claim without a third parallel field to keep in sync. + local: Option<(String, Arc)>, } impl RoutedInferencePort { @@ -723,22 +727,47 @@ impl RoutedInferencePort { connector: Option>, sdk: Option>, sdk_model_prefix: impl Into, + local: Option<(String, Arc)>, ) -> Self { Self { connector, sdk, sdk_model_prefix: sdk_model_prefix.into(), + local, } } - /// The model name carries the transport. A prefixed model that no SDK port - /// claims routes nowhere rather than falling back: a fallback is what would - /// make a typo in a model environment variable silent. + /// The model name carries the transport. A prefixed model that no + /// configured port claims routes nowhere rather than falling back: a + /// fallback is what would make a typo in a model environment variable + /// silent. When more than one prefix matches, the longest claim wins, so a + /// local prefix that extends the SDK's own (`claude-local/` beside + /// `claude`) is unambiguous without either operator having to avoid the + /// other's namespace. pub(super) fn route(&self, model: &str) -> Option<&Arc> { - if model.starts_with(&self.sdk_model_prefix) { - self.sdk.as_ref() - } else { - self.connector.as_ref() + // A prefix claims the model whether or not its port is configured: a + // claimed-but-unconfigured prefix refuses rather than falling back to + // the connector, which is what keeps a typo'd model prefix loud + // instead of silently reaching the wrong backend. + let local_claim = self + .local + .as_ref() + .filter(|(prefix, _)| model.starts_with(prefix.as_str())) + .map(|(prefix, port)| (prefix.len(), port)); + let sdk_claim = model + .starts_with(&self.sdk_model_prefix) + .then_some(self.sdk_model_prefix.len()); + match (local_claim, sdk_claim) { + (Some((local_len, local_port)), Some(sdk_len)) => { + if local_len >= sdk_len { + Some(local_port) + } else { + self.sdk.as_ref() + } + } + (Some((_, local_port)), None) => Some(local_port), + (None, Some(_)) => self.sdk.as_ref(), + (None, None) => self.connector.as_ref(), } } } @@ -1309,14 +1338,19 @@ mod tests { Some(Arc::clone(&connector)), Some(Arc::clone(&sdk)), DEFAULT_SDK_MODEL_PREFIX, + None, ); assert!(Arc::ptr_eq(routed.route("claude-opus-5").unwrap(), &sdk)); assert!(Arc::ptr_eq( routed.route("gpt-5.6-terra").unwrap(), &connector )); - let moved = - RoutedInferencePort::new(Some(Arc::clone(&connector)), Some(Arc::clone(&sdk)), "gpt-"); + let moved = RoutedInferencePort::new( + Some(Arc::clone(&connector)), + Some(Arc::clone(&sdk)), + "gpt-", + None, + ); assert!(Arc::ptr_eq(moved.route("gpt-5.6-terra").unwrap(), &sdk)); assert!(Arc::ptr_eq( moved.route("claude-opus-5").unwrap(), @@ -1354,11 +1388,11 @@ mod tests { }; assert!(matches!( - open_inference(None, Some(sdk()), &models("gpt-5.6-terra")), + open_inference(None, Some(sdk()), None, &models("gpt-5.6-terra").each()), Err(ControllerOpenError::UnroutableModel { .. }) )); assert!(matches!( - open_inference(Some(connector()), None, &models("claude-opus-5")), + open_inference(Some(connector()), None, None, &models("claude-opus-5").each()), Err(ControllerOpenError::UnroutableModel { .. }) )); assert!(matches!( @@ -1369,12 +1403,13 @@ mod tests { caller_runtime_id: TEST_RUNTIME.into(), model_prefix: DEFAULT_SDK_MODEL_PREFIX.into(), }), - &models(TEST_MODEL), + None, + &models(TEST_MODEL).each(), ), Err(ControllerOpenError::SdkSidecarMissing { .. }) )); assert!( - open_inference(Some(connector()), Some(sdk()), &models(TEST_MODEL)).is_ok(), + open_inference(Some(connector()), Some(sdk()), None, &models(TEST_MODEL).each()).is_ok(), "a configuration where every model routes did not open" ); } @@ -1768,6 +1803,7 @@ mod tests { )) as Arc), None, DEFAULT_SDK_MODEL_PREFIX, + None, ); assert!(connector_only.route(TEST_MODEL).is_none()); let prepared = @@ -1798,7 +1834,8 @@ mod tests { caller_runtime_id: TEST_RUNTIME.into(), model_prefix: DEFAULT_SDK_MODEL_PREFIX.into(), }), - &models(TEST_MODEL), + None, + &models(TEST_MODEL).each(), ) .is_ok() );