GameCult/Ghostlight
Watch
9
0
Fork
You've already forked Ghostlight
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Add the local inference port (play agent Cut 2, PA-Q1 A)

Browse source 
A third `InferencePort` beside the connector and the SDK sidecar: one
loopback-only OpenAI-compatible `POST /v1/chat/completions` call per
prepared request, tool calls returned inert for Rust's own evaluator to
drive the next round, exactly as the connector lane already does. No
credential is read or sent; `open_inference` refuses a non-loopback
`LocalBinding` and a model prefix two transports both claim.

`RoutedInferencePort` gains the local slot and routes by longest claimed
prefix, refusing rather than falling back to the connector when a claimed
prefix has no configured port behind it (the same rule the SDK prefix
already had, now made explicit for both). `ControllerModels::each` is
`pub` and `open_inference` takes `models: &[&str]` instead of the whole
struct, so a caller can extend the list with a model the struct does not
carry.

Dungeon's runtime reads `GHOSTLIGHT_LOCAL_ENDPOINT` and
`GHOSTLIGHT_LOCAL_MODEL_PREFIX` and wires the binding through; no other
Dungeon behavior changes in this cut.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-09-22 02:12:10 +02:00
commit 182f8034f7
7 changed files with 985 additions and 34 deletions
Download patch file Download diff file Expand all files Collapse all files

1
Cargo.lock generated
View file

@ -666,6 +666,7 @@ dependencies = [
"codex-connector",
"cultcache-rs",
"ghostlight-persona-projection",
"reqwest",
"rmp-serde",
"serde",
"serde_bytes",

29
crates/ghostlight-dungeon/src/runtime.rs
View file

@ -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(

1
crates/ghostlight/Cargo.toml
View file

@ -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"

61
crates/ghostlight/src/controllers.rs
View file

@ -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<ConnectorBinding>,
sdk: Option<SdkBinding>,
models: &ControllerModels,
local: Option<LocalBinding>,
models: &[&str],
) -> Result<Arc<dyn InferencePort>, ControllerOpenError> {
let connector: Option<Arc<dyn InferencePort>> = 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<dyn InferencePort>)> = 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();

2
crates/ghostlight/src/lib.rs
View file

@ -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::{

860
crates/ghostlight/src/local_inference.rs Normal file
View file

@ -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<String>,
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<LocalChatChoice>,
#[serde(default)]
usage: Option<LocalChatUsage>,
}
#[derive(Debug, Deserialize)]
struct LocalChatChoice {
message: LocalChatMessage,
#[serde(default)]
finish_reason: Option<String>,
}
#[derive(Debug, Deserialize)]
struct LocalChatMessage {
#[serde(default)]
content: Option<String>,
#[serde(default)]
tool_calls: Vec<LocalChatToolCall>,
}
#[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<String>, caller_runtime_id: impl Into<String>) -> 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<LocalChatResponse, InferenceFault> {
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::<LocalChatResponse>().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<Value, InferenceFault> {
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<InferenceOutput, InferenceFault> {
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, InferenceFault> {
PreparedInference::prepare(
&self.caller_runtime_id,
unix_ms()?.saturating_add(REQUEST_EXPIRY.as_millis() as u64),
request,
)
}
async fn infer(&self, request: PreparedInference) -> Result<InferenceOutput, InferenceFault> {
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<dyn InferencePort> {
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<Mutex<ScriptedResponderState>>,
}
#[derive(Default)]
struct ScriptedResponderState {
replies: VecDeque<(u16, String)>,
headers: Vec<String>,
bodies: Vec<String>,
}
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<String> {
self.state.lock().expect("the responder mutex is never poisoned").headers.clone()
}
fn bodies(&self) -> Vec<String> {
self.state.lock().expect("the responder mutex is never poisoned").bodies.clone()
}
}
async fn serve_one(mut stream: TcpStream, state: &Arc<Mutex<ScriptedResponderState>>) {
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<usize> {
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<dyn InferencePort>;
let sdk_port = Arc::new(port("127.0.0.1:1".parse().unwrap())) as Arc<dyn InferencePort>;
let connector_port = Arc::new(port("127.0.0.1:1".parse().unwrap())) as Arc<dyn InferencePort>;
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::<CodexToolDefinition>::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::<CodexToolDefinition>::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:?}"
);
}
}
}

65
crates/ghostlight/src/sdk_inference.rs
View file

@ -716,6 +716,10 @@ pub(super) struct RoutedInferencePort {
connector: Option<Arc<dyn InferencePort>>,
sdk: Option<Arc<dyn InferencePort>>,
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<dyn InferencePort>)>,
}
impl RoutedInferencePort {
@ -723,22 +727,47 @@ impl RoutedInferencePort {
connector: Option<Arc<dyn InferencePort>>,
sdk: Option<Arc<dyn InferencePort>>,
sdk_model_prefix: impl Into<String>,
local: Option<(String, Arc<dyn InferencePort>)>,
) -> 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<dyn InferencePort>> {
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<dyn InferencePort>),
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()
);