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

Expose managed Connector route presence

Browse source
This commit is contained in:
Meta Vens 2026-09-03 09:27:00 +02:00
commit 78b4331ee5
2 changed files with 336 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

217
src/daemon.rs
View file

@ -15,6 +15,12 @@ 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,
@ -321,8 +327,11 @@ pub fn serve(
config.max_expiry_skew_ms,
)?));
#[cfg(target_os = "linux")]
if let Some(publisher) = prepared_presence_publisher {
start_periodic_runtime_presence_publisher(publisher, backend.clone())?;
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!(
@ -349,13 +358,23 @@ pub fn serve(
};
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)?;
}
@ -379,11 +398,13 @@ fn prepare_runtime_presence_publisher(
#[cfg(target_os = "linux")]
fn start_periodic_runtime_presence_publisher(
mut publisher: crate::idunn_health::RuntimePresencePublisher,
publisher: Arc<Mutex<crate::idunn_health::RuntimePresencePublisher>>,
backend: Arc<CodexProviderBackend>,
) -> Result<(), CodexDaemonError> {
let (state, detail) = provider_health_observation(&backend);
publisher
.lock()
.map_err(|_| CodexDaemonError::RuntimePresencePoisoned)?
.publish(state, detail)
.map_err(CodexDaemonError::Health)?;
thread::Builder::new()
@ -392,7 +413,11 @@ fn start_periodic_runtime_presence_publisher(
loop {
thread::sleep(Duration::from_secs(10));
let (state, detail) = provider_health_observation(&backend);
if let Err(error) = publisher.publish(state, detail) {
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}");
}
}
@ -568,6 +593,9 @@ 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> {
@ -576,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)?
@ -592,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 {
@ -669,6 +763,12 @@ pub enum CodexDaemonError {
#[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")]
@ -682,6 +782,8 @@ pub enum CodexDaemonError {
#[cfg(test)]
mod tests {
use super::*;
#[cfg(target_os = "linux")]
use std::cell::Cell;
use std::io::Cursor;
fn config(root: &Path) -> CodexDaemonConfig {
@ -721,6 +823,105 @@ mod tests {
);
}
#[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!(

143
src/idunn_health.rs
View file

@ -32,7 +32,7 @@ const SYSTEMD_LISTEN_FDS_START: RawFd = 3;
const ACTIVATION_SIGNER_FD_NAME: &str = IDUNN_RUNTIME_ACTIVATION_CREDENTIAL_NAME;
const PROVIDER_SIGNER_FD_NAME: &str = "gamecult-runtime-presence-identity";
const CONNECTOR_TARGET: &str = "codex-connector";
pub(crate) const CONNECTOR_TARGET: &str = "codex-connector";
const CONNECTOR_CAPABILITY: &str = "gamecult.codex.subscription-inference";
const CONNECTOR_COMPATIBILITY: &str = "v2";
pub(crate) const CONNECTOR_HEALTH_CONTRACT: &str = "codex-connector.runtime-health.v1";
@ -219,6 +219,48 @@ impl RuntimePresencePublisher {
.context("Connector has not acquired its process-write lease")
}
pub(crate) fn route_observation(
&mut self,
message_id: &str,
backend_ready: bool,
) -> Result<CultNetMessage> {
ensure!(
!message_id.is_empty(),
"route-observation message id is empty"
);
let guard = self
.write_lease
.as_ref()
.context("Connector route observation lacks its process-write lease")?;
guard.require_current()?;
self.signed_route_observation(message_id, backend_ready)
}
fn signed_route_observation(
&mut self,
message_id: &str,
backend_ready: bool,
) -> Result<CultNetMessage> {
let state = if backend_ready { "active" } else { "degraded" };
let detail = format!("route-observation:{message_id}");
let record = self.signed_record(state, &detail, unix_millis()?)?;
let payload = rmp_serde::to_vec(&record)?;
ensure!(
rmp_serde::from_slice::<GameCultRuntimePresenceHealthRecord>(&payload)? == record,
"route observation did not round-trip canonically"
);
Ok(CultNetMessage::SnapshotResponseRaw {
message_id: message_id.into(),
documents: vec![presence_document(
&self.authority.expected.target,
&self.authority.expected.runtime_id,
&record,
payload,
None,
)?],
})
}
pub(crate) fn publish(&mut self, state: &str, detail: &str) -> Result<String> {
if let Some(guard) = &self.write_lease {
guard.require_current()?;
@ -627,21 +669,13 @@ fn publish_presence(
"codex-connector-presence:{}:{}:{}",
target, record.runtime_instance_id, record.publisher_sequence
),
document: CultNetRawDocumentRecord {
schema_id: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(),
record_key: target.into(),
stored_at: chrono::DateTime::from_timestamp_millis(
record.observed_at_unix_millis.try_into()?,
)
.context("runtime presence observation time is invalid")?
.to_rfc3339(),
payload_encoding: CultNetRawPayloadEncoding::Messagepack,
document: presence_document(
target,
runtime_id,
record,
payload,
source_runtime_id: Some(runtime_id.into()),
source_agent_id: Some(record.signer_identity_id.clone()),
source_role: Some("runtime-presence-health-publisher".into()),
tags: Some(vec![RUDP_PROTOCOL_ID.into()]),
},
Some(vec![RUDP_PROTOCOL_ID.into()]),
)?,
};
let socket = UdpSocket::bind(if endpoint.is_ipv4() {
"0.0.0.0:0"
@ -688,6 +722,30 @@ fn publish_presence(
Ok(())
}
fn presence_document(
target: &str,
runtime_id: &str,
record: &GameCultRuntimePresenceHealthRecord,
payload: Vec<u8>,
tags: Option<Vec<String>>,
) -> Result<CultNetRawDocumentRecord> {
Ok(CultNetRawDocumentRecord {
schema_id: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(),
record_key: target.into(),
stored_at: chrono::DateTime::from_timestamp_millis(
record.observed_at_unix_millis.try_into()?,
)
.context("runtime presence observation time is invalid")?
.to_rfc3339(),
payload_encoding: CultNetRawPayloadEncoding::Messagepack,
payload,
source_runtime_id: Some(runtime_id.into()),
source_agent_id: Some(record.signer_identity_id.clone()),
source_role: Some("runtime-presence-health-publisher".into()),
tags,
})
}
fn require_runtime_bundle(bundle: &Path) -> Result<()> {
ensure!(
bundle.is_absolute(),
@ -1104,6 +1162,7 @@ mod tests {
File::create(&lock_path)?;
let lock = OpenOptions::new().read(true).open(lock_path)?;
fs2::FileExt::lock_shared(&lock)?;
let lease_sha256 = sha('8');
publisher.write_lease = Some(ProcessWriteLeaseGuard {
held: Arc::new(HeldProcessWriteLease {
_lock: lock,
@ -1113,7 +1172,7 @@ mod tests {
activation: activation.clone(),
activation_sha256: activation.canonical_sha256()?,
warming_presence_sha256: warming.canonical_sha256()?,
lease_sha256: sha('8'),
lease_sha256: lease_sha256.clone(),
}),
});
let active = publisher.signed_record("active", "ready", issued_at)?;
@ -1138,6 +1197,58 @@ mod tests {
},
)?;
assert_eq!(claim.record(), &active);
let route_response = publisher.signed_route_observation("challenge-23", true)?;
let CultNetMessage::SnapshotResponseRaw {
message_id,
documents,
} = route_response
else {
panic!("route observation was not a raw snapshot response");
};
assert_eq!(message_id, "challenge-23");
let [document] = documents.as_slice() else {
panic!("route observation did not contain exactly one document");
};
assert_eq!(document.schema_id, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA);
assert_eq!(document.record_key, CONNECTOR_TARGET);
assert_eq!(document.tags, None);
let route_presence: GameCultRuntimePresenceHealthRecord =
rmp_serde::from_slice(&document.payload)?;
assert_eq!(route_presence.state, "active");
assert_eq!(route_presence.detail, "route-observation:challenge-23");
assert_eq!(
route_presence.write_lease_sha256.as_deref(),
Some(lease_sha256.as_str())
);
assert_eq!(route_presence.publisher_sequence, 3);
let claim = authenticate_runtime_presence_claim(
&document.payload,
&authority,
RuntimePresenceAuthenticationContext {
trusted_received_at_unix_millis: route_presence.observed_at_unix_millis,
maximum_age_millis: 1_000,
maximum_future_skew_millis: 1_000,
},
)?;
assert_eq!(claim.record(), &route_presence);
let degraded = publisher.signed_route_observation("challenge-24", false)?;
let CultNetMessage::SnapshotResponseRaw { documents, .. } = degraded else {
panic!("degraded route observation was not a raw snapshot response");
};
let degraded: GameCultRuntimePresenceHealthRecord =
rmp_serde::from_slice(&documents[0].payload)?;
assert_eq!(degraded.state, "degraded");
assert_eq!(degraded.detail, "route-observation:challenge-24");
// A held historical guard is not enough: every route challenge must
// re-read the exact Idunn lease before signing its data-plane proof.
assert!(
publisher
.route_observation("challenge-stale", true)
.is_err()
);
let contender = OpenOptions::new()
.read(true)
.write(true)