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

Compare commits

base: GameCult:main
Branches Tags
GameCult:main
GameCult:readme-front-door
GameCult:codex/ghostlight-release-binding
GameCult:codex/contract-only-provider-request
..
compare: GameCult:codex/contract-only-provider-request
Branches Tags
GameCult:main
GameCult:readme-front-door
GameCult:codex/ghostlight-release-binding
GameCult:codex/contract-only-provider-request
Author SHA1 Message Date
Meta Vens
6945061a0b Preserve safe SSE failure classes 2026-08-25 04:02:38 +02:00
Meta Vens
8c57be4b12 Document contract-only connector surface 2026-08-24 23:46:52 +02:00
Meta Vens
6dc80f6d26 Separate provider contract from connector client 2026-08-24 23:45:44 +02:00
12 changed files with 581 additions and 2393 deletions
Download patch file Download diff file Expand all files Collapse all files

6
AGENTS.md
View file

@ -16,6 +16,12 @@ repair the store. Consumers own native requests, provider lowering, prompts,
schemas, tool execution, retry between passes, interpretation, decisions, and schemas, tool execution, retry between passes, interpretation, decisions, and
state admission. state admission.
Keep the feature boundary exact. `--no-default-features` is the contract-only
surface used where a Mind seals provider request bytes. The default `client`
surface adds only authenticated framing and socket transport. `daemon` adds the
service authorities. Contract-only consumers must not compile client crypto or
socket physiology by accident.
Do not link upstream Codex crates into this package. Idunn freezes the official Do not link upstream Codex crates into this package. Idunn freezes the official
Codex binary as an exact package input. The child receives auth RPC only, never Codex binary as an exact package input. The child receives auth RPC only, never
consumer identity, prompts, tools, provider requests, or model output. consumer identity, prompts, tools, provider requests, or model output.

3
CLAUDE.md
View file

@ -1,3 +0,0 @@
Repository doctrine for this project lives in AGENTS.md, shared with Codex.
@AGENTS.md

56
Cargo.lock generated
View file

