Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
fd7cec4176 | ||
|
|
5b6ba919b7 | ||
|
|
3ddc12e9f3 | ||
|
|
65192892f0 | ||
|
|
b70ec4d4ba | ||
|
|
a7ee296602 | ||
|
|
78b4331ee5 | ||
|
|
ede3c303bf | ||
|
|
68fe94b947 | ||
|
|
b2e76841bf | ||
|
|
2f29b90eeb | ||
|
|
be1776ab45 | ||
|
|
80de3acc18 | ||
|
|
b6b5102cb9 | ||
|
|
e7ef6a565f |
11 changed files with 2414 additions and 528 deletions
3
CLAUDE.md
Normal file
3
CLAUDE.md
Normal file
|
|
@ -0,0 +1,3 @@
|
|||
Repository doctrine for this project lives in AGENTS.md, shared with Codex.
|
||||
|
||||
@AGENTS.md
|
||||
56
Cargo.lock
generated
56
Cargo.lock
generated
|
|
@ -145,18 +145,17 @@ dependencies = [
|
|||
"anyhow",
|
||||
"base64 0.23.1",
|
||||
"chrono",
|
||||
"cultcache-rs 0.1.0 (git+https://github.com/GameCult/cultcache-rs.git?rev=fdbf3bf0a4b3f6b617c224d7439feb8e69d4474c)",
|
||||
"cultcache-rs",
|
||||
"cultnet-rs",
|
||||
"ed25519-dalek",
|
||||
"fs2",
|
||||
"libc",
|
||||
"rand_core 0.6.4",
|
||||
"rmp-serde",
|
||||
"serde",
|
||||
"serde_bytes",
|
||||
"serde_json",
|
||||
"sha2",
|
||||
"thiserror",
|
||||
"ureq",
|
||||
"uuid",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
|
|
@ -204,28 +203,11 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "cultcache-rs"
|
||||
version = "0.1.0"
|
||||
source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f"
|
||||
source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029"
|
||||
dependencies = [
|
||||
"anyhow",
|
||||
"chrono",
|
||||
"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)",
|
||||
"cultcache-rs-derive",
|
||||
"fs2",
|
||||
"redb",
|
||||
"rmp-serde",
|
||||
|
|
@ -239,17 +221,7 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "cultcache-rs-derive"
|
||||
version = "0.1.0"
|
||||
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"
|
||||
source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029"
|
||||
dependencies = [
|
||||
"proc-macro2",
|
||||
"quote",
|
||||
|
|
@ -259,13 +231,14 @@ dependencies = [
|
|||
[[package]]
|
||||
name = "cultnet-rs"
|
||||
version = "0.1.0"
|
||||
source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f"
|
||||
source = "git+https://github.com/GameCult/CultLib.git?rev=e171eca32329baa928f4a1d810401a8b4c857029#e171eca32329baa928f4a1d810401a8b4c857029"
|
||||
dependencies = [
|
||||
"aes-gcm",
|
||||
"anyhow",
|
||||
"base64 0.22.1",
|
||||
"chrono",
|
||||
"cultcache-rs 0.1.0 (git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f)",
|
||||
"cultcache-rs",
|
||||
"ed25519-dalek",
|
||||
"hmac",
|
||||
"rand",
|
||||
"rmp-serde",
|
||||
|
|
@ -276,6 +249,8 @@ dependencies = [
|
|||
"sha2",
|
||||
"socket2",
|
||||
"uuid",
|
||||
"windows-sys 0.61.2",
|
||||
"zeroize",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
|
|
@ -1184,6 +1159,15 @@ dependencies = [
|
|||
"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]]
|
||||
name = "windows-targets"
|
||||
version = "0.52.6"
|
||||
|
|
|
|||
14
Cargo.toml
14
Cargo.toml
|
|
@ -23,10 +23,9 @@ daemon = [
|
|||
"dep:chrono",
|
||||
"dep:cultcache-rs",
|
||||
"dep:cultnet-rs",
|
||||
"dep:ed25519-dalek",
|
||||
"dep:serde_bytes",
|
||||
"dep:fs2",
|
||||
"dep:libc",
|
||||
"dep:ureq",
|
||||
"dep:uuid",
|
||||
]
|
||||
|
||||
[dependencies]
|
||||
|
|
@ -34,16 +33,15 @@ aes-gcm = "0.10"
|
|||
anyhow = { version = "1", optional = true }
|
||||
base64 = "0.23"
|
||||
chrono = { version = "0.4", features = ["serde"], 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 = "75c180782aeba7cfd22d6412877397708a4ed28f", optional = true }
|
||||
ed25519-dalek = { version = "2.2", optional = true }
|
||||
cultcache-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultcache-rs", optional = true }
|
||||
cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "e171eca32329baa928f4a1d810401a8b4c857029", package = "cultnet-rs", optional = true }
|
||||
fs2 = { version = "0.4", optional = true }
|
||||
libc = { version = "0.2", optional = true }
|
||||
rand_core = { version = "0.6", features = ["getrandom"] }
|
||||
rmp-serde = "1"
|
||||
serde = { version = "1", features = ["derive"] }
|
||||
serde_bytes = { version = "0.11", optional = true }
|
||||
serde_json = "1"
|
||||
sha2 = "0.10"
|
||||
thiserror = "2"
|
||||
ureq = { version = "3.4", default-features = false, features = ["rustls"], optional = true }
|
||||
uuid = { version = "1", features = ["v4"], optional = true }
|
||||
zeroize = "1"
|
||||
|
|
|
|||
60
README.md
60
README.md
|
|
@ -5,6 +5,14 @@ subscription inference. Epiphany and Ghostlight remain separate minds. They
|
|||
derive their own exact provider requests, execute their own tools, interpret
|
||||
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,
|
||||
exact provider-request receipts, and redacted CultMesh health. A private pinned
|
||||
official `codex app-server` child is the sole writer and refresh authority for
|
||||
|
|
@ -54,11 +62,53 @@ cargo check --bin codex-connector --features daemon
|
|||
```
|
||||
|
||||
The same production binary can initialize a typed single-caller `.cc`
|
||||
configuration for first deployment, then serve only from that admitted state.
|
||||
The daemon publishes private-free signed health to Idunn from its own runtime
|
||||
identity. Its source revision also owns the exact Codex Linux package and
|
||||
binary hashes in `deployment/codex-linux-x64.manifest`, so changing the provider
|
||||
toolchain necessarily changes the connector release selected by Idunn.
|
||||
configuration for first deployment, then serve from that state. Outside an
|
||||
Idunn-managed launch the configuration's loopback bind remains available for
|
||||
local operation. In a managed launch `GAMECULT_IDUNN_CANDIDATE_BIND` is the
|
||||
effective private bind. Idunn admits the candidate and controls route promotion;
|
||||
the Connector only binds and serves its private endpoint.
|
||||
|
||||
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
|
||||
|
||||
|
|
|
|||
|
|
@ -1,4 +0,0 @@
|
|||
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
Normal file
122
deployment/idunn/recipe.toml
Normal file
|
|
@ -0,0 +1,122 @@
|
|||
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
805
src/daemon.rs
|
|
@ -7,11 +7,20 @@ use std::sync::{Arc, Mutex};
|
|||
use std::thread;
|
||||
use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
use std::os::unix::fs::MetadataExt;
|
||||
|
||||
use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use thiserror::Error;
|
||||
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::{
|
||||
CodexAppServerConfig, CodexCallerAdmission, CodexProviderBackend, CodexProviderBackendError,
|
||||
CodexTransportAdmission, CodexTransportEnvelope, CodexTransportService, ServiceError,
|
||||
|
|
@ -110,18 +119,24 @@ impl CodexDaemonConfig {
|
|||
{
|
||||
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();
|
||||
for caller in &self.callers {
|
||||
let mut allowed_models = HashSet::new();
|
||||
if caller.caller_runtime_id.trim().is_empty()
|
||||
|| caller.caller_runtime_id.trim() != caller.caller_runtime_id
|
||||
|| !caller_ids.insert(caller.caller_runtime_id.as_str())
|
||||
|| caller.connection_key_file.as_os_str().is_empty()
|
||||
|| caller.connection_key_epoch == 0
|
||||
|| caller.allowed_models.is_empty()
|
||||
|| caller
|
||||
.allowed_models
|
||||
.iter()
|
||||
.any(|model| model.trim().is_empty() || model.trim() != model)
|
||||
|| caller.allowed_models.iter().any(|model| {
|
||||
model.trim().is_empty()
|
||||
|| model.trim() != model
|
||||
|| !allowed_models.insert(model.as_str())
|
||||
})
|
||||
|| caller.max_concurrent_requests == 0
|
||||
|| caller.max_payload_bytes < 4096
|
||||
|| caller.max_payload_bytes > self.max_frame_bytes - FRAME_OVERHEAD_BUDGET
|
||||
|
|
@ -132,6 +147,40 @@ impl CodexDaemonConfig {
|
|||
}
|
||||
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> {
|
||||
|
|
@ -150,6 +199,30 @@ pub fn load_daemon_config(path: &Path) -> Result<CodexDaemonConfig, CodexDaemonE
|
|||
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(
|
||||
path: &Path,
|
||||
config: &CodexDaemonConfig,
|
||||
|
|
@ -167,37 +240,109 @@ pub fn write_daemon_config(
|
|||
Ok(())
|
||||
}
|
||||
|
||||
pub fn serve(config_path: &Path) -> Result<(), CodexDaemonError> {
|
||||
let config = load_daemon_config(config_path)?;
|
||||
let bind = config.validate()?;
|
||||
let service = Arc::new(Mutex::new(CodexTransportService::open(
|
||||
&config.replay_store,
|
||||
load_admissions(&config)?,
|
||||
config.max_expiry_skew_ms,
|
||||
)?));
|
||||
pub fn serve(
|
||||
config_path: &Path,
|
||||
managed_state_root: Option<&Path>,
|
||||
) -> Result<(), CodexDaemonError> {
|
||||
#[cfg(target_os = "linux")]
|
||||
if std::env::var_os(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).is_some() {
|
||||
crate::idunn_health::require_managed_config_store(config_path)
|
||||
.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 {
|
||||
executable: config.codex_executable.clone(),
|
||||
executable: codex_executable,
|
||||
executable_sha256: config.codex_executable_sha256,
|
||||
codex_home: config.codex_home.clone(),
|
||||
codex_home,
|
||||
max_result_bytes: config.max_frame_bytes - FRAME_OVERHEAD_BUDGET,
|
||||
})?);
|
||||
let readiness = backend.readiness()?;
|
||||
let listener = TcpListener::bind(bind).map_err(CodexDaemonError::Listen)?;
|
||||
let service = Arc::new(Mutex::new(CodexTransportService::open(
|
||||
&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));
|
||||
eprintln!(
|
||||
"codex-connector ready at {} for {} callers using {:?}",
|
||||
listener.local_addr().map_err(CodexDaemonError::Listen)?,
|
||||
bound,
|
||||
config.callers.len(),
|
||||
readiness.auth_mode
|
||||
);
|
||||
#[cfg(target_os = "linux")]
|
||||
start_idunn_health_publisher()?;
|
||||
|
||||
for accepted in listener.incoming() {
|
||||
let stream = match accepted {
|
||||
Ok(stream) => stream,
|
||||
Err(error) => return Err(CodexDaemonError::Listen(error)),
|
||||
};
|
||||
loop {
|
||||
let (stream, _) = listener.accept().map_err(CodexDaemonError::Listen)?;
|
||||
if !stream
|
||||
.peer_addr()
|
||||
.map_err(CodexDaemonError::Connection)?
|
||||
|
|
@ -213,49 +358,66 @@ pub fn serve(config_path: &Path) -> Result<(), CodexDaemonError> {
|
|||
};
|
||||
let service = service.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 timeout = Duration::from_millis(config.socket_timeout_ms);
|
||||
thread::Builder::new()
|
||||
.name("codex-connector-request".to_string())
|
||||
.spawn(move || {
|
||||
let _permit = permit;
|
||||
let _ = serve_connection(stream, &service, &backend, max_frame_bytes, timeout);
|
||||
let _ = serve_connection(
|
||||
stream,
|
||||
&service,
|
||||
&backend,
|
||||
#[cfg(target_os = "linux")]
|
||||
runtime_presence_publisher.as_deref(),
|
||||
max_frame_bytes,
|
||||
timeout,
|
||||
);
|
||||
})
|
||||
.map_err(CodexDaemonError::Thread)?;
|
||||
}
|
||||
Err(CodexDaemonError::ListenerClosed)
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
fn start_idunn_health_publisher() -> Result<(), CodexDaemonError> {
|
||||
let Some(endpoint) = std::env::var_os("CODEX_CONNECTOR_IDUNN_RUDP") else {
|
||||
return Ok(());
|
||||
};
|
||||
let endpoint = endpoint
|
||||
.to_string_lossy()
|
||||
.parse::<SocketAddr>()
|
||||
.map_err(|_| CodexDaemonError::InvalidConfig("Idunn health endpoint"))?;
|
||||
let daemon_id = required_environment("CODEX_CONNECTOR_IDUNN_DAEMON")?;
|
||||
let runtime_id = required_environment("CODEX_CONNECTOR_RUNTIME_ID")?;
|
||||
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)
|
||||
fn prepare_runtime_presence_publisher(
|
||||
candidate: SocketAddr,
|
||||
config: &CodexDaemonConfig,
|
||||
) -> Result<Option<crate::idunn_health::RuntimePresencePublisher>, CodexDaemonError> {
|
||||
if std::env::var_os(crate::idunn_health::IDUNN_RUNTIME_BUNDLE_ENVIRONMENT).is_none() {
|
||||
return Ok(None);
|
||||
}
|
||||
let capacity = u32::try_from(caller_request_capacity(&config.callers)?)
|
||||
.map_err(|_| CodexDaemonError::InvalidConfig("runtime capability capacity"))?;
|
||||
let publisher =
|
||||
crate::idunn_health::RuntimePresencePublisher::open_from_environment(candidate, capacity)
|
||||
.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
|
||||
.publish("active", "credential-isolated-transport-ready")
|
||||
.lock()
|
||||
.map_err(|_| CodexDaemonError::RuntimePresencePoisoned)?
|
||||
.publish(state, detail)
|
||||
.map_err(CodexDaemonError::Health)?;
|
||||
thread::Builder::new()
|
||||
.name("codex-connector-health".to_string())
|
||||
.spawn(move || {
|
||||
loop {
|
||||
thread::sleep(Duration::from_secs(10));
|
||||
if let Err(error) =
|
||||
publisher.publish("active", "credential-isolated-transport-ready")
|
||||
{
|
||||
let (state, detail) = provider_health_observation(&backend);
|
||||
let publication = publisher
|
||||
.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}");
|
||||
}
|
||||
}
|
||||
|
|
@ -265,25 +427,153 @@ fn start_idunn_health_publisher() -> Result<(), CodexDaemonError> {
|
|||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
fn required_environment(name: &'static str) -> Result<String, CodexDaemonError> {
|
||||
std::env::var(name)
|
||||
.ok()
|
||||
.filter(|value| !value.trim().is_empty() && value == value.trim())
|
||||
.ok_or(CodexDaemonError::InvalidConfig(name))
|
||||
fn provider_health_observation(backend: &CodexProviderBackend) -> (&'static str, &'static str) {
|
||||
provider_health_state(backend.readiness().is_ok())
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
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(
|
||||
config: &CodexDaemonConfig,
|
||||
) -> Result<Vec<CodexCallerAdmission>, CodexDaemonError> {
|
||||
config
|
||||
.callers
|
||||
.iter()
|
||||
.map(|caller| {
|
||||
let secret = Zeroizing::new(
|
||||
fs::read_to_string(&caller.connection_key_file)
|
||||
.map_err(CodexDaemonError::ConnectionKey)?,
|
||||
);
|
||||
let secret = secret.trim_end_matches(['\r', '\n']);
|
||||
let mut admissions = Vec::with_capacity(config.callers.len());
|
||||
for caller in &config.callers {
|
||||
let raw = Zeroizing::new(
|
||||
fs::read(&caller.connection_key_file).map_err(CodexDaemonError::ConnectionKey)?,
|
||||
);
|
||||
let secret = std::str::from_utf8(raw.as_slice())
|
||||
.map_err(|_| CodexDaemonError::InvalidConfig("caller connection key encoding"))?
|
||||
.trim_end_matches(['\r', '\n']);
|
||||
admissions.push(
|
||||
CodexCallerAdmission::new(
|
||||
caller.caller_runtime_id.clone(),
|
||||
secret.to_string(),
|
||||
|
|
@ -293,15 +583,19 @@ fn load_admissions(
|
|||
caller.max_payload_bytes,
|
||||
caller.max_output_tokens,
|
||||
)
|
||||
.map_err(CodexDaemonError::Service)
|
||||
})
|
||||
.collect()
|
||||
.map_err(CodexDaemonError::Service)?,
|
||||
);
|
||||
}
|
||||
Ok(admissions)
|
||||
}
|
||||
|
||||
fn serve_connection(
|
||||
mut stream: TcpStream,
|
||||
service: &Mutex<CodexTransportService>,
|
||||
backend: &CodexProviderBackend,
|
||||
#[cfg(target_os = "linux")] runtime_presence_publisher: Option<
|
||||
&Mutex<crate::idunn_health::RuntimePresencePublisher>,
|
||||
>,
|
||||
max_frame_bytes: usize,
|
||||
timeout: Duration,
|
||||
) -> Result<(), CodexDaemonError> {
|
||||
|
|
@ -310,8 +604,37 @@ fn serve_connection(
|
|||
.and_then(|()| stream.set_write_timeout(Some(timeout)))
|
||||
.map_err(CodexDaemonError::Connection)?;
|
||||
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 =
|
||||
rmp_serde::from_slice(&request).map_err(|_| CodexDaemonError::FrameEncoding)?;
|
||||
rmp_serde::from_slice(request).map_err(|_| CodexDaemonError::FrameEncoding)?;
|
||||
let admission = service
|
||||
.lock()
|
||||
.map_err(|_| CodexDaemonError::ServicePoisoned)?
|
||||
|
|
@ -326,8 +649,45 @@ fn serve_connection(
|
|||
.complete(*claim, result)?
|
||||
}
|
||||
};
|
||||
let response = rmp_serde::to_vec(&response).map_err(|_| CodexDaemonError::FrameEncoding)?;
|
||||
write_transport_frame(&mut stream, &response, max_frame_bytes).map_err(frame_error)
|
||||
rmp_serde::to_vec(&response).map_err(|_| CodexDaemonError::FrameEncoding)
|
||||
}
|
||||
|
||||
#[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 {
|
||||
|
|
@ -398,11 +758,17 @@ pub enum CodexDaemonError {
|
|||
#[error("connector request thread failed")]
|
||||
Thread(#[source] std::io::Error),
|
||||
#[cfg(target_os = "linux")]
|
||||
#[error("connector health publication could not start: {0}")]
|
||||
#[error("connector runtime presence failed: {0}")]
|
||||
Health(#[source] anyhow::Error),
|
||||
#[cfg(target_os = "linux")]
|
||||
#[error("connector health thread could not start")]
|
||||
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")]
|
||||
Connection(#[source] std::io::Error),
|
||||
#[error("connector frame exceeded its bound")]
|
||||
|
|
@ -411,13 +777,13 @@ pub enum CodexDaemonError {
|
|||
FrameEncoding,
|
||||
#[error("system clock is before the Unix epoch")]
|
||||
Clock,
|
||||
#[error("connector listener closed")]
|
||||
ListenerClosed,
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
#[cfg(target_os = "linux")]
|
||||
use std::cell::Cell;
|
||||
use std::io::Cursor;
|
||||
|
||||
fn config(root: &Path) -> CodexDaemonConfig {
|
||||
|
|
@ -444,6 +810,118 @@ 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]
|
||||
fn typed_cultcache_config_round_trips_and_refuses_epoch_substitution() {
|
||||
let root = std::env::temp_dir().join(format!(
|
||||
|
|
@ -456,6 +934,8 @@ mod tests {
|
|||
let expected = config(&root);
|
||||
write_daemon_config(&path, &expected).unwrap();
|
||||
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;
|
||||
stale.epoch = 0;
|
||||
|
|
@ -466,6 +946,74 @@ mod tests {
|
|||
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]
|
||||
fn single_caller_initializer_derives_daemon_bounds_without_reowning_policy() {
|
||||
let root = Path::new("/srv/codex-connector");
|
||||
|
|
@ -494,6 +1042,125 @@ mod tests {
|
|||
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]
|
||||
fn direct_pipe_frame_is_big_endian_and_refuses_oversize_before_allocation() {
|
||||
let mut encoded = Vec::new();
|
||||
|
|
|
|||
1678
src/idunn_health.rs
1678
src/idunn_health.rs
File diff suppressed because it is too large
Load diff
50
src/lib.rs
50
src/lib.rs
|
|
@ -37,11 +37,6 @@ pub use daemon::load_daemon_config;
|
|||
pub use daemon::serve;
|
||||
#[cfg(feature = "daemon")]
|
||||
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")]
|
||||
pub use provider_backend::CodexAppServerConfig;
|
||||
#[cfg(feature = "daemon")]
|
||||
|
|
@ -727,8 +722,10 @@ impl CodexConnectorClient {
|
|||
max_frame_bytes: usize,
|
||||
response_timeout: Option<Duration>,
|
||||
) -> Result<Self, CodexConnectorClientError> {
|
||||
if !endpoint.ip().is_loopback()
|
||||
|| endpoint.port() == 0
|
||||
if endpoint.port() == 0
|
||||
|| endpoint.ip().is_unspecified()
|
||||
|| endpoint.ip().is_multicast()
|
||||
|| endpoint.ip() == std::net::Ipv4Addr::BROADCAST
|
||||
|| !(4096..=u32::MAX as usize).contains(&max_frame_bytes)
|
||||
|| response_timeout.is_some_and(|timeout| timeout.is_zero())
|
||||
{
|
||||
|
|
@ -770,6 +767,45 @@ impl CodexConnectorClient {
|
|||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
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)]
|
||||
pub enum CodexConnectorClientError {
|
||||
#[error("invalid connector client configuration")]
|
||||
|
|
|
|||
119
src/main.rs
119
src/main.rs
|
|
@ -1,35 +1,13 @@
|
|||
use std::collections::HashSet;
|
||||
use std::ffi::OsString;
|
||||
use std::path::PathBuf;
|
||||
|
||||
use codex_connector::{CodexCallerConfig, CodexDaemonConfig};
|
||||
|
||||
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";
|
||||
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";
|
||||
|
||||
fn main() {
|
||||
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
|
||||
.first()
|
||||
.is_some_and(|value| value == "--initialize-single-caller-config")
|
||||
|
|
@ -41,19 +19,68 @@ fn main() {
|
|||
}
|
||||
return;
|
||||
}
|
||||
let config = match args.as_slice() {
|
||||
[flag, path] if flag == "--config" => PathBuf::from(path),
|
||||
if args
|
||||
.first()
|
||||
.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}");
|
||||
std::process::exit(2);
|
||||
}
|
||||
};
|
||||
if let Err(error) = codex_connector::serve(&config) {
|
||||
if let Err(error) = codex_connector::serve(&config, state_root.as_deref()) {
|
||||
eprintln!("codex-connector stopped: {error}");
|
||||
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> {
|
||||
let [
|
||||
path,
|
||||
|
|
@ -65,7 +92,7 @@ fn initialize_single_caller_config(args: &[OsString]) -> Result<(), String> {
|
|||
caller_runtime_id,
|
||||
connection_key_file,
|
||||
connection_key_epoch,
|
||||
allowed_model,
|
||||
allowed_models,
|
||||
max_concurrent_requests,
|
||||
max_payload_bytes,
|
||||
max_output_tokens,
|
||||
|
|
@ -83,7 +110,7 @@ fn initialize_single_caller_config(args: &[OsString]) -> Result<(), String> {
|
|||
caller_runtime_id: os_text(caller_runtime_id, "caller runtime")?,
|
||||
connection_key_file: PathBuf::from(connection_key_file),
|
||||
connection_key_epoch: parse_number(connection_key_epoch, "connection key epoch")?,
|
||||
allowed_models: vec![os_text(allowed_model, "allowed model")?],
|
||||
allowed_models: parse_allowed_models(allowed_models)?,
|
||||
max_concurrent_requests: parse_number(
|
||||
max_concurrent_requests,
|
||||
"max concurrent requests",
|
||||
|
|
@ -104,6 +131,19 @@ fn os_text(value: &OsString, field: &str) -> Result<String, String> {
|
|||
.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>
|
||||
where
|
||||
T: std::str::FromStr,
|
||||
|
|
@ -113,6 +153,27 @@ where
|
|||
.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> {
|
||||
let value = os_text(value, "Codex SHA-256")?;
|
||||
if value.len() != 64 || !value.bytes().all(|byte| byte.is_ascii_hexdigit()) {
|
||||
|
|
|
|||
|
|
@ -72,7 +72,11 @@ impl CodexProviderBackend {
|
|||
let max_result_bytes = config.max_result_bytes;
|
||||
let http: ureq::Agent = ureq::Agent::config_builder()
|
||||
.timeout_connect(Some(Duration::from_secs(30)))
|
||||
.timeout_recv_response(Some(Duration::from_secs(30)))
|
||||
// ureq carries the receive-response deadline into streamed body
|
||||
// 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)))
|
||||
.max_redirects(0)
|
||||
.http_status_as_error(false)
|
||||
|
|
@ -500,7 +504,7 @@ fn parse_responses_sse(reader: impl Read) -> Result<ParsedProviderStream, &'stat
|
|||
.by_ref()
|
||||
.take((MAX_SSE_FRAME_BYTES + 1) as u64)
|
||||
.read_until(b'\n', &mut line)
|
||||
.map_err(|error| provider_sse_read_error(error.kind()))?;
|
||||
.map_err(|error| provider_sse_read_error(&error))?;
|
||||
if read == 0 {
|
||||
if !data.is_empty() {
|
||||
state.accept_frame(&data)?;
|
||||
|
|
@ -544,7 +548,24 @@ fn parse_responses_sse(reader: impl Read) -> Result<ParsedProviderStream, &'stat
|
|||
})
|
||||
}
|
||||
|
||||
fn provider_sse_read_error(kind: io::ErrorKind) -> &'static str {
|
||||
fn provider_sse_read_error(error: &io::Error) -> &'static str {
|
||||
if let Some(ureq_error) = error
|
||||
.get_ref()
|
||||
.and_then(|source| source.downcast_ref::<ureq::Error>())
|
||||
{
|
||||
return match ureq_error {
|
||||
ureq::Error::Timeout(_) => "provider SSE read timed out",
|
||||
ureq::Error::Protocol(_) => "provider SSE protocol failed",
|
||||
ureq::Error::Tls(_) | ureq::Error::Rustls(_) => "provider SSE secure transport failed",
|
||||
ureq::Error::BodyStalled => "provider SSE body stalled",
|
||||
ureq::Error::Io(inner) => provider_sse_io_error(inner.kind()),
|
||||
_ => "provider SSE read failed",
|
||||
};
|
||||
}
|
||||
provider_sse_io_error(error.kind())
|
||||
}
|
||||
|
||||
fn provider_sse_io_error(kind: io::ErrorKind) -> &'static str {
|
||||
match kind {
|
||||
io::ErrorKind::TimedOut | io::ErrorKind::WouldBlock => "provider SSE read timed out",
|
||||
io::ErrorKind::ConnectionReset
|
||||
|
|
@ -614,6 +635,7 @@ impl CodexAppServerAuthority {
|
|||
.stdin(Stdio::piped())
|
||||
.stdout(Stdio::piped())
|
||||
.stderr(Stdio::inherit());
|
||||
strip_systemd_descriptor_environment(&mut command);
|
||||
let mut child = command.spawn().map_err(CodexProviderBackendError::Spawn)?;
|
||||
let stdin = match child.stdin.take() {
|
||||
Some(stdin) => stdin,
|
||||
|
|
@ -676,6 +698,13 @@ 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(
|
||||
user_agent: &str,
|
||||
) -> Result<String, CodexProviderBackendError> {
|
||||
|
|
@ -1057,6 +1086,14 @@ mod tests {
|
|||
}
|
||||
}
|
||||
|
||||
struct WrappedFailingReader(Option<ureq::Error>);
|
||||
|
||||
impl Read for WrappedFailingReader {
|
||||
fn read(&mut self, _buf: &mut [u8]) -> io::Result<usize> {
|
||||
Err(io::Error::other(self.0.take().unwrap()))
|
||||
}
|
||||
}
|
||||
|
||||
fn jwt(payload: Value) -> String {
|
||||
let payload = URL_SAFE_NO_PAD.encode(serde_json::to_vec(&payload).unwrap());
|
||||
format!("e30.{payload}.signature")
|
||||
|
|
@ -1257,6 +1294,10 @@ mod tests {
|
|||
parse_responses_sse(FailingReader(io::ErrorKind::Other)).err(),
|
||||
Some("provider SSE read failed")
|
||||
);
|
||||
assert_eq!(
|
||||
parse_responses_sse(WrappedFailingReader(Some(ureq::Error::BodyStalled))).err(),
|
||||
Some("provider SSE body stalled")
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
|
|
|||
Loading…
Reference in a new issue