Adopt Idunn runtime presence authority

This commit is contained in:
Meta Vens 2026-09-03 05:53:42 +02:00
commit ede3c303bf
12 changed files with 1628 additions and 2011 deletions

6
.gitmodules vendored Normal file
View file

@ -0,0 +1,6 @@
[submodule "vendor/cultnet-rs"]
path = vendor/cultnet-rs
url = https://github.com/GameCult/cultnet-rs.git
[submodule "vendor/cultcache-rs"]
path = vendor/cultcache-rs
url = https://github.com/GameCult/cultcache-rs.git

62
Cargo.lock generated
View file

@ -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=4b61f4df32d7ee1ba718596b03504d444489e7b4)",
"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,12 +203,12 @@ dependencies = [
[[package]]
name = "cultcache-rs"
version = "0.1.0"
source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f"
dependencies = [
"anyhow",
"chrono",
"cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f)",
"cultcache-rs-derive",
"fs2",
"redb",
"rmp-serde",
"serde",
"serde_bytes",
@ -218,36 +217,9 @@ dependencies = [
"windows-sys 0.59.0",
]
[[package]]
name = "cultcache-rs"
version = "0.1.0"
source = "git+https://github.com/GameCult/cultcache-rs.git?rev=4b61f4df32d7ee1ba718596b03504d444489e7b4#4b61f4df32d7ee1ba718596b03504d444489e7b4"
dependencies = [
"anyhow",
"chrono",
"cultcache-rs-derive 0.1.0 (git+https://github.com/GameCult/cultcache-rs.git?rev=4b61f4df32d7ee1ba718596b03504d444489e7b4)",
"fs2",
"rmp-serde",
"serde",
"serde_bytes",
"serde_json",
"uuid",
]
[[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=4b61f4df32d7ee1ba718596b03504d444489e7b4#4b61f4df32d7ee1ba718596b03504d444489e7b4"
dependencies = [
"proc-macro2",
"quote",
@ -257,13 +229,13 @@ dependencies = [
[[package]]
name = "cultnet-rs"
version = "0.1.0"
source = "git+https://github.com/GameCult/CultLib.git?rev=75c180782aeba7cfd22d6412877397708a4ed28f#75c180782aeba7cfd22d6412877397708a4ed28f"
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",
@ -274,6 +246,8 @@ dependencies = [
"sha2",
"socket2",
"uuid",
"windows-sys 0.61.2",
"zeroize",
]
[[package]]
@ -673,6 +647,15 @@ dependencies = [
"getrandom 0.3.4",
]
[[package]]
name = "redb"
version = "4.2.0"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "de6c3b63e007e90ce536ec2ae4690826136a20ec8dbbbb400daef1bb999d2e36"
dependencies = [
"libc",
]
[[package]]
name = "ring"
version = "0.17.14"
@ -1173,6 +1156,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"

View file

@ -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 = "4b61f4df32d7ee1ba718596b03504d444489e7b4", optional = true }
cultnet-rs = { git = "https://github.com/GameCult/CultLib.git", rev = "75c180782aeba7cfd22d6412877397708a4ed28f", optional = true }
ed25519-dalek = { version = "2.2", optional = true }
cultcache-rs = { path = "vendor/cultcache-rs", optional = true }
cultnet-rs = { path = "vendor/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"

View file

@ -54,33 +54,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.
On Linux, serving requires Idunn. The daemon publishes one private-free signed
`warming / traffic-admission-pending` statement, then waits for Idunn's exact
root-owned typed traffic grant before it launches the provider child, opens
replay state, or binds a socket.
The grant also binds the root-observed systemd process ID and Linux process
start time, so another process with the same service identity cannot consume
it. The connector revalidates the grant before binding, after a complete
transport frame has decoded and immediately before service admission, and while
idle. That successful request check transfers one command to Connector-owned
execution; later revocation stops later commands but does not rewrite the
admitted result. The connector has read-only access to the grant and its
shared-snapshot lock and cannot create, repair, or replace either file.
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 health signing identity is created by Idunn's generic provisioning
authority. CodexConnector can use that identity to sign health but cannot enroll
it or export its public key. Signed health is derived from the active release:
the root-owned runtime activation request, the immutable adjacent `DEPLOYMENT`
witness, the compiled source commit, and the observed hashes of both the running
connector and configured Codex binary must agree exactly. Initial health setup
and the probation publication are startup requirements. Later periodic health
publication is observational and does not acquire traffic authority. The source
revision owns the exact Codex Linux package and binary hashes in
`deployment/codex-linux-x64.manifest`; that manifest is compiled into the daemon
and must agree with the immutable release witness. Activation timestamps live
in Idunn's mutable deployment receipt. Changing the provider 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

View file

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

View file

@ -0,0 +1,124 @@
schema = "gamecult.idunn.target_declaration.v1"
target = "codex-connector"
source_stamp_environment = "CODEX_CONNECTOR_BUILD_COMMIT"
required_gitlinks = ["vendor/cultcache-rs", "vendor/cultnet-rs"]
[[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",
]
optional_environment = ["RUST_LOG"]
[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"

View file

@ -8,11 +8,10 @@ use std::thread;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
#[cfg(target_os = "linux")]
use std::time::Instant;
use std::os::unix::fs::MetadataExt;
use cultcache_rs::{CultCache, DatabaseEntry, SingleFileMessagePackBackingStore};
use serde::{Deserialize, Serialize};
use sha2::Digest;
use thiserror::Error;
use zeroize::Zeroizing;
@ -166,12 +165,15 @@ impl CodexDaemonConfig {
}
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)
})
.map(|total| total.max(8))
.ok_or(CodexDaemonError::InvalidConfig("connection capacity"))
}
@ -191,11 +193,9 @@ pub fn load_daemon_config(path: &Path) -> Result<CodexDaemonConfig, CodexDaemonE
Ok(config)
}
fn load_daemon_config_snapshot(
path: &Path,
) -> Result<(CodexDaemonConfig, Vec<u8>), CodexDaemonError> {
fn load_daemon_config_read_only(path: &Path) -> Result<CodexDaemonConfig, CodexDaemonError> {
let store = SingleFileMessagePackBackingStore::new(path);
let (config, snapshot) = store
let config = store
.with_read_only_shared_snapshot(|entries| {
let [envelope] = entries.as_slice() else {
anyhow::bail!("connector configuration must contain exactly one record");
@ -210,12 +210,11 @@ fn load_daemon_config_snapshot(
if rmp_serde::to_vec(&config)? != envelope.payload {
anyhow::bail!("connector configuration is not canonical positional MessagePack");
}
let snapshot = fs::read(path)?;
Ok((config, snapshot))
Ok(config)
})
.map_err(config_store_error)?;
config.validate()?;
Ok((config, snapshot))
Ok(config)
}
pub fn write_daemon_config(
@ -235,73 +234,106 @@ pub fn write_daemon_config(
Ok(())
}
pub fn serve(config_path: &Path) -> Result<(), CodexDaemonError> {
let (config, config_bytes) = load_daemon_config_snapshot(config_path)?;
let bind = config.validate()?;
let (admissions, caller_key_bindings) = load_admissions(&config)?;
pub fn serve(
config_path: &Path,
managed_state_root: Option<&Path>,
) -> Result<(), CodexDaemonError> {
#[cfg(target_os = "linux")]
let (traffic_admission, prepared_health_publisher) =
start_idunn_health_publisher(config_path, &config_bytes, &config, &caller_key_bindings)?;
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)?;
// Probation performs only immutable activation verification and its one
// signed warming publication. Provider process, replay, and socket
// actuation begin only after Idunn grants that exact startup statement.
#[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 service = Arc::new(Mutex::new(CodexTransportService::open(
&config.replay_store,
&replay_store,
admissions,
config.max_expiry_skew_ms,
)?));
#[cfg(target_os = "linux")]
traffic_admission
.require_current()
.map_err(CodexDaemonError::Health)?;
let listener = TcpListener::bind(bind).map_err(CodexDaemonError::Listen)?;
#[cfg(target_os = "linux")]
listener
.set_nonblocking(true)
.map_err(CodexDaemonError::Listen)?;
#[cfg(target_os = "linux")]
start_periodic_idunn_health_publisher(prepared_health_publisher, backend.clone())?;
if let Some(publisher) = prepared_presence_publisher {
start_periodic_runtime_presence_publisher(publisher, backend.clone())?;
}
let active_connections = Arc::new(AtomicUsize::new(0));
#[cfg(target_os = "linux")]
let mut next_traffic_admission_check = Instant::now();
eprintln!(
"codex-connector ready at {} for {} callers using {:?}",
listener.local_addr().map_err(CodexDaemonError::Listen)?,
bound,
config.callers.len(),
readiness.auth_mode
);
loop {
#[cfg(target_os = "linux")]
if Instant::now() >= next_traffic_admission_check {
traffic_admission
.require_current()
.map_err(CodexDaemonError::Health)?;
next_traffic_admission_check = Instant::now() + Duration::from_millis(100);
}
let stream = match listener.accept() {
Ok((stream, _)) => stream,
Err(error) if error.kind() == std::io::ErrorKind::WouldBlock => {
thread::sleep(Duration::from_millis(100));
continue;
}
Err(error) => return Err(CodexDaemonError::Listen(error)),
};
#[cfg(target_os = "linux")]
// The polling check terminates an idle process after revocation. This
// second check is the traffic commit point: an accepted socket receives
// no request worker under a stale grant.
traffic_admission
.require_current()
.map_err(CodexDaemonError::Health)?;
let (stream, _) = listener.accept().map_err(CodexDaemonError::Listen)?;
if !stream
.peer_addr()
.map_err(CodexDaemonError::Connection)?
@ -317,93 +349,43 @@ pub fn serve(config_path: &Path) -> Result<(), CodexDaemonError> {
};
let service = service.clone();
let backend = backend.clone();
#[cfg(target_os = "linux")]
let request_traffic_admission = traffic_admission.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,
#[cfg(target_os = "linux")]
&request_traffic_admission,
max_frame_bytes,
timeout,
);
let _ = serve_connection(stream, &service, &backend, max_frame_bytes, timeout);
})
.map_err(CodexDaemonError::Thread)?;
}
}
#[cfg(target_os = "linux")]
fn start_idunn_health_publisher(
config_path: &Path,
config_bytes: &[u8],
fn prepare_runtime_presence_publisher(
candidate: SocketAddr,
config: &CodexDaemonConfig,
caller_key_bindings: &[LoadedCallerKeyBinding],
) -> Result<
(
crate::idunn_health::ConnectorTrafficAdmissionGate,
crate::idunn_health::ProviderHealthPublisher,
),
CodexDaemonError,
> {
let endpoint = required_environment("CODEX_CONNECTOR_IDUNN_RUDP")?
.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::idunn_health::CODEX_CONNECTOR_IDUNN_HEALTH_CONTRACT.to_string());
let identity = PathBuf::from(required_environment(
"CODEX_CONNECTOR_IDUNN_HEALTH_IDENTITY",
)?);
let release = crate::idunn_health::active_release_binding(
&config.codex_executable,
&config.codex_executable_sha256,
)
.map_err(CodexDaemonError::Health)?;
let activation = crate::idunn_health::active_activation_binding(
config_path,
config_bytes,
config,
caller_key_bindings,
)
.map_err(CodexDaemonError::Health)?;
let mut publisher = crate::idunn_health::ProviderHealthPublisher::open(
endpoint,
daemon_id,
runtime_id,
contract,
&identity,
release.clone(),
activation.clone(),
)
.map_err(CodexDaemonError::Health)?;
let published = publisher
.publish("warming", "traffic-admission-pending")
.map_err(CodexDaemonError::Health)?;
let traffic_admission = crate::idunn_health::ConnectorTrafficAdmissionGate::from_environment(
&release,
&activation,
&published,
)
.map_err(CodexDaemonError::Health)?;
traffic_admission
.wait_until_granted(Duration::from_secs(120))
.map_err(CodexDaemonError::Health)?;
Ok((traffic_admission, publisher))
) -> 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_idunn_health_publisher(
mut publisher: crate::idunn_health::ProviderHealthPublisher,
fn start_periodic_runtime_presence_publisher(
mut publisher: crate::idunn_health::RuntimePresencePublisher,
backend: Arc<CodexProviderBackend>,
) -> Result<(), CodexDaemonError> {
let (state, detail) = provider_health_observation(&backend);
publisher
.publish(state, detail)
.map_err(CodexDaemonError::Health)?;
thread::Builder::new()
.name("codex-connector-health".to_string())
.spawn(move || {
@ -434,26 +416,131 @@ fn provider_health_state(ready: bool) -> (&'static str, &'static str) {
}
#[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 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(),
)
}
#[derive(Clone)]
pub(crate) struct LoadedCallerKeyBinding {
pub(crate) caller_runtime_id: String,
pub(crate) connection_key_file: PathBuf,
pub(crate) connection_key_epoch: u32,
pub(crate) raw_file_sha256: [u8; 32],
#[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>, Vec<LoadedCallerKeyBinding>), CodexDaemonError> {
) -> Result<Vec<CodexCallerAdmission>, CodexDaemonError> {
let mut admissions = Vec::with_capacity(config.callers.len());
let mut bindings = 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)?,
@ -473,22 +560,14 @@ fn load_admissions(
)
.map_err(CodexDaemonError::Service)?,
);
bindings.push(LoadedCallerKeyBinding {
caller_runtime_id: caller.caller_runtime_id.clone(),
connection_key_file: caller.connection_key_file.clone(),
connection_key_epoch: caller.connection_key_epoch,
raw_file_sha256: sha2::Sha256::digest(raw.as_slice()).into(),
});
}
Ok((admissions, bindings))
Ok(admissions)
}
fn serve_connection(
mut stream: TcpStream,
service: &Mutex<CodexTransportService>,
backend: &CodexProviderBackend,
#[cfg(target_os = "linux")]
traffic_admission: &crate::idunn_health::ConnectorTrafficAdmissionGate,
max_frame_bytes: usize,
timeout: Duration,
) -> Result<(), CodexDaemonError> {
@ -499,13 +578,6 @@ fn serve_connection(
let request = read_transport_frame(&mut stream, max_frame_bytes).map_err(frame_error)?;
let envelope: CodexTransportEnvelope =
rmp_serde::from_slice(&request).map_err(|_| CodexDaemonError::FrameEncoding)?;
#[cfg(target_os = "linux")]
// This is the traffic-to-Connector authority transfer. The complete
// request is inert until the current root grant admits service.begin;
// backend consequence remains owned by the admitted request claim.
traffic_admission
.require_current()
.map_err(CodexDaemonError::Health)?;
let admission = service
.lock()
.map_err(|_| CodexDaemonError::ServicePoisoned)?
@ -592,7 +664,7 @@ pub enum CodexDaemonError {
#[error("connector request thread failed")]
Thread(#[source] std::io::Error),
#[cfg(target_os = "linux")]
#[error("connector signed health or traffic authority failed: {0}")]
#[error("connector runtime presence failed: {0}")]
Health(#[source] anyhow::Error),
#[cfg(target_os = "linux")]
#[error("connector health thread could not start")]
@ -661,10 +733,8 @@ mod tests {
let expected = config(&root);
write_daemon_config(&path, &expected).unwrap();
assert_eq!(load_daemon_config(&path).unwrap(), expected);
let exact_bytes = fs::read(&path).unwrap();
let (loaded, loaded_bytes) = load_daemon_config_snapshot(&path).unwrap();
let loaded = load_daemon_config_read_only(&path).unwrap();
assert_eq!(loaded, expected);
assert_eq!(loaded_bytes, exact_bytes);
let mut stale = expected;
stale.epoch = 0;
@ -689,12 +759,60 @@ mod tests {
let lock_path = path.with_file_name("connector.cc.lock");
fs::remove_file(&lock_path).unwrap();
assert!(load_daemon_config_snapshot(&path).is_err());
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");
@ -782,6 +900,66 @@ mod tests {
));
}
#[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();

File diff suppressed because it is too large Load diff

View file

@ -4,7 +4,7 @@ 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_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 --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<_>>();
@ -29,14 +29,19 @@ fn main() {
}
return;
}
let config = match args.as_slice() {
[flag, path] if flag == "--config" => PathBuf::from(path),
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);
}

View file

@ -635,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,
@ -697,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> {

1
vendor/cultcache-rs vendored Submodule

@ -0,0 +1 @@
Subproject commit 73b2555bbcd5f05f091d2cf17fcb86ac30695cae

1
vendor/cultnet-rs vendored Submodule

@ -0,0 +1 @@
Subproject commit 9dcb88df1361b157a48502f43093fe7a25ca22b7