@ -145,17 +145,18 @@ dependencies = [
"anyhow", "anyhow",
"base64 0.23.1", "base64 0.23.1",
"chrono", "chrono",
"cultcache-rs", "cultcache-rs 0.1.0 (git+https://github.com/GameCult/cultcache-rs.git?rev=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c)",
"cultnet-rs", "cultnet-rs",
"fs2", "ed25519-dalek",
"libc",
"rand_core 0.6.4", "rand_core 0.6.4",
"rmp-serde", "rmp-serde",
"serde", "serde",
"serde_bytes",
"serde_json", "serde_json",
"sha2", "sha2",
"thiserror", "thiserror",
"ureq", "ureq",
"uuid",
"zeroize", "zeroize",
] ]
@ -203,11 +204,28 @@ dependencies = [
[[package]] [[package]]
name = "cultcache-rs" name = "cultcache-rs"
version = "0.1.0" version = "0.1.0"
source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f"
dependencies = [ dependencies = [
"anyhow", "anyhow",
"chrono", "chrono",
"cultcache-rs-derive", "cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f)",
"fs2",
"rmp-serde",
"serde",
"serde_bytes",
"serde_json",
"uuid",
"windows-sys 0.59.0",
]
[[package]]
name = "cultcache-rs"
version = "0.1.0"
source = "git+https://github.com/GameCult/cultcache-rs.git?rev=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c#fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c"
dependencies = [
"anyhow",
"chrono",
"cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/cultcache-rs.git?rev=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c)",
"fs2", "fs2",
"redb", "redb",
"rmp-serde", "rmp-serde",
@ -221,7 +239,17 @@ dependencies = [
[[package]] [[package]]
name = "cultcache-rs-derive" name = "cultcache-rs-derive"
version = "0.1.0" version = "0.1.0"
source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f"
dependencies = [
"proc-macro2",
"quote",
"syn 2.0.119",
]
[[package]]
name = "cultcache-rs-derive"
version = "0.1.0"
source = "git+https://github.com/GameCult/cultcache-rs.git?rev=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c#fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c"
dependencies = [ dependencies = [
"proc-macro2", "proc-macro2",
"quote", "quote",
@ -231,14 +259,13 @@ dependencies = [
[[package]] [[package]]
name = "cultnet-rs" name = "cultnet-rs"
version = "0.1.0" version = "0.1.0"
source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029" source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f"
dependencies = [ dependencies = [
"aes-gcm", "aes-gcm",
"anyhow", "anyhow",
"base64 0.22.1", "base64 0.22.1",
"chrono", "chrono",
"cultcache-rs", "cultcache-rs 0.1.0 (git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f)",
"ed25519-dalek",
"hmac", "hmac",
"rand", "rand",
"rmp-serde", "rmp-serde",
@ -249,8 +276,6 @@ dependencies = [
"sha2", "sha2",
"socket2", "socket2",
"uuid", "uuid",
"windows-sys 0.61.2",
"zeroize",
] ]
[[package]] [[package]]
@ -1159,15 +1184,6 @@ dependencies = [
"windows-targets", "windows-targets",
] ]
[[package]]
name = "windows-sys"
version = "0.61.2"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "ae137229bcbd6cdf0f7b80a31df61766145077ddf49416a728b02cb3921ff3fc"
dependencies = [
"windows-link",
]
[[package]] [[package]]
name = "windows-targets" name = "windows-targets"
version = "0.52.6" version = "0.52.6"

28
Cargo.toml
View file

@ -17,31 +17,39 @@ path = "src/main.rs"
required-features = ["daemon"] required-features = ["daemon"]
[features] [features]
default = [] default = ["client"]
client = [
"dep:aes-gcm",
"dep:rand_core",
"dep:zeroize",
]
daemon = [ daemon = [
"client",
"dep:anyhow", "dep:anyhow",
"dep:chrono", "dep:chrono",
"dep:cultcache-rs", "dep:cultcache-rs",
"dep:cultnet-rs", "dep:cultnet-rs",
"dep:fs2", "dep:ed25519-dalek",
"dep:libc", "dep:serde_bytes",
"dep:ureq", "dep:ureq",
"dep:uuid",
] ]
[dependencies] [dependencies]
aes-gcm = "0.10" aes-gcm = { version = "0.10", optional = true }
anyhow = { version = "1", optional = true } anyhow = { version = "1", optional = true }
base64 = "0.23" base64 = "0.23"
chrono = { version = "0.4", features = ["serde"], optional = true } chrono = { version = "0.4", features = ["serde"], optional = true }
cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultcache-rs", optional = true } cultcache-rs = { git = "https://github.com/GameCult/cultcache-rs.git", rev = "fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c", optional = true }
cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultnet-rs", optional = true } cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "75c180782aeba7cfd22d6412877397708a4ed28f", optional = true }
fs2 = { version = "0.4", optional = true } ed25519-dalek = { version = "2.2", optional = true }
libc = { version = "0.2", optional = true } rand_core = { version = "0.6", features = ["getrandom"], optional = true }
rand_core = { version = "0.6", features = ["getrandom"] }
rmp-serde = "1" rmp-serde = "1"
serde = { version = "1", features = ["derive"] } serde = { version = "1", features = ["derive"] }
serde_bytes = { version = "0.11", optional = true }
serde_json = "1" serde_json = "1"
sha2 = "0.10" sha2 = "0.10"
thiserror = "2" thiserror = "2"
ureq = { version = "3.4", default-features = false, features = ["rustls"], optional = true } ureq = { version = "3.4", default-features = false, features = ["rustls"], optional = true }
zeroize = "1" uuid = { version = "1", features = ["v4"], optional = true }
zeroize = { version = "1", optional = true }

75
README.md
View file

@ -5,14 +5,6 @@ subscription inference. Epiphany and Ghostlight remain separate minds. They
derive their own exact provider requests, execute their own tools, interpret derive their own exact provider requests, execute their own tools, interpret
their own results, and admit their own state. their own results, and admit their own state.
**Valuable here:** the consumer ABI in `src/lib.rs`. It holds the v2 typed contracts (`CodexProviderRequest`, `CodexTransportInvocation`, `CodexTransportReceipt`), exact Responses lowering (`canonical_responses_body`), AES-256-GCM envelopes (`encrypt_invocation`, `decrypt_result`) and the one-request client (`CodexConnectorClient`). The daemon behind it (`src/daemon.rs`, `src/provider_backend.rs`) is the only process that ever holds the Codex credential.
**Problem it solves:** several minds (Epiphany, Ghostlight) need Codex-compatible subscription inference, but must not each hold or refresh the credential, copy wire law, or hand cognition to a transport service. The connector isolates the credential and transports exact provider requests with receipts, and decides nothing about prompts or tools.
**Entry points:** library `src/lib.rs` (no daemon code, safe to depend on); daemon feature `--features daemon`, binary `codex-connector` (`src/main.rs`); Idunn declaration `deployment/idunn/recipe.toml`, deployed with `idunn up codex-connector`; verification commands are listed below.
**Status:** live as the library plus the daemon and its Idunn recipe, last changed 2026-09-13. This repo records no planned work; changes to the contract below are the open surface.
The service owns authenticated caller admission, bounded transport physiology, The service owns authenticated caller admission, bounded transport physiology,
exact provider-request receipts, and redacted CultMesh health. A private pinned exact provider-request receipts, and redacted CultMesh health. A private pinned
official `codex app-server` child is the sole writer and refresh authority for official `codex app-server` child is the sole writer and refresh authority for
@ -20,11 +12,13 @@ one credential store. The public connector reads the credential only after the
child's private auth RPC completes, then performs the exact Responses request. child's private auth RPC completes, then performs the exact Responses request.
It owns no prompt policy or decision authority. It owns no prompt policy or decision authority.
The default library is the shared consumer ABI: v2 typed contracts, exact The library has three explicit compile surfaces. With no features it exposes
provider lowering, encryption, bounded direct-pipe framing, one-request socket only the v2 provider/result contracts and their deterministic validation. The
transport, and exact result validation. It does not compile CultCache, HTTP, default `client` feature adds encryption, bounded direct-pipe framing, and the
TLS, provider auth, replay, or daemon code. Epiphany and Ghostlight consume this one-request socket client. It still compiles no CultCache, HTTP, TLS, provider
surface instead of copying wire law. auth, replay, or daemon code. The `daemon` feature includes the client surface
and the service. Epiphany and Ghostlight can therefore type and seal their exact
provider requests without importing transport physiology into their Minds.
Connection establishment is bounded independently. Response read/write timeout Connection establishment is bounded independently. Response read/write timeout
is optional so a consumer's typed outer pass may remain the sole deadline owner. is optional so a consumer's typed outer pass may remain the sole deadline owner.
@ -53,62 +47,21 @@ No upstream Codex crate is linked into the package. The official binary is a
pinned deployment input, keeping Codex's application build graph outside both pinned deployment input, keeping Codex's application build graph outside both
consumers and this small daemon. consumers and this small daemon.
Focused verification uses the two real build surfaces: Focused verification proves each real build surface independently:
```text ```text
cargo test --lib --no-default-features
cargo test --lib cargo test --lib
cargo test --lib --features daemon cargo test --lib --features daemon
cargo check --bin codex-connector --features daemon cargo check --bin codex-connector --features daemon
``` ```
The same production binary can initialize a typed single-caller `.cc` The same production binary can initialize a typed single-caller `.cc`
configuration for first deployment, then serve from that state. Outside an configuration for first deployment, then serve only from that admitted state.
Idunn-managed launch the configuration's loopback bind remains available for The daemon publishes private-free signed health to Idunn from its own runtime
local operation. In a managed launch `GAMECULT_IDUNN_CANDIDATE_BIND` is the identity. Its source revision also owns the exact Codex Linux package and
effective private bind. Idunn admits the candidate and controls route promotion; binary hashes in `deployment/codex-linux-x64.manifest`, so changing the provider
the Connector only binds and serves its private endpoint. toolchain necessarily changes the connector release selected by Idunn.
The repo-visible Idunn declaration is
`deployment/idunn/recipe.toml`. A normal deployment is `idunn up
codex-connector`. The declaration pins the Codex package and extracted binary,
builds the one daemon, declares its process-bound replay and Codex credential
state, and requires a compatible Odin rendezvous capability before start.
`CODEX_CONNECTOR_CULTNET_RUDP` is only the current transport address for that
rendezvous. The selected managed dependency must carry an explicit `rudp://`,
`udp://`, or deliberately bare socket endpoint, and both addresses must agree.
Idunn supplies a root-owned `GAMECULT_IDUNN_RUNTIME_BUNDLE` containing the exact
typed Expected and activation records. Systemd opens the root-owned activation
and stable-identity sources and passes exactly two named descriptors,
`gamecult-idunn-runtime-activation-key` and
`gamecult-runtime-presence-identity`, to the Connector parent. Connector marks
both close-on-exec, consumes and closes them before spawning the Codex
app-server, and strips systemd's descriptor environment from that child.
Connector publishes `gamecult.runtime_presence_health.v2` through CultNet with
both the stable provider proof and the activation-scoped proof. Target, plan,
incarnation, release, runtime, endpoint, health contract, state lineage, and
capability claims are bound to Expected rather than reconstructed from service
configuration. The binary pins the canonical two-slot state-contract digest, so
an Expected projection for a different recipe cannot be attested under the same
generation label.
The managed command receives Idunn's admitted state root as an explicit launch
binding. Connector derives `codex-home` and `replay.cc` from that root and
rejects configuration that names a different layout. The managed configuration
and its CultCache lock must themselves be root-custodied read-only files; they
may describe caller admission and fixed runtime bounds, but cannot mutate those
inputs behind an admitted incarnation.
The candidate socket is bound before the initial `warming` presence so the
reported endpoint is an observed service fact. Connector then waits for the
root-owned typed process-write lease that names that exact warming presence
before opening replay state or starting the official child that may refresh
`codex-home/auth.json`. It holds the lease's pre-created sibling lock shared for
the lifetime of both writers, so Idunn stops the process before moving write
authority to another incarnation. Periodic `active` or `degraded` presence is
factual provider observation and carries the exact current lease digest. Odin
derives Present and Ready by correlating these statements; Idunn decides route
admission and lifecycle actuation.
## Contract invariant ## Contract invariant

4
deployment/codex-linux-x64.manifest Normal file
View file

@ -0,0 +1,4 @@
schema_version=gamecult.codex_connector.codex_toolchain.v1
package_url=https://registry.npmjs.org/@openai/codex/-/codex-0.150.0-alpha.7-linux-x64.tgz
package_sha256=8aca3112aef60127d4ffaf4c6f08e524fa7315279a1d08e7aa9c5e7f1a6943e0
binary_sha256=aca268dc02dcef8b1ea9f528a10173d5071ae76d659a03cce6458bbebe228bee

122
deployment/idunn/recipe.toml
View file

@ -1,122 +0,0 @@
schema = "gamecult.idunn.target_declaration.v1"
target = "codex-connector"
source_stamp_environment = "CODEX_CONNECTOR_BUILD_COMMIT"
[[steps]]
id = "extract-codex"
phase = "prepare"
runner = "rust-build"
argv = [
"tar",
"--extract",
"--gzip",
"--file",
".idunn-inputs/codex-linux-x64.tgz",
"--strip-components=4",
"package/vendor/x86_64-unknown-linux-musl/bin/codex",
]
[[steps]]
id = "test-contract"
phase = "test"
runner = "rust-build"
argv = ["cargo", "test", "--locked"]
[[steps]]
id = "test-daemon"
phase = "test"
runner = "rust-build"
argv = ["cargo", "test", "--locked", "--features", "daemon"]
[[steps]]
id = "build-daemon"
phase = "build"
runner = "rust-build"
argv = [
"cargo",
"build",
"--locked",
"--release",
"--features",
"daemon",
"--bin",
"codex-connector",
]
[[external_inputs]]
id = "codex-linux-x64"
url = "https://registry.npmjs.org/@openai/codex/-/codex-0.150.0-alpha.7-linux-x64.tgz"
sha256 = "8aca3112aef60127d4ffaf4c6f08e524fa7315279a1d08e7aa9c5e7f1a6943e0"
runner = "rust-build"
destination = ".idunn-inputs/codex-linux-x64.tgz"
[[artifacts]]
id = "codex-connector"
source_kind = "runner-output"
runner = "rust-build"
source = "target/release/codex-connector"
destination = "codex-connector"
executable = true
[[artifacts]]
id = "codex"
source_kind = "runner-output"
runner = "rust-build"
source = "codex"
destination = "codex"
expected_sha256 = "aca268dc02dcef8b1ea9f528a10173d5071ae76d659a03cce6458bbebe228bee"
executable = true
[service]
executable_artifact = "codex-connector"
required_adjacent_artifacts = ["codex"]
arguments = [
{ kind = "literal", value = "--config" },
{ kind = "binding", name = "config_path" },
{ kind = "literal", value = "--state-root" },
{ kind = "binding", name = "state_root" },
]
transport = "tcp"
route_required = true
required_environment = [
"CODEX_CONNECTOR_CULTNET_RUDP",
"GAMECULT_IDUNN_CANDIDATE_BIND",
"GAMECULT_IDUNN_PROCESS_WRITE_LEASE",
"GAMECULT_IDUNN_RUNTIME_BUNDLE",
]
[service.health]
contract = "codex-connector.runtime-health.v1"
[state]
schema_generation = "connector-v1"
[[state.slots]]
id = "replay"
relative_path = "replay.cc"
kind = "cultcache-file"
schema = "gamecult.codex.replay_record.v1"
writer = "process-bound-single-writer"
recovery = "preserve"
startup = "create-or-open-after-write-lease"
[[state.slots]]
id = "codex-auth"
relative_path = "codex-home/auth.json"
kind = "external-file"
schema = "openai.codex.auth-json.v1"
writer = "process-bound-single-writer"
recovery = "preserve"
startup = "create-or-open-after-write-lease"
[[provides]]
capability = "gamecult.codex.subscription-inference"
schema = "gamecult.codex.transport_envelope.v2"
compatibility = "v2"
[[dependencies]]
kind = "bootstrap"
capability = "odin.verse-rendezvous"
schema = "odin.verse-topology.v1"
compatibility = "v1"
startup = "before-start"

805
src/daemon.rs
View file

@ -7,20 +7,11 @@ use std::sync::{Arc, Mutex};
use std::thread; use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH}; use std::time::{Duration, SystemTime, UNIX_EPOCH};
#[cfg(target_os = "linux")]
use std::os::unix::fs::MetadataExt;
use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore}; use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use thiserror::Error; use thiserror::Error;
use zeroize::Zeroizing; use zeroize::Zeroizing;
#[cfg(target_os = "linux")]
use cultnet_rs::{
CultNetMessage, CultNetWireContract, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA,
decode_cultnet_message_from_slice, encode_cultnet_message_to_vec,
};
use crate::{ use crate::{
CodexAppServerConfig, CodexCallerAdmission, CodexProviderBackend, CodexProviderBackendError, CodexAppServerConfig, CodexCallerAdmission, CodexProviderBackend, CodexProviderBackendError,
CodexTransportAdmission, CodexTransportEnvelope, CodexTransportService, ServiceError, CodexTransportAdmission, CodexTransportEnvelope, CodexTransportService, ServiceError,
@ -119,24 +110,18 @@ impl CodexDaemonConfig {
{ {
return Err(CodexDaemonError::InvalidConfig("runtime bounds")); return Err(CodexDaemonError::InvalidConfig("runtime bounds"));
} }
let required_connections = required_connection_capacity(&self.callers)?;
if self.max_connections < required_connections {
return Err(CodexDaemonError::InvalidConfig("connection capacity"));
}
let mut caller_ids = HashSet::new(); let mut caller_ids = HashSet::new();
for caller in &self.callers { for caller in &self.callers {
let mut allowed_models = HashSet::new();
if caller.caller_runtime_id.trim().is_empty() if caller.caller_runtime_id.trim().is_empty()
|| caller.caller_runtime_id.trim() != caller.caller_runtime_id || caller.caller_runtime_id.trim() != caller.caller_runtime_id
|| !caller_ids.insert(caller.caller_runtime_id.as_str()) || !caller_ids.insert(caller.caller_runtime_id.as_str())
|| caller.connection_key_file.as_os_str().is_empty() || caller.connection_key_file.as_os_str().is_empty()
|| caller.connection_key_epoch == 0 || caller.connection_key_epoch == 0
|| caller.allowed_models.is_empty() || caller.allowed_models.is_empty()
|| caller.allowed_models.iter().any(|model| { || caller
model.trim().is_empty() .allowed_models
|| model.trim() != model .iter()
|| !allowed_models.insert(model.as_str()) .any(|model| model.trim().is_empty() || model.trim() != model)
})
|| caller.max_concurrent_requests == 0 || caller.max_concurrent_requests == 0
|| caller.max_payload_bytes < 4096 || caller.max_payload_bytes < 4096
|| caller.max_payload_bytes > self.max_frame_bytes - FRAME_OVERHEAD_BUDGET || caller.max_payload_bytes > self.max_frame_bytes - FRAME_OVERHEAD_BUDGET
@ -147,40 +132,6 @@ impl CodexDaemonConfig {
} }
Ok(bind) Ok(bind)
} }
pub fn admit_caller(&mut self, caller: CodexCallerConfig) -> Result<(), CodexDaemonError> {
let mut candidate = self.clone();
if candidate
.callers
.iter()
.any(|existing| existing.caller_runtime_id == caller.caller_runtime_id)
{
return Err(CodexDaemonError::InvalidConfig("duplicate caller"));
}
candidate.max_frame_bytes = candidate.max_frame_bytes.max(
caller
.max_payload_bytes
.saturating_add(FRAME_OVERHEAD_BUDGET),
);
candidate.callers.push(caller);
candidate.max_connections = required_connection_capacity(&candidate.callers)?;
candidate.validate()?;
*self = candidate;
Ok(())
}
}
fn required_connection_capacity(callers: &[CodexCallerConfig]) -> Result<usize, CodexDaemonError> {
Ok(caller_request_capacity(callers)?.max(8))
}
fn caller_request_capacity(callers: &[CodexCallerConfig]) -> Result<usize, CodexDaemonError> {
callers
.iter()
.try_fold(0_usize, |total, caller| {
total.checked_add(caller.max_concurrent_requests)
})
.ok_or(CodexDaemonError::InvalidConfig("connection capacity"))
} }
pub fn load_daemon_config(path: &Path) -> Result<CodexDaemonConfig, CodexDaemonError> { pub fn load_daemon_config(path: &Path) -> Result<CodexDaemonConfig, CodexDaemonError> {
@ -199,30 +150,6 @@ pub fn load_daemon_config(path: &Path) -> Result<CodexDaemonConfig, CodexDaemonE
Ok(config) Ok(config)
} }
fn load_daemon_config_read_only(path: &Path) -> Result<CodexDaemonConfig, CodexDaemonError> {
let store = SingleFileMessagePackBackingStore::new(path);
let config = store
.with_read_only_shared_snapshot(|entries| {
let [envelope] = entries.as_slice() else {
anyhow::bail!("connector configuration must contain exactly one record");
};
if envelope.key != CONFIG_KEY
|| envelope.r#type != CodexDaemonConfig::TYPE
|| envelope.schema_id.as_deref() != Some(CodexDaemonConfig::TYPE)
{
anyhow::bail!("connector configuration has the wrong typed envelope");
}
let config: CodexDaemonConfig = rmp_serde::from_slice(&envelope.payload)?;
if rmp_serde::to_vec(&config)? != envelope.payload {
anyhow::bail!("connector configuration is not canonical positional MessagePack");
}
Ok(config)
})
.map_err(config_store_error)?;
config.validate()?;
Ok(config)
}
pub fn write_daemon_config( pub fn write_daemon_config(
path: &Path, path: &Path,
config: &CodexDaemonConfig, config: &CodexDaemonConfig,
@ -240,109 +167,37 @@ pub fn write_daemon_config(
Ok(()) Ok(())
} }
pub fn serve( pub fn serve(config_path: &Path) -> Result<(), CodexDaemonError> {
config_path: &Path, let config = load_daemon_config(config_path)?;
managed_state_root: Option<&Path>, let bind = config.validate()?;
) -> Result<(), CodexDaemonError> { let service = Arc::new(Mutex::new(CodexTransportService::open(
#[cfg(target_os = "linux")] &config.replay_store,
if std::env::var_os(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).is_some() { load_admissions(&config)?,
crate::idunn_health::require_managed_config_store(config_path) config.max_expiry_skew_ms,
.map_err(CodexDaemonError::Health)?; )?));
}
let config = load_daemon_config_read_only(config_path)?;
let configured_bind = config.validate()?;
#[cfg(target_os = "linux")]
let bind = effective_bind_from_environment(configured_bind)?;
#[cfg(not(target_os = "linux"))]
let bind = configured_bind;
#[cfg(target_os = "linux")]
let (mut codex_executable, mut codex_home, mut replay_store) =
effective_runtime_paths_from_environment(&config, managed_state_root)?;
#[cfg(not(target_os = "linux"))]
let (codex_executable, codex_home, replay_store) = match managed_state_root {
None => (
config.codex_executable.clone(),
config.codex_home.clone(),
config.replay_store.clone(),
),
Some(_) => {
return Err(CodexDaemonError::InvalidConfig(
"Idunn state root on an unmanaged platform",
));
}
};
let admissions = load_admissions(&config)?;
#[cfg(target_os = "linux")]
// Consume the systemd-owned signer descriptors before opening another
// process descriptor. Their numeric 3/4 contract is valid only at the
// managed process entry boundary.
let mut prepared_presence_publisher = prepare_runtime_presence_publisher(bind, &config)?;
// Binding the candidate socket is service physiology. Idunn owns whether
// that candidate receives traffic, so Connector carries no second route
// or traffic-admission gate.
let listener = TcpListener::bind(bind).map_err(CodexDaemonError::Listen)?;
let bound = listener.local_addr().map_err(CodexDaemonError::Listen)?;
#[cfg(target_os = "linux")]
if let Some(publisher) = prepared_presence_publisher.as_mut() {
if bound != bind {
return Err(CodexDaemonError::InvalidConfig("Idunn candidate bind"));
}
let warming_sha256 = publisher
.publish("warming", "awaiting-process-write-lease")
.map_err(CodexDaemonError::Health)?;
publisher
.acquire_process_write_lease(&warming_sha256, Duration::from_secs(120))
.map_err(CodexDaemonError::Health)?;
// The incumbent is fenced only when Idunn can replace the lease under
// our shared lock. Re-observe every write-capable path after that
// handoff so a pre-lease filesystem shape cannot survive into Active.
(codex_executable, codex_home, replay_store) =
effective_runtime_paths_from_environment(&config, managed_state_root)?;
}
#[cfg(target_os = "linux")]
// Keep the shared lease lock in the main service body as well as the
// publisher thread. A health-thread failure must not release write
// authority while replay or the official credential writer remains live.
let _process_write_lease_guard = prepared_presence_publisher
.as_ref()
.map(|publisher| publisher.process_write_lease_guard())
.transpose()
.map_err(CodexDaemonError::Health)?;
// Opening replay state is the first persistent write-capable actuation.
// Managed launches reach this point only after the exact warming process
// has received Idunn's typed process-write lease.
let backend = Arc::new(CodexProviderBackend::start(CodexAppServerConfig { let backend = Arc::new(CodexProviderBackend::start(CodexAppServerConfig {
executable: codex_executable, executable: config.codex_executable.clone(),
executable_sha256: config.codex_executable_sha256, executable_sha256: config.codex_executable_sha256,
codex_home, codex_home: config.codex_home.clone(),
max_result_bytes: config.max_frame_bytes - FRAME_OVERHEAD_BUDGET, max_result_bytes: config.max_frame_bytes - FRAME_OVERHEAD_BUDGET,
})?); })?);
let readiness = backend.readiness()?; let readiness = backend.readiness()?;
let service = Arc::new(Mutex::new(CodexTransportService::open( let listener = TcpListener::bind(bind).map_err(CodexDaemonError::Listen)?;
&replay_store,
admissions,
config.max_expiry_skew_ms,
)?));
#[cfg(target_os = "linux")]
let runtime_presence_publisher =
prepared_presence_publisher.map(|publisher| Arc::new(Mutex::new(publisher)));
#[cfg(target_os = "linux")]
if let Some(publisher) = runtime_presence_publisher.as_ref() {
start_periodic_runtime_presence_publisher(publisher.clone(), backend.clone())?;
}
let active_connections = Arc::new(AtomicUsize::new(0)); let active_connections = Arc::new(AtomicUsize::new(0));
eprintln!( eprintln!(
"codex-connector ready at {} for {} callers using {:?}", "codex-connector ready at {} for {} callers using {:?}",
bound, listener.local_addr().map_err(CodexDaemonError::Listen)?,
config.callers.len(), config.callers.len(),
readiness.auth_mode readiness.auth_mode
); );
#[cfg(target_os = "linux")]
start_idunn_health_publisher()?;
loop { for accepted in listener.incoming() {
let (stream, _) = listener.accept().map_err(CodexDaemonError::Listen)?; let stream = match accepted {
Ok(stream) => stream,
Err(error) => return Err(CodexDaemonError::Listen(error)),
};
if !stream if !stream
.peer_addr() .peer_addr()
.map_err(CodexDaemonError::Connection)? .map_err(CodexDaemonError::Connection)?
@ -358,66 +213,49 @@ pub fn serve(
}; };
let service = service.clone(); let service = service.clone();
let backend = backend.clone(); let backend = backend.clone();
#[cfg(target_os = "linux")]
let runtime_presence_publisher = runtime_presence_publisher.clone();
let max_frame_bytes = config.max_frame_bytes; let max_frame_bytes = config.max_frame_bytes;
let timeout = Duration::from_millis(config.socket_timeout_ms); let timeout = Duration::from_millis(config.socket_timeout_ms);
thread::Builder::new() thread::Builder::new()
.name("codex-connector-request".to_string()) .name("codex-connector-request".to_string())
.spawn(move || { .spawn(move || {
let _permit = permit; let _permit = permit;
let _ = serve_connection( let _ = serve_connection(stream, &service, &backend, max_frame_bytes, timeout);
stream,
&service,
&backend,
#[cfg(target_os = "linux")]
runtime_presence_publisher.as_deref(),
max_frame_bytes,
timeout,
);
}) })
.map_err(CodexDaemonError::Thread)?; .map_err(CodexDaemonError::Thread)?;
} }
Err(CodexDaemonError::ListenerClosed)
} }
#[cfg(target_os = "linux")] #[cfg(target_os = "linux")]
fn prepare_runtime_presence_publisher( fn start_idunn_health_publisher() -> Result<(), CodexDaemonError> {
candidate: SocketAddr, let Some(endpoint) = std::env::var_os("CODEX_CONNECTOR_IDUNN_RUDP") else {
config: &CodexDaemonConfig, return Ok(());
) -> Result<Option<crate::idunn_health::RuntimePresencePublisher>, CodexDaemonError> { };
if std::env::var_os(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).is_none() { let endpoint = endpoint
return Ok(None); .to_string_lossy()
} .parse::<SocketAddr>()
let capacity = u32::try_from(caller_request_capacity(&config.callers)?) .map_err(|_| CodexDaemonError::InvalidConfig("Idunn health endpoint"))?;
.map_err(|_| CodexDaemonError::InvalidConfig("runtime capability capacity"))?; let daemon_id = required_environment("CODEX_CONNECTOR_IDUNN_DAEMON")?;
let publisher = let runtime_id = required_environment("CODEX_CONNECTOR_RUNTIME_ID")?;
crate::idunn_health::RuntimePresencePublisher::open_from_environment(candidate, capacity) let contract = std::env::var("CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT")
.unwrap_or_else(|_| crate::CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT.to_string());
let identity = PathBuf::from(required_environment(
"CODEX_CONNECTOR_IDUNN_HEALTH_IDENTITY",
)?);
let mut publisher =
crate::ProviderHealthPublisher::open(endpoint, daemon_id, runtime_id, contract, &identity)
.map_err(CodexDaemonError::Health)?; .map_err(CodexDaemonError::Health)?;
Ok(Some(publisher))
}
#[cfg(target_os = "linux")]
fn start_periodic_runtime_presence_publisher(
publisher: Arc<Mutex<crate::idunn_health::RuntimePresencePublisher>>,
backend: Arc<CodexProviderBackend>,
) -> Result<(), CodexDaemonError> {
let (state, detail) = provider_health_observation(&backend);
publisher publisher
.lock() .publish("active", "credential-isolated-transport-ready")
.map_err(|_| CodexDaemonError::RuntimePresencePoisoned)?
.publish(state, detail)
.map_err(CodexDaemonError::Health)?; .map_err(CodexDaemonError::Health)?;
thread::Builder::new() thread::Builder::new()
.name("codex-connector-health".to_string()) .name("codex-connector-health".to_string())
.spawn(move || { .spawn(move || {
loop { loop {
thread::sleep(Duration::from_secs(10)); thread::sleep(Duration::from_secs(10));
let (state, detail) = provider_health_observation(&backend); if let Err(error) =
let publication = publisher publisher.publish("active", "credential-isolated-transport-ready")
.lock() {
.map_err(|_| anyhow::anyhow!("runtime presence publisher lock was poisoned"))
.and_then(|mut publisher| publisher.publish(state, detail));
if let Err(error) = publication {
eprintln!("codex-connector health publication failed: {error}"); eprintln!("codex-connector health publication failed: {error}");
} }
} }
@ -427,153 +265,25 @@ fn start_periodic_runtime_presence_publisher(
} }
#[cfg(target_os = "linux")] #[cfg(target_os = "linux")]
fn provider_health_observation(backend: &CodexProviderBackend) -> (&'static str, &'static str) { fn required_environment(name: &'static str) -> Result<String, CodexDaemonError> {
provider_health_state(backend.readiness().is_ok()) std::env::var(name)
} .ok()
.filter(|value| !value.trim().is_empty() && value == value.trim())
#[cfg(target_os = "linux")] .ok_or(CodexDaemonError::InvalidConfig(name))
fn provider_health_state(ready: bool) -> (&'static str, &'static str) {
if ready {
("active", "credential-isolated-transport-ready")
} else {
("degraded", "provider-backend-unready")
}
}
#[cfg(target_os = "linux")]
fn effective_bind_from_environment(configured: SocketAddr) -> Result<SocketAddr, CodexDaemonError> {
select_effective_bind(
configured,
std::env::var(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).ok(),
std::env::var(crate::idunn_health::IDUNN_CANDIDATE_BIND_ENVIRONMENT).ok(),
)
}
#[cfg(target_os = "linux")]
fn effective_runtime_paths_from_environment(
config: &CodexDaemonConfig,
managed_state_root: Option<&Path>,
) -> Result<(PathBuf, PathBuf, PathBuf), CodexDaemonError> {
if std::env::var_os(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).is_none() {
if managed_state_root.is_some() {
return Err(CodexDaemonError::InvalidConfig(
"Idunn state root outside a managed launch",
));
}
return Ok((
config.codex_executable.clone(),
config.codex_home.clone(),
config.replay_store.clone(),
));
}
let managed_state_root =
managed_state_root.ok_or(CodexDaemonError::InvalidConfig("missing Idunn state root"))?;
let running = std::env::current_exe()
.and_then(fs::canonicalize)
.map_err(|_| CodexDaemonError::InvalidConfig("managed Connector executable"))?;
managed_runtime_paths(config, &running, managed_state_root)
}
#[cfg(target_os = "linux")]
fn managed_runtime_paths(
config: &CodexDaemonConfig,
running: &Path,
state_root: &Path,
) -> Result<(PathBuf, PathBuf, PathBuf), CodexDaemonError> {
let adjacent_codex = running
.parent()
.ok_or(CodexDaemonError::InvalidConfig(
"managed Connector executable",
))?
.join("codex");
let adjacent_codex = fs::canonicalize(adjacent_codex)
.map_err(|_| CodexDaemonError::InvalidConfig("managed Codex executable"))?;
let configured_codex = fs::canonicalize(&config.codex_executable)
.map_err(|_| CodexDaemonError::InvalidConfig("managed Codex executable"))?;
if configured_codex != adjacent_codex {
return Err(CodexDaemonError::InvalidConfig("managed Codex executable"));
}
if !state_root.is_absolute() {
return Err(CodexDaemonError::InvalidConfig("managed state root"));
}
let canonical_state_root = fs::canonicalize(state_root)
.map_err(|_| CodexDaemonError::InvalidConfig("managed state root"))?;
let state_root_metadata = fs::symlink_metadata(state_root)
.map_err(|_| CodexDaemonError::InvalidConfig("managed state root"))?;
if canonical_state_root != state_root
|| !state_root_metadata.is_dir()
|| state_root_metadata.file_type().is_symlink()
{
return Err(CodexDaemonError::InvalidConfig("managed state root"));
}
let codex_home = state_root.join("codex-home");
if config.codex_home != codex_home {
return Err(CodexDaemonError::InvalidConfig("managed state layout"));
}
match fs::symlink_metadata(&codex_home) {
Ok(metadata) if metadata.is_dir() && !metadata.file_type().is_symlink() => {}
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
_ => return Err(CodexDaemonError::InvalidConfig("managed state layout")),
}
let replay_store = state_root.join("replay.cc");
if config.replay_store != replay_store {
return Err(CodexDaemonError::InvalidConfig("managed state layout"));
}
match fs::symlink_metadata(&replay_store) {
Ok(metadata)
if metadata.is_file()
&& !metadata.file_type().is_symlink()
&& metadata.nlink() == 1 => {}
Ok(_) => return Err(CodexDaemonError::InvalidConfig("managed state layout")),
Err(error) if error.kind() == std::io::ErrorKind::NotFound => {}
Err(_) => return Err(CodexDaemonError::InvalidConfig("managed state layout")),
}
Ok((adjacent_codex, codex_home, replay_store))
}
#[cfg(target_os = "linux")]
fn select_effective_bind(
configured: SocketAddr,
runtime_bundle: Option<String>,
candidate_bind: Option<String>,
) -> Result<SocketAddr, CodexDaemonError> {
match (runtime_bundle, candidate_bind) {
(None, None) => Ok(configured),
(Some(bundle), Some(candidate))
if !bundle.trim().is_empty()
&& bundle.trim() == bundle
&& candidate.trim() == candidate =>
{
let bind = candidate
.parse::<SocketAddr>()
.map_err(|_| CodexDaemonError::InvalidConfig("Idunn candidate bind"))?;
if !bind.ip().is_loopback() {
return Err(CodexDaemonError::InvalidConfig("Idunn candidate bind"));
}
Ok(bind)
}
_ => Err(CodexDaemonError::InvalidConfig(
"partial Idunn managed runtime environment",
)),
}
} }
fn load_admissions( fn load_admissions(
config: &CodexDaemonConfig, config: &CodexDaemonConfig,
) -> Result<Vec<CodexCallerAdmission>, CodexDaemonError> { ) -> Result<Vec<CodexCallerAdmission>, CodexDaemonError> {
let mut admissions = Vec::with_capacity(config.callers.len()); config
for caller in &config.callers { .callers
let raw = Zeroizing::new( .iter()
fs::read(&caller.connection_key_file).map_err(CodexDaemonError::ConnectionKey)?, .map(|caller| {
); let secret = Zeroizing::new(
let secret = std::str::from_utf8(raw.as_slice()) fs::read_to_string(&caller.connection_key_file)
.map_err(|_| CodexDaemonError::InvalidConfig("caller connection key encoding"))? .map_err(CodexDaemonError::ConnectionKey)?,
.trim_end_matches(['\r', '\n']); );
admissions.push( let secret = secret.trim_end_matches(['\r', '\n']);
CodexCallerAdmission::new( CodexCallerAdmission::new(
caller.caller_runtime_id.clone(), caller.caller_runtime_id.clone(),
secret.to_string(), secret.to_string(),
@ -583,19 +293,15 @@ fn load_admissions(
caller.max_payload_bytes, caller.max_payload_bytes,
caller.max_output_tokens, caller.max_output_tokens,
) )
.map_err(CodexDaemonError::Service)?, .map_err(CodexDaemonError::Service)
); })
} .collect()
Ok(admissions)
} }
fn serve_connection( fn serve_connection(
mut stream: TcpStream, mut stream: TcpStream,
service: &Mutex<CodexTransportService>, service: &Mutex<CodexTransportService>,
backend: &CodexProviderBackend, backend: &CodexProviderBackend,
#[cfg(target_os = "linux")] runtime_presence_publisher: Option<
&Mutex<crate::idunn_health::RuntimePresencePublisher>,
>,
max_frame_bytes: usize, max_frame_bytes: usize,
timeout: Duration, timeout: Duration,
) -> Result<(), CodexDaemonError> { ) -> Result<(), CodexDaemonError> {
@ -604,37 +310,8 @@ fn serve_connection(
.and_then(|()| stream.set_write_timeout(Some(timeout))) .and_then(|()| stream.set_write_timeout(Some(timeout)))
.map_err(CodexDaemonError::Connection)?; .map_err(CodexDaemonError::Connection)?;
let request = read_transport_frame(&mut stream, max_frame_bytes).map_err(frame_error)?; let request = read_transport_frame(&mut stream, max_frame_bytes).map_err(frame_error)?;
#[cfg(target_os = "linux")]
let response = dispatch_native_frame(
&request,
runtime_presence_publisher.is_some(),
|message_id| {
let publisher =
runtime_presence_publisher.ok_or(CodexDaemonError::RouteObservationRefused)?;
let mut publisher = publisher
.lock()
.map_err(|_| CodexDaemonError::RuntimePresencePoisoned)?;
let backend_ready = backend.readiness().is_ok();
let response = publisher
.route_observation(message_id, backend_ready)
.map_err(CodexDaemonError::Health)?;
encode_cultnet_message_to_vec(&response, CultNetWireContract::CultNetSchemaV0)
.map_err(|_| CodexDaemonError::FrameEncoding)
},
|request| execute_inference_frame(request, service, backend),
)?;
#[cfg(not(target_os = "linux"))]
let response = execute_inference_frame(&request, service, backend)?;
write_transport_frame(&mut stream, &response, max_frame_bytes).map_err(frame_error)
}
fn execute_inference_frame(
request: &[u8],
service: &Mutex<CodexTransportService>,
backend: &CodexProviderBackend,
) -> Result<Vec<u8>, CodexDaemonError> {
let envelope: CodexTransportEnvelope = let envelope: CodexTransportEnvelope =
rmp_serde::from_slice(request).map_err(|_| CodexDaemonError::FrameEncoding)?; rmp_serde::from_slice(&request).map_err(|_| CodexDaemonError::FrameEncoding)?;
let admission = service let admission = service
.lock() .lock()
.map_err(|_| CodexDaemonError::ServicePoisoned)? .map_err(|_| CodexDaemonError::ServicePoisoned)?
@ -649,45 +326,8 @@ fn execute_inference_frame(
.complete(*claim, result)? .complete(*claim, result)?
} }
}; };
rmp_serde::to_vec(&response).map_err(|_| CodexDaemonError::FrameEncoding) let response = rmp_serde::to_vec(&response).map_err(|_| CodexDaemonError::FrameEncoding)?;
} write_transport_frame(&mut stream, &response, max_frame_bytes).map_err(frame_error)
#[cfg(target_os = "linux")]
fn dispatch_native_frame<RouteObservation, Inference>(
request: &[u8],
managed: bool,
route_observation: RouteObservation,
inference: Inference,
) -> Result<Vec<u8>, CodexDaemonError>
where
RouteObservation: FnOnce(&str) -> Result<Vec<u8>, CodexDaemonError>,
Inference: FnOnce(&[u8]) -> Result<Vec<u8>, CodexDaemonError>,
{
let Ok(message) =
decode_cultnet_message_from_slice(request, CultNetWireContract::CultNetSchemaV0)
else {
return inference(request);
};
let CultNetMessage::SnapshotRequest {
message_id,
schema_ids,
record_keys,
} = message
else {
return inference(request);
};
let exact_schema = matches!(
schema_ids.as_deref(),
Some([schema_id]) if schema_id == GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA
);
let exact_record = matches!(
record_keys.as_deref(),
Some([record_key]) if record_key == crate::idunn_health::CONNECTOR_TARGET
);
if !managed || !exact_schema || !exact_record {
return Err(CodexDaemonError::RouteObservationRefused);
}
route_observation(&message_id)
} }
fn frame_error(error: TransportFrameError) -> CodexDaemonError { fn frame_error(error: TransportFrameError) -> CodexDaemonError {
@ -758,17 +398,11 @@ pub enum CodexDaemonError {
#[error("connector request thread failed")] #[error("connector request thread failed")]
Thread(#[source] std::io::Error), Thread(#[source] std::io::Error),
#[cfg(target_os = "linux")] #[cfg(target_os = "linux")]
#[error("connector runtime presence failed: {0}")] #[error("connector health publication could not start: {0}")]
Health(#[source] anyhow::Error), Health(#[source] anyhow::Error),
#[cfg(target_os = "linux")] #[cfg(target_os = "linux")]
#[error("connector health thread could not start")] #[error("connector health thread could not start")]
HealthThread(#[source] std::io::Error), HealthThread(#[source] std::io::Error),
#[cfg(target_os = "linux")]
#[error("connector runtime presence publisher lock was poisoned")]
RuntimePresencePoisoned,
#[cfg(target_os = "linux")]
#[error("connector route observation request was refused")]
RouteObservationRefused,
#[error("connector connection failed")] #[error("connector connection failed")]
Connection(#[source] std::io::Error), Connection(#[source] std::io::Error),
#[error("connector frame exceeded its bound")] #[error("connector frame exceeded its bound")]
@ -777,13 +411,13 @@ pub enum CodexDaemonError {
FrameEncoding, FrameEncoding,
#[error("system clock is before the Unix epoch")] #[error("system clock is before the Unix epoch")]
Clock, Clock,
#[error("connector listener closed")]
ListenerClosed,
} }
#[cfg(test)] #[cfg(test)]
mod tests { mod tests {
use super::*; use super::*;
#[cfg(target_os = "linux")]
use std::cell::Cell;
use std::io::Cursor; use std::io::Cursor;
fn config(root: &Path) -> CodexDaemonConfig { fn config(root: &Path) -> CodexDaemonConfig {
@ -810,118 +444,6 @@ mod tests {
} }
} }
#[test]
#[cfg(target_os = "linux")]
fn signed_health_never_reports_an_unready_provider_active() {
assert_eq!(
provider_health_state(true),
("active", "credential-isolated-transport-ready")
);
assert_eq!(
provider_health_state(false),
("degraded", "provider-backend-unready")
);
}
#[test]
#[cfg(target_os = "linux")]
fn exact_managed_route_probe_bypasses_inference_execution() {
let request = encode_cultnet_message_to_vec(
&CultNetMessage::SnapshotRequest {
message_id: "route-challenge-17".into(),
schema_ids: Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]),
record_keys: Some(vec![crate::idunn_health::CONNECTOR_TARGET.into()]),
},
CultNetWireContract::CultNetSchemaV0,
)
.unwrap();
let inference_calls = Cell::new(0);
let route_calls = Cell::new(0);
let response = dispatch_native_frame(
&request,
true,
|message_id| {
route_calls.set(route_calls.get() + 1);
assert_eq!(message_id, "route-challenge-17");
Ok(b"signed-route-observation".to_vec())
},
|_| {
inference_calls.set(inference_calls.get() + 1);
Ok(b"provider-response".to_vec())
},
)
.unwrap();
assert_eq!(response, b"signed-route-observation");
assert_eq!(route_calls.get(), 1);
assert_eq!(inference_calls.get(), 0);
}
#[test]
#[cfg(target_os = "linux")]
fn route_probe_refuses_broad_foreign_and_unmanaged_requests() {
let requests = [
(
CultNetMessage::SnapshotRequest {
message_id: "broad".into(),
schema_ids: None,
record_keys: None,
},
true,
),
(
CultNetMessage::SnapshotRequest {
message_id: "foreign-schema".into(),
schema_ids: Some(vec!["gamecult.other.v1".into()]),
record_keys: Some(vec![crate::idunn_health::CONNECTOR_TARGET.into()]),
},
true,
),
(
CultNetMessage::SnapshotRequest {
message_id: "foreign-record".into(),
schema_ids: Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]),
record_keys: Some(vec!["other-target".into()]),
},
true,
),
(
CultNetMessage::SnapshotRequest {
message_id: "unmanaged".into(),
schema_ids: Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]),
record_keys: Some(vec![crate::idunn_health::CONNECTOR_TARGET.into()]),
},
false,
),
];
for (request, managed) in requests {
let request =
encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0)
.unwrap();
let route_calls = Cell::new(0);
let inference_calls = Cell::new(0);
assert!(matches!(
dispatch_native_frame(
&request,
managed,
|_| {
route_calls.set(route_calls.get() + 1);
Ok(Vec::new())
},
|_| {
inference_calls.set(inference_calls.get() + 1);
Ok(Vec::new())
},
),
Err(CodexDaemonError::RouteObservationRefused)
));
assert_eq!(route_calls.get(), 0);
assert_eq!(inference_calls.get(), 0);
}
}
#[test] #[test]
fn typed_cultcache_config_round_trips_and_refuses_epoch_substitution() { fn typed_cultcache_config_round_trips_and_refuses_epoch_substitution() {
let root = std::env::temp_dir().join(format!( let root = std::env::temp_dir().join(format!(
@ -934,8 +456,6 @@ mod tests {
let expected = config(&root); let expected = config(&root);
write_daemon_config(&path, &expected).unwrap(); write_daemon_config(&path, &expected).unwrap();
assert_eq!(load_daemon_config(&path).unwrap(), expected); assert_eq!(load_daemon_config(&path).unwrap(), expected);
let loaded = load_daemon_config_read_only(&path).unwrap();
assert_eq!(loaded, expected);
let mut stale = expected; let mut stale = expected;
stale.epoch = 0; stale.epoch = 0;
@ -946,74 +466,6 @@ mod tests {
fs::remove_dir_all(root).unwrap(); fs::remove_dir_all(root).unwrap();
} }
#[test]
fn production_config_snapshot_never_creates_its_authority_lock() {
let root = std::env::temp_dir().join(format!(
"codex-connector-read-only-config-{}-{}",
std::process::id(),
unix_ms().unwrap()
));
fs::create_dir_all(&root).unwrap();
let path = root.join("connector.cc");
write_daemon_config(&path, &config(&root)).unwrap();
let before = fs::read(&path).unwrap();
let lock_path = path.with_file_name("connector.cc.lock");
fs::remove_file(&lock_path).unwrap();
assert!(load_daemon_config_read_only(&path).is_err());
assert!(!lock_path.exists());
assert_eq!(fs::read(&path).unwrap(), before);
fs::remove_dir_all(root).unwrap();
}
#[test]
#[cfg(target_os = "linux")]
fn managed_runtime_refuses_a_mutable_config_authority() {
let root = std::env::temp_dir().join(format!(
"codex-connector-managed-config-{}-{}",
std::process::id(),
unix_ms().unwrap()
));
fs::create_dir_all(&root).unwrap();
let path = root.join("connector.cc");
write_daemon_config(&path, &config(&root)).unwrap();
assert!(crate::idunn_health::require_managed_config_store(&path).is_err());
fs::remove_dir_all(root).unwrap();
}
#[test]
#[cfg(target_os = "linux")]
fn idunn_candidate_bind_replaces_config_only_for_a_complete_managed_launch() {
let configured = "127.0.0.1:4103".parse().unwrap();
let candidate = "127.0.0.1:18831".parse().unwrap();
assert_eq!(
select_effective_bind(configured, None, None).unwrap(),
configured
);
assert_eq!(
select_effective_bind(
configured,
Some("/run/idunn/runtime/instance".into()),
Some("127.0.0.1:18831".into()),
)
.unwrap(),
candidate
);
assert!(
select_effective_bind(configured, Some("/run/idunn/runtime/instance".into()), None,)
.is_err()
);
assert!(
select_effective_bind(
configured,
Some("/run/idunn/runtime/instance".into()),
Some("0.0.0.0:18831".into()),
)
.is_err()
);
}
#[test] #[test]
fn single_caller_initializer_derives_daemon_bounds_without_reowning_policy() { fn single_caller_initializer_derives_daemon_bounds_without_reowning_policy() {
let root = Path::new("/srv/codex-connector"); let root = Path::new("/srv/codex-connector");
@ -1042,125 +494,6 @@ mod tests {
config.validate().unwrap(); config.validate().unwrap();
} }
#[test]
fn typed_config_preserves_a_unique_multi_model_caller_allowlist() {
let root = Path::new("/srv/codex-connector");
let mut config = config(root);
config.callers[0].allowed_models =
vec!["gpt-5.6-luna".to_string(), "gpt-5.6-terra".to_string()];
config.validate().unwrap();
config.callers[0]
.allowed_models
.push("gpt-5.6-luna".to_string());
assert!(matches!(
config.validate(),
Err(CodexDaemonError::InvalidConfig("caller admission"))
));
}
#[test]
fn caller_admission_reserves_every_independently_admitted_request_slot() {
let root = Path::new("/srv/codex-connector");
let first = CodexCallerConfig {
caller_runtime_id: "ghostlight-dungeon-yggdrasil".to_string(),
connection_key_file: root.join("ghostlight.key"),
connection_key_epoch: 1,
allowed_models: vec!["gpt-5.6-luna".to_string()],
max_concurrent_requests: 8,
max_payload_bytes: 1_048_576,
max_output_tokens: 16_384,
};
let mut config = CodexDaemonConfig::single_caller(
"127.0.0.1:4103",
root.join("codex"),
[7; 32],
root.join("codex-home"),
root.join("replay.cc"),
first,
);
config
.admit_caller(CodexCallerConfig {
caller_runtime_id: "epiphany-model-runtime".to_string(),
connection_key_file: root.join("epiphany.key"),
connection_key_epoch: 1,
allowed_models: vec!["gpt-5.6-luna".to_string()],
max_concurrent_requests: 8,
max_payload_bytes: 1_048_576,
max_output_tokens: 16_384,
})
.unwrap();
assert_eq!(config.max_connections, 16);
config.validate().unwrap();
config.max_connections = 15;
assert!(matches!(
config.validate(),
Err(CodexDaemonError::InvalidConfig("connection capacity"))
));
}
#[test]
fn provider_capacity_is_caller_execution_capacity_not_listener_capacity() {
let root = Path::new("/srv/codex-connector");
let mut config = config(root);
config.max_connections = 64;
assert_eq!(caller_request_capacity(&config.callers).unwrap(), 4);
assert_eq!(required_connection_capacity(&config.callers).unwrap(), 8);
}
#[test]
#[cfg(target_os = "linux")]
fn managed_runtime_uses_the_sealed_adjacent_codex_and_declared_state_layout() {
let root = std::env::temp_dir().join(format!(
"codex-connector-layout-{}-{}",
std::process::id(),
unix_ms().unwrap()
));
let release = root.join("release");
let state = root.join("state");
fs::create_dir_all(release.clone()).unwrap();
fs::create_dir_all(&state).unwrap();
fs::write(release.join("codex-connector"), b"connector").unwrap();
fs::write(release.join("codex"), b"codex").unwrap();
let mut config = config(&root);
config.codex_executable = release.join("codex");
config.codex_home = state.join("codex-home");
config.replay_store = state.join("replay.cc");
let (_, codex_home, replay_store) =
managed_runtime_paths(&config, &release.join("codex-connector"), &state).unwrap();
assert_eq!(codex_home, state.join("codex-home"));
assert_eq!(
replay_store,
fs::canonicalize(&state).unwrap().join("replay.cc")
);
fs::write(&replay_store, b"replay").unwrap();
let replay_alias = state.join("replay-alias.cc");
fs::hard_link(&replay_store, &replay_alias).unwrap();
assert!(managed_runtime_paths(&config, &release.join("codex-connector"), &state).is_err());
fs::remove_file(&replay_alias).unwrap();
fs::remove_file(&replay_store).unwrap();
std::os::unix::fs::symlink(state.join("missing-replay.cc"), &replay_store).unwrap();
assert!(managed_runtime_paths(&config, &release.join("codex-connector"), &state).is_err());
fs::remove_file(&replay_store).unwrap();
config.codex_executable = root.join("other-codex");
fs::write(&config.codex_executable, b"other").unwrap();
assert!(managed_runtime_paths(&config, &release.join("codex-connector"), &state).is_err());
config.codex_executable = release.join("codex");
let other_state = root.join("other-state");
fs::create_dir_all(other_state.join("codex-home")).unwrap();
assert!(
managed_runtime_paths(&config, &release.join("codex-connector"), &other_state,)
.is_err()
);
fs::remove_dir_all(root).unwrap();
}
#[test] #[test]
fn direct_pipe_frame_is_big_endian_and_refuses_oversize_before_allocation() { fn direct_pipe_frame_is_big_endian_and_refuses_oversize_before_allocation() {
let mut encoded = Vec::new(); let mut encoded = Vec::new();

1688
src/idunn_health.rs
View file

File diff suppressed because it is too large Load diff

80
src/lib.rs
View file

@ -1,6 +1,9 @@
use std::collections::HashSet; use std::collections::HashSet;
#[cfg(feature = "client")]
use std::io::{Read, Write}; use std::io::{Read, Write};
#[cfg(feature = "client")]
use std::net::{SocketAddr, TcpStream}; use std::net::{SocketAddr, TcpStream};
#[cfg(feature = "client")]
use std::time::Duration; use std::time::Duration;
#[cfg(feature = "daemon")] #[cfg(feature = "daemon")]
@ -8,14 +11,18 @@ use std::collections::HashMap;
#[cfg(feature = "daemon")] #[cfg(feature = "daemon")]
use std::path::Path; use std::path::Path;
#[cfg(feature = "client")]
use aes_gcm::aead::{Aead, Payload}; use aes_gcm::aead::{Aead, Payload};
#[cfg(feature = "client")]
use aes_gcm::{Aes256Gcm, KeyInit, Nonce}; use aes_gcm::{Aes256Gcm, KeyInit, Nonce};
#[cfg(feature = "daemon")] #[cfg(feature = "daemon")]
use cultcache_rs::{CultCache, DatabaseEntry, OwnedRedbMessagePackBackingStore}; use cultcache_rs::{CultCache, DatabaseEntry, OwnedRedbMessagePackBackingStore};
#[cfg(feature = "client")]
use rand_core::{OsRng, RngCore}; use rand_core::{OsRng, RngCore};
use serde::{Deserialize, Serialize}; use serde::{Deserialize, Serialize};
use sha2::{Digest, Sha256}; use sha2::{Digest, Sha256};
use thiserror::Error; use thiserror::Error;
#[cfg(feature = "client")]
use zeroize::{Zeroize, Zeroizing}; use zeroize::{Zeroize, Zeroizing};
#[cfg(feature = "daemon")] #[cfg(feature = "daemon")]
@ -37,6 +44,11 @@ pub use daemon::load_daemon_config;
pub use daemon::serve; pub use daemon::serve;
#[cfg(feature = "daemon")] #[cfg(feature = "daemon")]
pub use daemon::write_daemon_config; pub use daemon::write_daemon_config;
#[cfg(all(feature = "daemon", target_os = "linux"))]
pub use idunn_health::{
CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT, ProviderHealthPublisher,
enroll_provider_health_identity, provider_health_public_key_hex,
};
#[cfg(feature = "daemon")] #[cfg(feature = "daemon")]
pub use provider_backend::CodexAppServerConfig; pub use provider_backend::CodexAppServerConfig;
#[cfg(feature = "daemon")] #[cfg(feature = "daemon")]
@ -54,9 +66,11 @@ pub const RESULT_SCHEMA_ID: &str = "gamecult.codex.transport_result.v2";
pub const RECEIPT_SCHEMA_ID: &str = "gamecult.codex.transport_receipt.v2"; pub const RECEIPT_SCHEMA_ID: &str = "gamecult.codex.transport_receipt.v2";
pub const ENVELOPE_SCHEMA_ID: &str = "gamecult.codex.transport_envelope.v2"; pub const ENVELOPE_SCHEMA_ID: &str = "gamecult.codex.transport_envelope.v2";
#[cfg(feature = "client")]
#[derive(Clone, PartialEq, Eq)] #[derive(Clone, PartialEq, Eq)]
pub struct CodexTransportKey([u8; 32]); pub struct CodexTransportKey([u8; 32]);
#[cfg(feature = "client")]
impl CodexTransportKey { impl CodexTransportKey {
pub fn from_connection_secret(secret: &str) -> Result<Self, ServiceError> { pub fn from_connection_secret(secret: &str) -> Result<Self, ServiceError> {
if secret.trim().is_empty() || secret.trim() != secret { if secret.trim().is_empty() || secret.trim() != secret {
@ -66,6 +80,7 @@ impl CodexTransportKey {
} }
} }
#[cfg(feature = "client")]
impl Drop for CodexTransportKey { impl Drop for CodexTransportKey {
fn drop(&mut self) { fn drop(&mut self) {
self.0.zeroize(); self.0.zeroize();
@ -629,12 +644,14 @@ pub fn canonical_invocation_bytes(
rmp_serde::to_vec(invocation).map_err(|_| ContractError::Encoding) rmp_serde::to_vec(invocation).map_err(|_| ContractError::Encoding)
} }
#[cfg(feature = "client")]
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
pub enum CodexEnvelopeKind { pub enum CodexEnvelopeKind {
Invocation, Invocation,
Result, Result,
} }
#[cfg(feature = "client")]
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] #[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
pub struct CodexTransportEnvelope { pub struct CodexTransportEnvelope {
pub schema_id: String, pub schema_id: String,
@ -645,6 +662,7 @@ pub struct CodexTransportEnvelope {
pub ciphertext: Vec<u8>, pub ciphertext: Vec<u8>,
} }
#[cfg(feature = "client")]
pub fn encrypt_invocation( pub fn encrypt_invocation(
invocation: &CodexTransportInvocation, invocation: &CodexTransportInvocation,
security: &CodexTransportKey, security: &CodexTransportKey,
@ -659,6 +677,7 @@ pub fn encrypt_invocation(
) )
} }
#[cfg(feature = "client")]
pub fn decrypt_invocation( pub fn decrypt_invocation(
envelope: &CodexTransportEnvelope, envelope: &CodexTransportEnvelope,
security: &CodexTransportKey, security: &CodexTransportKey,
@ -675,6 +694,7 @@ pub fn decrypt_invocation(
Ok(invocation) Ok(invocation)
} }
#[cfg(feature = "client")]
pub fn encrypt_result( pub fn encrypt_result(
result: &CodexTransportResult, result: &CodexTransportResult,
security: &CodexTransportKey, security: &CodexTransportKey,
@ -689,6 +709,7 @@ pub fn encrypt_result(
) )
} }
#[cfg(feature = "client")]
pub fn decrypt_result( pub fn decrypt_result(
envelope: &CodexTransportEnvelope, envelope: &CodexTransportEnvelope,
security: &CodexTransportKey, security: &CodexTransportKey,
@ -707,6 +728,7 @@ pub fn decrypt_result(
Ok(result) Ok(result)
} }
#[cfg(feature = "client")]
#[derive(Clone)] #[derive(Clone)]
pub struct CodexConnectorClient { pub struct CodexConnectorClient {
endpoint: SocketAddr, endpoint: SocketAddr,
@ -715,6 +737,7 @@ pub struct CodexConnectorClient {
response_timeout: Option<Duration>, response_timeout: Option<Duration>,
} }
#[cfg(feature = "client")]
impl CodexConnectorClient { impl CodexConnectorClient {
pub fn new( pub fn new(
endpoint: SocketAddr, endpoint: SocketAddr,
@ -722,10 +745,8 @@ impl CodexConnectorClient {
max_frame_bytes: usize, max_frame_bytes: usize,
response_timeout: Option<Duration>, response_timeout: Option<Duration>,
) -> Result<Self, CodexConnectorClientError> { ) -> Result<Self, CodexConnectorClientError> {
if endpoint.port() == 0 if !endpoint.ip().is_loopback()
|| endpoint.ip().is_unspecified() || endpoint.port() == 0
|| endpoint.ip().is_multicast()
|| endpoint.ip() == std::net::Ipv4Addr::BROADCAST
|| !(4096..=u32::MAX as usize).contains(&max_frame_bytes) || !(4096..=u32::MAX as usize).contains(&max_frame_bytes)
|| response_timeout.is_some_and(|timeout| timeout.is_zero()) || response_timeout.is_some_and(|timeout| timeout.is_zero())
{ {
@ -767,45 +788,7 @@ impl CodexConnectorClient {
} }
} }
#[cfg(test)] #[cfg(feature = "client")]
mod client_endpoint_tests {
use super::*;
#[test]
fn client_admits_explicit_unicast_and_refuses_non_destinations() {
for endpoint in ["127.0.0.1:4103", "172.19.0.1:4103", "[::1]:4103"] {
assert!(
CodexConnectorClient::new(
endpoint.parse().unwrap(),
"bounded-client-endpoint-key",
64 * 1024,
Some(Duration::from_secs(1)),
)
.is_ok(),
"{endpoint} should be an admissible explicit destination"
);
}
for endpoint in [
"0.0.0.0:4103",
"255.255.255.255:4103",
"224.0.0.1:4103",
"[::]:4103",
"[ff02::1]:4103",
"127.0.0.1:0",
] {
assert!(matches!(
CodexConnectorClient::new(
endpoint.parse().unwrap(),
"bounded-client-endpoint-key",
64 * 1024,
Some(Duration::from_secs(1)),
),
Err(CodexConnectorClientError::InvalidConfig)
));
}
}
}
#[derive(Debug, Error)] #[derive(Debug, Error)]
pub enum CodexConnectorClientError { pub enum CodexConnectorClientError {
#[error("invalid connector client configuration")] #[error("invalid connector client configuration")]
@ -820,12 +803,14 @@ pub enum CodexConnectorClientError {
Transport(#[from] ServiceError), Transport(#[from] ServiceError),
} }
#[cfg(feature = "client")]
#[derive(Debug)] #[derive(Debug)]
enum TransportFrameError { enum TransportFrameError {
Connection(std::io::Error), Connection(std::io::Error),
Size, Size,
} }
#[cfg(feature = "client")]
fn client_frame_error(error: TransportFrameError) -> CodexConnectorClientError { fn client_frame_error(error: TransportFrameError) -> CodexConnectorClientError {
match error { match error {
TransportFrameError::Connection(error) => CodexConnectorClientError::Connection(error), TransportFrameError::Connection(error) => CodexConnectorClientError::Connection(error),
@ -833,6 +818,7 @@ fn client_frame_error(error: TransportFrameError) -> CodexConnectorClientError {
} }
} }
#[cfg(feature = "client")]
fn read_transport_frame( fn read_transport_frame(
reader: &mut impl Read, reader: &mut impl Read,
max_frame_bytes: usize, max_frame_bytes: usize,
@ -852,6 +838,7 @@ fn read_transport_frame(
Ok(payload) Ok(payload)
} }
#[cfg(feature = "client")]
fn write_transport_frame( fn write_transport_frame(
writer: &mut impl Write, writer: &mut impl Write,
payload: &[u8], payload: &[u8],
@ -867,6 +854,7 @@ fn write_transport_frame(
.map_err(TransportFrameError::Connection) .map_err(TransportFrameError::Connection)
} }
#[cfg(feature = "client")]
fn encrypt_message( fn encrypt_message(
caller_runtime_id: &str, caller_runtime_id: &str,
request_id: &str, request_id: &str,
@ -904,6 +892,7 @@ fn encrypt_message(
}) })
} }
#[cfg(feature = "client")]
fn decrypt_message( fn decrypt_message(
envelope: &CodexTransportEnvelope, envelope: &CodexTransportEnvelope,
security: &CodexTransportKey, security: &CodexTransportKey,
@ -926,6 +915,7 @@ fn decrypt_message(
.map_err(|_| ServiceError::Encryption) .map_err(|_| ServiceError::Encryption)
} }
#[cfg(feature = "client")]
fn envelope_aad( fn envelope_aad(
schema_id: &str, schema_id: &str,
caller_runtime_id: &str, caller_runtime_id: &str,
@ -936,6 +926,7 @@ fn envelope_aad(
.map_err(|_| ServiceError::Encoding) .map_err(|_| ServiceError::Encoding)
} }
#[cfg(feature = "client")]
fn validate_envelope( fn validate_envelope(
envelope: &CodexTransportEnvelope, envelope: &CodexTransportEnvelope,
expected_kind: CodexEnvelopeKind, expected_kind: CodexEnvelopeKind,
@ -1679,6 +1670,7 @@ mod tests {
.unwrap() .unwrap()
} }
#[cfg(feature = "client")]
fn security(key: &str) -> CodexTransportKey { fn security(key: &str) -> CodexTransportKey {
CodexTransportKey::from_connection_secret(key).unwrap() CodexTransportKey::from_connection_secret(key).unwrap()
} }
@ -1690,6 +1682,7 @@ mod tests {
CodexTransportInvocation::new(caller_runtime_id, 2_000, [7; 32], request).unwrap() CodexTransportInvocation::new(caller_runtime_id, 2_000, [7; 32], request).unwrap()
} }
#[cfg(feature = "client")]
#[test] #[test]
fn encrypted_invocation_hides_content_and_binds_outer_identity() { fn encrypted_invocation_hides_content_and_binds_outer_identity() {
let invocation = invocation(); let invocation = invocation();
@ -1718,6 +1711,7 @@ mod tests {
); );
} }
#[cfg(feature = "client")]
#[test] #[test]
fn shared_client_transports_one_exact_encrypted_request_and_result() { fn shared_client_transports_one_exact_encrypted_request_and_result() {
let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap(); let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();

119
src/main.rs
View file

@ -1,13 +1,35 @@
use std::collections::HashSet;
use std::ffi::OsString; use std::ffi::OsString;
use std::path::PathBuf; use std::path::PathBuf;
use codex_connector::{CodexCallerConfig, CodexDaemonConfig}; use codex_connector::{CodexCallerConfig, CodexDaemonConfig};
const USAGE: &str = "usage:\n codex-connector --config PATH.cc\n codex-connector --config PATH.cc --state-root PATH\n codex-connector --initialize-single-caller-config PATH.cc BIND CODEX_EXECUTABLE CODEX_SHA256 CODEX_HOME REPLAY_STORE CALLER_RUNTIME_ID CONNECTION_KEY_FILE CONNECTION_KEY_EPOCH ALLOWED_MODELS_CSV MAX_CONCURRENT_REQUESTS MAX_PAYLOAD_BYTES MAX_OUTPUT_TOKENS\n codex-connector --admit-caller-config SOURCE.cc DESTINATION.cc CALLER_RUNTIME_ID CONNECTION_KEY_FILE CONNECTION_KEY_EPOCH ALLOWED_MODELS_CSV MAX_CONCURRENT_REQUESTS MAX_PAYLOAD_BYTES MAX_OUTPUT_TOKENS"; const USAGE: &str = "usage:\n codex-connector --config PATH.cc\n codex-connector --initialize-single-caller-config PATH.cc BIND CODEX_EXECUTABLE CODEX_SHA256 CODEX_HOME REPLAY_STORE CALLER_RUNTIME_ID CONNECTION_KEY_FILE CONNECTION_KEY_EPOCH ALLOWED_MODEL MAX_CONCURRENT_REQUESTS MAX_PAYLOAD_BYTES MAX_OUTPUT_TOKENS\n codex-connector --enroll-provider-health-identity PATH.cc\n codex-connector --provider-health-public-key PATH.cc";
fn main() { fn main() {
let args = std::env::args_os().skip(1).collect::<Vec<_>>(); let args = std::env::args_os().skip(1).collect::<Vec<_>>();
#[cfg(target_os = "linux")]
if let [flag, path] = args.as_slice() {
if flag == "--enroll-provider-health-identity" {
match codex_connector::enroll_provider_health_identity(&PathBuf::from(path)) {
Ok(public_key) => println!("{public_key}"),
Err(error) => {
eprintln!("codex-connector health identity enrollment failed: {error}");
std::process::exit(2);
}
}
return;
}
if flag == "--provider-health-public-key" {
match codex_connector::provider_health_public_key_hex(&PathBuf::from(path)) {
Ok(public_key) => println!("{public_key}"),
Err(error) => {
eprintln!("codex-connector health identity read failed: {error}");
std::process::exit(2);
}
}
return;
}
}
if args if args
.first() .first()
.is_some_and(|value| value == "--initialize-single-caller-config") .is_some_and(|value| value == "--initialize-single-caller-config")
@ -19,68 +41,19 @@ fn main() {
} }
return; return;
} }
if args let config = match args.as_slice() {
.first() [flag, path] if flag == "--config" => PathBuf::from(path),
.is_some_and(|value| value == "--admit-caller-config")
{
if let Err(error) = admit_caller_config(&args[1..]) {
eprintln!("codex-connector caller admission failed: {error}");
std::process::exit(2);
}
return;
}
let (config, state_root) = match args.as_slice() {
[flag, path] if flag == "--config" => (PathBuf::from(path), None),
[config_flag, config, state_flag, state_root]
if config_flag == "--config" && state_flag == "--state-root" =>
{
(PathBuf::from(config), Some(PathBuf::from(state_root)))
}
_ => { _ => {
eprintln!("{USAGE}"); eprintln!("{USAGE}");
std::process::exit(2); std::process::exit(2);
} }
}; };
if let Err(error) = codex_connector::serve(&config, state_root.as_deref()) { if let Err(error) = codex_connector::serve(&config) {
eprintln!("codex-connector stopped: {error}"); eprintln!("codex-connector stopped: {error}");
std::process::exit(1); std::process::exit(1);
} }
} }
fn admit_caller_config(args: &[OsString]) -> Result<(), String> {
let [
source,
destination,
caller_runtime_id,
connection_key_file,
connection_key_epoch,
allowed_models,
max_concurrent_requests,
max_payload_bytes,
max_output_tokens,
] = args
else {
return Err("expected exactly nine caller-admission values".to_string());
};
let mut config = codex_connector::load_daemon_config(&PathBuf::from(source))
.map_err(|error| error.to_string())?;
let caller_runtime_id = os_text(caller_runtime_id, "caller runtime")?;
let caller = CodexCallerConfig {
caller_runtime_id,
connection_key_file: PathBuf::from(connection_key_file),
connection_key_epoch: parse_number(connection_key_epoch, "connection key epoch")?,
allowed_models: parse_allowed_models(allowed_models)?,
max_concurrent_requests: parse_number(max_concurrent_requests, "max concurrent requests")?,
max_payload_bytes: parse_number(max_payload_bytes, "max payload bytes")?,
max_output_tokens: parse_number(max_output_tokens, "max output tokens")?,
};
config
.admit_caller(caller)
.map_err(|error| error.to_string())?;
codex_connector::write_daemon_config(&PathBuf::from(destination), &config)
.map_err(|error| error.to_string())
}
fn initialize_single_caller_config(args: &[OsString]) -> Result<(), String> { fn initialize_single_caller_config(args: &[OsString]) -> Result<(), String> {
let [ let [
path, path,
@ -92,7 +65,7 @@ fn initialize_single_caller_config(args: &[OsString]) -> Result<(), String> {
caller_runtime_id, caller_runtime_id,
connection_key_file, connection_key_file,
connection_key_epoch, connection_key_epoch,
allowed_models, allowed_model,
max_concurrent_requests, max_concurrent_requests,
max_payload_bytes, max_payload_bytes,
max_output_tokens, max_output_tokens,
@ -110,7 +83,7 @@ fn initialize_single_caller_config(args: &[OsString]) -> Result<(), String> {
caller_runtime_id: os_text(caller_runtime_id, "caller runtime")?, caller_runtime_id: os_text(caller_runtime_id, "caller runtime")?,
connection_key_file: PathBuf::from(connection_key_file), connection_key_file: PathBuf::from(connection_key_file),
connection_key_epoch: parse_number(connection_key_epoch, "connection key epoch")?, connection_key_epoch: parse_number(connection_key_epoch, "connection key epoch")?,
allowed_models: parse_allowed_models(allowed_models)?, allowed_models: vec![os_text(allowed_model, "allowed model")?],
max_concurrent_requests: parse_number( max_concurrent_requests: parse_number(
max_concurrent_requests, max_concurrent_requests,
"max concurrent requests", "max concurrent requests",
@ -131,19 +104,6 @@ fn os_text(value: &OsString, field: &str) -> Result<String, String> {
.ok_or_else(|| format!("{field} must be non-empty UTF-8")) .ok_or_else(|| format!("{field} must be non-empty UTF-8"))
} }
fn parse_allowed_models(value: &OsString) -> Result<Vec<String>, String> {
let raw = os_text(value, "allowed models")?;
let mut seen = HashSet::new();
let mut models = Vec::new();
for model in raw.split(',') {
if model.is_empty() || model.trim() != model || !seen.insert(model) {
return Err("allowed models must be unique non-empty comma-separated IDs".to_string());
}
models.push(model.to_string());
}
Ok(models)
}
fn parse_number<T>(value: &OsString, field: &str) -> Result<T, String> fn parse_number<T>(value: &OsString, field: &str) -> Result<T, String>
where where
T: std::str::FromStr, T: std::str::FromStr,
@ -153,27 +113,6 @@ where
.map_err(|_| format!("{field} is not a valid integer")) .map_err(|_| format!("{field} is not a valid integer"))
} }
#[cfg(test)]
mod tests {
use super::parse_allowed_models;
use std::ffi::OsString;
#[test]
fn model_allowlist_parser_preserves_one_or_many_exact_models() {
assert_eq!(
parse_allowed_models(&OsString::from("gpt-5.6-luna")).unwrap(),
vec!["gpt-5.6-luna"]
);
assert_eq!(
parse_allowed_models(&OsString::from("gpt-5.6-luna,gpt-5.6-terra")).unwrap(),
vec!["gpt-5.6-luna", "gpt-5.6-terra"]
);
assert!(parse_allowed_models(&OsString::from("gpt-5.6-luna,")).is_err());
assert!(parse_allowed_models(&OsString::from("gpt-5.6-luna, gpt-5.6-terra")).is_err());
assert!(parse_allowed_models(&OsString::from("gpt-5.6-luna,gpt-5.6-luna")).is_err());
}
}
fn parse_sha256(value: &OsString) -> Result<[u8; 32], String> { fn parse_sha256(value: &OsString) -> Result<[u8; 32], String> {
let value = os_text(value, "Codex SHA-256")?; let value = os_text(value, "Codex SHA-256")?;
if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) { if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {

14
src/provider_backend.rs
View file

@ -72,11 +72,7 @@ impl CodexProviderBackend {
let max_result_bytes = config.max_result_bytes; let max_result_bytes = config.max_result_bytes;
let http: ureq::Agent = ureq::Agent::config_builder() let http: ureq::Agent = ureq::Agent::config_builder()
.timeout_connect(Some(Duration::from_secs(30))) .timeout_connect(Some(Duration::from_secs(30)))
// ureq carries the receive-response deadline into streamed body .timeout_recv_response(Some(Duration::from_secs(30)))
// reads as a preceding phase. Keep both receive phases on the
// same provider window so a 30-second header bound cannot silently
// cap a valid long-running SSE response.
.timeout_recv_response(Some(Duration::from_secs(300)))
.timeout_recv_body(Some(Duration::from_secs(300))) .timeout_recv_body(Some(Duration::from_secs(300)))
.max_redirects(0) .max_redirects(0)
.http_status_as_error(false) .http_status_as_error(false)
@ -635,7 +631,6 @@ impl CodexAppServerAuthority {
.stdin(Stdio::piped()) .stdin(Stdio::piped())
.stdout(Stdio::piped()) .stdout(Stdio::piped())
.stderr(Stdio::inherit()); .stderr(Stdio::inherit());
strip_systemd_descriptor_environment(&mut command);
let mut child = command.spawn().map_err(CodexProviderBackendError::Spawn)?; let mut child = command.spawn().map_err(CodexProviderBackendError::Spawn)?;
let stdin = match child.stdin.take() { let stdin = match child.stdin.take() {
Some(stdin) => stdin, Some(stdin) => stdin,
@ -698,13 +693,6 @@ impl CodexAppServerAuthority {
} }
} }
pub(crate) fn strip_systemd_descriptor_environment(command: &mut Command) {
command
.env_remove("LISTEN_PID")
.env_remove("LISTEN_FDS")
.env_remove("LISTEN_FDNAMES");
}
fn codex_cli_version_from_user_agent( fn codex_cli_version_from_user_agent(
user_agent: &str, user_agent: &str,
) -> Result<String, CodexProviderBackendError> { ) -> Result<String, CodexProviderBackendError> {