From 92566480bb0d2d883cd994156bc53bf213b0c38b Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Thu, 3 Sep 2026 09:25:32 +0200 Subject: [PATCH] Expose managed Ghostlight route presence --- crates/ghostlight-dungeon/src/idunn_health.rs | 187 ++++++++-- crates/ghostlight-dungeon/src/runtime.rs | 345 ++++++++++++++++-- 2 files changed, 479 insertions(+), 53 deletions(-) diff --git a/crates/ghostlight-dungeon/src/idunn_health.rs b/crates/ghostlight-dungeon/src/idunn_health.rs index 2b192b7..c576c0d 100644 --- a/crates/ghostlight-dungeon/src/idunn_health.rs +++ b/crates/ghostlight-dungeon/src/idunn_health.rs @@ -41,7 +41,7 @@ pub const IDUNN_RUNTIME_BUNDLE_ENVIRONMENT: &str = "GAMECULT_IDUNN_RUNTIME_BUNDL pub const IDUNN_RUNTIME_CANDIDATE_BIND_ENVIRONMENT: &str = "GAMECULT_IDUNN_CANDIDATE_BIND"; pub const IDUNN_PROCESS_WRITE_LEASE_ENVIRONMENT: &str = "GAMECULT_IDUNN_PROCESS_WRITE_LEASE"; -const TARGET: &str = "ghostlight"; +pub(crate) const TARGET: &str = "ghostlight"; const STATE_SCHEMA_GENERATION: &str = "world-v2"; const STATE_CONTRACT_SHA256: &str = "sha256-bf6ec06d885a59ddb237c6224d0abb4ccceac8c7ba23761d1326d7f562a4c21e"; @@ -305,6 +305,32 @@ impl RuntimePresencePublisher { self.publish("active", "world-owner-serving", Some(lease_sha256)) } + pub(crate) fn route_observation( + &mut self, + message_id: &str, + ready: bool, + lease: &ProcessWriteLeaseGuard, + ) -> Result { + ensure!( + !message_id.is_empty(), + "route-observation message id is empty" + ); + lease.require_current()?; + let lease_sha256 = lease.canonical_sha256(); + ensure!( + self.active_write_lease_sha256.as_deref() == Some(lease_sha256), + "route observation does not carry the active process write lease" + ); + let state = if ready { "active" } else { "degraded" }; + let detail = format!("route-observation:{message_id}"); + let record = self.signed_record(state, &detail, Some(lease_sha256.into()))?; + let payload = canonical_presence_payload(&record)?; + Ok(CultNetMessage::SnapshotResponseRaw { + message_id: message_id.into(), + documents: vec![runtime_presence_document(&record, payload, None)?], + }) + } + fn publish( &mut self, state: &str, @@ -953,36 +979,21 @@ fn publish_runtime_presence( record: &GameCultRuntimePresenceHealthRecord, ) -> Result { record.validate()?; - let payload = rmp_serde::to_vec(record).context("encoding canonical runtime presence")?; - let decoded: GameCultRuntimePresenceHealthRecord = rmp_serde::from_slice(&payload)?; - ensure!( - decoded == *record && rmp_serde::to_vec(&decoded)? == payload, - "runtime presence encoding is noncanonical" - ); + let payload = canonical_presence_payload(record)?; let canonical_sha256 = record.canonical_sha256()?; let message = CultNetMessage::DocumentPutRaw { message_id: format!( "runtime-presence:{}:{}:{}", record.target, record.runtime_instance_id, record.publisher_sequence ), - document: CultNetRawDocumentRecord { - schema_id: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(), - record_key: record.target.clone(), - stored_at: chrono::DateTime::from_timestamp_millis( - record.observed_at_unix_millis.try_into()?, - ) - .context("runtime presence timestamp is invalid")? - .to_rfc3339(), - payload_encoding: CultNetRawPayloadEncoding::Messagepack, + document: runtime_presence_document( + record, payload, - source_runtime_id: Some(record.runtime_id.clone()), - source_agent_id: Some(record.signer_identity_id.clone()), - source_role: Some("runtime-presence-publisher".into()), - tags: Some(vec![ + Some(vec![ CULTNET_RUDP_PROTOCOL_ID.into(), "runtime-presence".into(), ]), - }, + )?, }; let bind = if endpoint.is_ipv4() { "0.0.0.0:0" @@ -1032,6 +1043,38 @@ fn publish_runtime_presence( Ok(canonical_sha256) } +fn canonical_presence_payload(record: &GameCultRuntimePresenceHealthRecord) -> Result> { + let payload = rmp_serde::to_vec(record).context("encoding canonical runtime presence")?; + let decoded: GameCultRuntimePresenceHealthRecord = rmp_serde::from_slice(&payload)?; + ensure!( + decoded == *record && rmp_serde::to_vec(&decoded)? == payload, + "runtime presence encoding is noncanonical" + ); + Ok(payload) +} + +fn runtime_presence_document( + record: &GameCultRuntimePresenceHealthRecord, + payload: Vec, + tags: Option>, +) -> Result { + Ok(CultNetRawDocumentRecord { + schema_id: GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into(), + record_key: record.target.clone(), + stored_at: chrono::DateTime::from_timestamp_millis( + record.observed_at_unix_millis.try_into()?, + ) + .context("runtime presence timestamp is invalid")? + .to_rfc3339(), + payload_encoding: CultNetRawPayloadEncoding::Messagepack, + payload, + source_runtime_id: Some(record.runtime_id.clone()), + source_agent_id: Some(record.signer_identity_id.clone()), + source_role: Some("runtime-presence-publisher".into()), + tags, + }) +} + fn sibling_lock_path(path: &Path) -> PathBuf { let mut name = path .file_name() @@ -1042,7 +1085,7 @@ fn sibling_lock_path(path: &Path) -> PathBuf { } #[cfg(test)] -mod tests { +pub(crate) mod tests { use super::*; use cultnet_rs::{ GameCultRuntimePresenceHealthPurpose, IdunnExpectedCapability, IdunnExpectedRoute, @@ -1061,6 +1104,12 @@ mod tests { activation: IdunnRuntimeActivationRecord, } + pub(crate) struct RouteObservationFixture { + pub(crate) publisher: RuntimePresencePublisher, + pub(crate) write_lease: ProcessWriteLeaseGuard, + pub(crate) authority: cultnet_rs::VerifiedRuntimeAuthority, + } + fn managed_dependency( kind: &str, capability: &str, @@ -1189,6 +1238,51 @@ mod tests { fixture_with_state_contract(root, STATE_CONTRACT_SHA256) } + pub(crate) fn route_observation_fixture(root: &Path) -> Result { + let mut fixture = fixture(root)?; + let warming = + fixture + .publisher + .signed_record("warming", "process-write-lease-pending", None)?; + let warming_sha256 = warming.canonical_sha256()?; + let lease = IdunnProcessWriteLeaseRecord { + schema_version: IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into(), + target: fixture.expected.target.clone(), + expected_projection_sha256: fixture.expected.canonical_sha256()?, + plan_id: fixture.expected.plan_id.clone(), + incarnation_id: fixture.expected.incarnation_id.clone(), + sealed_release_id: fixture.expected.sealed_release_id.clone(), + activation_witness_sha256: fixture.activation.canonical_sha256()?, + state_schema_generation: STATE_SCHEMA_GENERATION.into(), + state_contract_sha256: fixture.expected.state_contract_sha256.clone().unwrap(), + runtime_id: fixture.expected.runtime_id.clone(), + runtime_instance_id: fixture.activation.runtime_instance_id.clone(), + warming_presence_sha256: warming_sha256.clone(), + lease_epoch: 1, + issued_at_unix_millis: 200, + }; + let path = root.join("write-lease.cc"); + SingleFileMessagePackBackingStore::new(&path).push(&CultCacheEnvelope { + key: lease.target.clone(), + r#type: IdunnProcessWriteLeaseRecord::TYPE.into(), + payload: lease.canonical_bytes()?, + stored_at: "2026-09-03T00:00:00Z".into(), + schema_id: Some(IDUNN_PROCESS_WRITE_LEASE_SCHEMA.into()), + })?; + let write_lease = ProcessWriteLeaseGuard::acquire_fixture( + path, + fixture.expected, + fixture.activation, + warming_sha256, + )?; + fixture.publisher.active_write_lease_sha256 = Some(write_lease.canonical_sha256().into()); + Ok(RouteObservationFixture { + publisher: fixture.publisher, + write_lease, + authority: fixture.authority, + }) + } + #[cfg(target_os = "linux")] #[test] fn inherited_authority_fd_layout_rejects_substitution_and_extras() -> Result<()> { @@ -1361,6 +1455,55 @@ mod tests { )?; guard.require_current()?; assert_eq!(guard.canonical_sha256(), lease.canonical_sha256()?); + + let lease_sha256 = lease.canonical_sha256()?; + fixture.publisher.active_write_lease_sha256 = Some(lease_sha256.clone()); + let response = fixture + .publisher + .route_observation("challenge-31", true, &guard)?; + let CultNetMessage::SnapshotResponseRaw { + message_id, + documents, + } = response + else { + panic!("route observation was not a raw snapshot response"); + }; + assert_eq!(message_id, "challenge-31"); + 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, TARGET); + assert_eq!(document.tags, None); + let presence: GameCultRuntimePresenceHealthRecord = + rmp_serde::from_slice(&document.payload)?; + assert_eq!(presence.state, "active"); + assert_eq!(presence.detail, "route-observation:challenge-31"); + assert_eq!( + presence.write_lease_sha256.as_deref(), + Some(lease_sha256.as_str()) + ); + let claim = authenticate_runtime_presence_claim( + &document.payload, + &fixture.authority, + RuntimePresenceAuthenticationContext { + trusted_received_at_unix_millis: presence.observed_at_unix_millis, + maximum_age_millis: 1_000, + maximum_future_skew_millis: 10, + }, + )?; + assert_eq!(claim.record(), &presence); + + let degraded = fixture + .publisher + .route_observation("challenge-32", false, &guard)?; + 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-32"); drop(guard); let mut wrong = lease; diff --git a/crates/ghostlight-dungeon/src/runtime.rs b/crates/ghostlight-dungeon/src/runtime.rs index 6264b67..0b69515 100644 --- a/crates/ghostlight-dungeon/src/runtime.rs +++ b/crates/ghostlight-dungeon/src/runtime.rs @@ -6,6 +6,7 @@ use crate::{ heimdall::{self, HeimdallClient}, idunn_health::{ IDUNN_RUNTIME_CANDIDATE_BIND_ENVIRONMENT, ProcessWriteLeaseGuard, RuntimePresencePublisher, + TARGET as GHOSTLIGHT_TARGET, }, mesh::{self, MeshPublisher, MeshRuntimeIdentity}, world::{ @@ -19,7 +20,8 @@ use crate::{ use anyhow::{Context, bail, ensure}; use axum::{ Json, Router, - extract::{Path, State}, + body::Bytes, + extract::{ConnectInfo, DefaultBodyLimit, Path, State}, http::{HeaderMap, HeaderValue, StatusCode, header}, response::{ IntoResponse, Response, @@ -28,6 +30,10 @@ use axum::{ routing::{get, post}, }; use chrono::{DateTime, Utc}; +use cultnet_rs::{ + CultNetMessage, CultNetWireContract, GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA, + decode_cultnet_message_from_slice, encode_cultnet_message_to_vec, +}; use serde::Deserialize; use serde_json::{Value, json}; use std::{convert::Infallible, fs, net::SocketAddr, path::PathBuf, sync::Arc, time::Duration}; @@ -36,6 +42,13 @@ use tower_http::services::ServeDir; use tracing_subscriber::EnvFilter; const COOKIE_NAME: &str = "ghostlight_session"; +const CULTNET_SNAPSHOT_BODY_LIMIT: usize = 16 * 1024; + +#[derive(Clone)] +struct RuntimeHealthOwner { + publisher: Arc>, + write_lease: Arc, +} #[derive(Clone)] struct AppState { @@ -45,6 +58,7 @@ struct AppState { heimdall: Arc, mesh: Option, mesh_identity: MeshRuntimeIdentity, + runtime_health: Option, revisions: broadcast::Sender, fatal: mpsc::UnboundedSender, } @@ -191,7 +205,7 @@ pub(crate) async fn run(state_root_binding: Option) -> anyhow::Result<( }; require_no_runtime_custody_failure(&mut fatal_events)?; let (revisions, _) = broadcast::channel(32); - let state = AppState { + let mut state = AppState { world, controllers: Arc::new(Mutex::new(controllers)), sessions: Arc::new(Mutex::new(sessions)), @@ -200,10 +214,10 @@ pub(crate) async fn run(state_root_binding: Option) -> anyhow::Result<( mesh_identity: mesh_identity.clone(), revisions, fatal, + runtime_health: None, }; require_no_runtime_custody_failure(&mut fatal_events)?; publish_projection(&state).await?; - tokio::spawn(maintain_mesh_projection(state.clone())); let release_web_root = std::env::current_exe()? .parent() @@ -211,7 +225,6 @@ pub(crate) async fn run(state_root_binding: Option) -> anyhow::Result<( let web_root = release_web_root .filter(|path| path.join("index.html").is_file()) .unwrap_or_else(|| PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../web/dist")); - let app = app_router(state.clone(), web_root); require_no_runtime_custody_failure(&mut fatal_events)?; require_current_write_lease(write_lease_guard.as_ref())?; canonical_readiness(&state) @@ -226,6 +239,16 @@ pub(crate) async fn run(state_root_binding: Option) -> anyhow::Result<( .context("publishing initial active runtime presence")?; } require_current_write_lease(write_lease_guard.as_ref())?; + state.runtime_health = match (idunn_health, write_lease_guard) { + (Some(publisher), Some(write_lease)) => Some(RuntimeHealthOwner { + publisher: Arc::new(Mutex::new(publisher)), + write_lease: Arc::new(write_lease), + }), + (None, None) => None, + _ => bail!("managed runtime health authority is partial"), + }; + tokio::spawn(maintain_mesh_projection(state.clone())); + let app = app_router(state.clone(), web_root); tracing::info!(address = %bound_address, "Ghostlight Dungeon world owner serving"); let server = axum::serve( listener, @@ -237,11 +260,7 @@ pub(crate) async fn run(state_root_binding: Option) -> anyhow::Result<( result.context("world-owner task panicked")?; bail!("world-owner task stopped while the daemon was serving") } - result = maintain_runtime_health( - idunn_health, - state.clone(), - write_lease_guard.as_ref(), - ) => { + result = maintain_runtime_health(state.clone()) => { result?; bail!("runtime readiness owner stopped while the daemon was serving") } @@ -303,6 +322,10 @@ fn app_router(state: AppState, web_root: PathBuf) -> Router { fn api_router(state: AppState) -> Router { Router::new() .route("/health", get(health)) + .route( + "/cultnet/snapshot", + post(cultnet_snapshot).layer(DefaultBodyLimit::max(CULTNET_SNAPSHOT_BODY_LIMIT)), + ) .route("/api/eve/provider", get(eve_provider)) .route("/api/eve/surfaces/{surface_id}", get(eve_surface)) .route("/api/eve/commands", post(eve_command)) @@ -310,6 +333,85 @@ fn api_router(state: AppState) -> Router { .with_state(state) } +async fn cultnet_snapshot( + ConnectInfo(peer): ConnectInfo, + State(state): State, + headers: HeaderMap, + body: Bytes, +) -> Response { + if !peer.ip().is_loopback() { + return StatusCode::FORBIDDEN.into_response(); + } + if headers.get(header::CONTENT_TYPE) != Some(&HeaderValue::from_static("application/msgpack")) { + return StatusCode::UNSUPPORTED_MEDIA_TYPE.into_response(); + } + let message_id = match exact_route_observation_message_id(&body) { + Ok(message_id) => message_id, + Err(_) => return StatusCode::BAD_REQUEST.into_response(), + }; + let Some(owner) = state.runtime_health.clone() else { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + }; + + // Hold the one publisher across readiness observation and signing so a + // periodic publication cannot age the checked state before this challenge + // is bound into the signed record. + let write_lease = owner.write_lease.clone(); + let mut publisher = owner.publisher.clone().lock_owned().await; + let ready = canonical_readiness(&state).await.is_ok(); + let response = tokio::task::spawn_blocking(move || { + let response = publisher.route_observation(&message_id, ready, &write_lease)?; + encode_cultnet_message_to_vec(&response, CultNetWireContract::CultNetSchemaV0) + }) + .await; + let Ok(Ok(response)) = response else { + return StatusCode::SERVICE_UNAVAILABLE.into_response(); + }; + ( + StatusCode::OK, + [(header::CONTENT_TYPE, "application/msgpack")], + response, + ) + .into_response() +} + +fn exact_route_observation_message_id(body: &[u8]) -> anyhow::Result { + let request = decode_cultnet_message_from_slice(body, CultNetWireContract::CultNetSchemaV0) + .context("decoding route-observation request")?; + ensure!( + encode_cultnet_message_to_vec(&request, CultNetWireContract::CultNetSchemaV0)? == body, + "route-observation request is not canonical MessagePack" + ); + let CultNetMessage::SnapshotRequest { + message_id, + schema_ids, + record_keys, + } = request + else { + bail!("route-observation request is not a snapshot request"); + }; + ensure!( + !message_id.is_empty(), + "route-observation message id is empty" + ); + ensure!( + matches!( + schema_ids.as_deref(), + Some([schema_id]) if schema_id == GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA + ) && matches!( + record_keys.as_deref(), + Some([record_key]) if record_key == GHOSTLIGHT_TARGET + ), + "route-observation request is not the exact Ghostlight presence record" + ); + let detail = format!("route-observation:{message_id}"); + ensure!( + detail.len() <= 512 && !detail.chars().any(char::is_control), + "route-observation message id cannot be signed canonically" + ); + Ok(message_id) +} + async fn health(State(state): State) -> Response { match runtime_readiness(&state).await { Ok(health) => Json(health).into_response(), @@ -1464,12 +1566,7 @@ async fn initialize_production_admission( })) } -async fn maintain_runtime_health( - publisher: Option, - state: AppState, - write_lease: Option<&ProcessWriteLeaseGuard>, -) -> anyhow::Result<()> { - let mut publisher = publisher; +async fn maintain_runtime_health(state: AppState) -> anyhow::Result<()> { let mut interval = tokio::time::interval(Duration::from_secs(10)); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); loop { @@ -1477,22 +1574,25 @@ async fn maintain_runtime_health( canonical_readiness(&state) .await .context("runtime is not ready for signed health")?; - require_current_write_lease(write_lease) + let Some(owner) = state.runtime_health.clone() else { + continue; + }; + owner + .write_lease + .require_current() .context("runtime process write lease is not current")?; - if let Some(mut owned) = publisher.take() { - let published = tokio::task::spawn_blocking(move || { - let result = owned.republish_active(); - (owned, result) - }) - .await; - let Ok((returned, result)) = published else { - tracing::warn!("runtime-presence worker panicked; publication is disabled"); - continue; - }; - publisher = Some(returned); - if let Err(error) = result { - tracing::warn!(%error, "signed runtime-presence publication failed"); - } + let mut publisher = owner.publisher.clone().lock_owned().await; + let published = tokio::task::spawn_blocking(move || { + publisher.republish_active()?; + Ok::<(), anyhow::Error>(()) + }) + .await; + let Ok(result) = published else { + tracing::warn!("runtime-presence worker panicked; publication will retry"); + continue; + }; + if let Err(error) = result { + tracing::warn!(%error, "signed runtime-presence publication failed"); } } } @@ -1658,8 +1758,16 @@ fn sibling_state_lock_path(path: &std::path::Path) -> PathBuf { #[cfg(test)] mod tests { use super::*; + use crate::idunn_health::tests::route_observation_fixture; use crate::world::AffordanceKind; - use axum::{body::Body, http::Request}; + use axum::{ + body::{Body, to_bytes}, + http::Request, + }; + use cultnet_rs::{ + GameCultRuntimePresenceHealthRecord, RuntimePresenceAuthenticationContext, + authenticate_runtime_presence_claim, + }; use tower::ServiceExt; #[cfg(target_os = "linux")] @@ -1789,6 +1897,7 @@ mod tests { heimdall: Arc::new(HeimdallClient::fixture()), mesh: Some(mesh), mesh_identity, + runtime_health: None, revisions, fatal, }; @@ -1800,6 +1909,180 @@ mod tests { } } + fn route_snapshot_request( + message_id: &str, + schema_ids: Option>, + record_keys: Option>, + ) -> Vec { + encode_cultnet_message_to_vec( + &CultNetMessage::SnapshotRequest { + message_id: message_id.into(), + schema_ids, + record_keys, + }, + CultNetWireContract::CultNetSchemaV0, + ) + .unwrap() + } + + #[test] + fn route_observation_request_is_one_exact_canonical_record() { + let exact = route_snapshot_request( + "route-challenge-41", + Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]), + Some(vec![GHOSTLIGHT_TARGET.into()]), + ); + assert_eq!( + exact_route_observation_message_id(&exact).unwrap(), + "route-challenge-41" + ); + + for refused in [ + route_snapshot_request("broad", None, None), + route_snapshot_request( + "foreign-schema", + Some(vec!["gamecult.other.v1".into()]), + Some(vec![GHOSTLIGHT_TARGET.into()]), + ), + route_snapshot_request( + "foreign-record", + Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]), + Some(vec!["other-target".into()]), + ), + ] { + assert!(exact_route_observation_message_id(&refused).is_err()); + } + } + + #[tokio::test] + async fn http_route_probe_bypasses_provider_and_enforces_exact_admission() { + let mut fixture = fixture().await; + let body = route_snapshot_request( + "route-challenge-42", + Some(vec![GAMECULT_RUNTIME_PRESENCE_HEALTH_SCHEMA.into()]), + Some(vec![GHOSTLIGHT_TARGET.into()]), + ); + let request = |peer: &str, content_type: Option<&str>, body: Vec| { + let mut builder = Request::builder() + .method("POST") + .uri("/cultnet/snapshot") + .extension(ConnectInfo(peer.parse::().unwrap())); + if let Some(content_type) = content_type { + builder = builder.header(header::CONTENT_TYPE, content_type); + } + builder.body(Body::from(body)).unwrap() + }; + + let unmanaged = api_router(fixture.state.clone()) + .oneshot(request( + "127.0.0.1:39001", + Some("application/msgpack"), + body.clone(), + )) + .await + .unwrap(); + assert_eq!(unmanaged.status(), StatusCode::SERVICE_UNAVAILABLE); + + let remote = api_router(fixture.state.clone()) + .oneshot(request( + "192.0.2.9:39001", + Some("application/msgpack"), + body.clone(), + )) + .await + .unwrap(); + assert_eq!(remote.status(), StatusCode::FORBIDDEN); + + let wrong_media = api_router(fixture.state.clone()) + .oneshot(request( + "127.0.0.1:39001", + Some("application/json"), + body.clone(), + )) + .await + .unwrap(); + assert_eq!(wrong_media.status(), StatusCode::UNSUPPORTED_MEDIA_TYPE); + + let broad = api_router(fixture.state.clone()) + .oneshot(request( + "127.0.0.1:39001", + Some("application/msgpack"), + route_snapshot_request("broad", None, None), + )) + .await + .unwrap(); + assert_eq!(broad.status(), StatusCode::BAD_REQUEST); + + let health = route_observation_fixture(fixture._directory.path()).unwrap(); + let authority = health.authority; + let write_lease_sha256 = health.write_lease.canonical_sha256().to_owned(); + fixture.state.runtime_health = Some(RuntimeHealthOwner { + publisher: Arc::new(Mutex::new(health.publisher)), + write_lease: Arc::new(health.write_lease), + }); + + // If this route enters the controller/provider boundary, the held lock + // makes the request time out. Route admission and signing need only the + // canonical health owner and managed runtime authority. + let controllers = fixture.state.controllers.clone(); + let _provider_execution_barrier = controllers.lock().await; + let response = tokio::time::timeout( + Duration::from_secs(2), + api_router(fixture.state.clone()).oneshot(request( + "127.0.0.1:39001", + Some("application/msgpack"), + body, + )), + ) + .await + .expect("route observation entered provider execution") + .unwrap(); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!( + response.headers().get(header::CONTENT_TYPE), + Some(&HeaderValue::from_static("application/msgpack")) + ); + let bytes = to_bytes(response.into_body(), 64 * 1024).await.unwrap(); + let message = + decode_cultnet_message_from_slice(&bytes, CultNetWireContract::CultNetSchemaV0) + .unwrap(); + assert_eq!( + encode_cultnet_message_to_vec(&message, CultNetWireContract::CultNetSchemaV0).unwrap(), + bytes + ); + let CultNetMessage::SnapshotResponseRaw { + message_id, + documents, + } = message + else { + panic!("route observation was not a raw snapshot response"); + }; + assert_eq!(message_id, "route-challenge-42"); + 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, GHOSTLIGHT_TARGET); + let presence: GameCultRuntimePresenceHealthRecord = + rmp_serde::from_slice(&document.payload).unwrap(); + assert_eq!(presence.state, "active"); + assert_eq!(presence.detail, "route-observation:route-challenge-42"); + assert_eq!( + presence.write_lease_sha256.as_deref(), + Some(write_lease_sha256.as_str()) + ); + authenticate_runtime_presence_claim( + &document.payload, + &authority, + RuntimePresenceAuthenticationContext { + trusted_received_at_unix_millis: presence.observed_at_unix_millis, + maximum_age_millis: 1_000, + maximum_future_skew_millis: 10, + }, + ) + .unwrap(); + } + fn invocation(operation: &str, schema: &str, version: u64, payload: Value, id: &str) -> Value { json!({ "schema":"gamecult.eve.command_invocation.v1",