diff --git a/architecture/sandbox.md b/architecture/sandbox.md index 0484abc8e..ecd105be2 100644 --- a/architecture/sandbox.md +++ b/architecture/sandbox.md @@ -659,7 +659,18 @@ sequenceDiagram GW->>DB: Build fresh authoritative bootstrap GW->>SUP: SessionAccepted(bootstrap) SUP->>SUP: Initialize from bootstrap - SUP->>GW: ConfigBootstrapResult + SUP->>GW: ConfigBootstrapResult(component outcomes, admission) + GW->>DB: Persist admission for the delivered generation + GW->>SUP: ConfigurationAdmission(durable state) + alt Configuration accepted + SUP->>SUP: Launch workload + else Configuration rejected + GW->>SUP: ConfigUpdate(repaired snapshot) + SUP->>GW: ConfigUpdateResult(component outcome, admission) + GW->>DB: Persist accepted repair + GW->>SUP: ConfigurationAdmission(accepted) + SUP->>SUP: Launch workload on the same supervisor + end ``` The gateway can send complete component replacements on the accepted stream @@ -682,21 +693,25 @@ without changing publishers. Provider payloads can contain credentials, so the gateway does not persist or render complete stream messages in logs. -The supervisor currently parses and ignores stream-delivered configuration. -Polling remains the only path that changes runtime state and repairs dropped or -unavailable delivery. The gateway serializes construction per sandbox and -component, and coalesces repeated mutations into the latest full snapshot. An -enqueue result means only that the local stream queue accepted the message. A -bounded scope fanout scheduler coalesces repeated workspace and global changes, -and semaphores sized from the database pool bound delivery workers and snapshot -builds. Fanout waits for worker capacity before admitting each recipient, so a -fleet-wide change cannot create a fleet-sized task backlog or saturate the store -and credential backends. Snapshot construction has a deadline that starts once -a build holds a permit, and the gateway rejects encoded stream messages that -approach the transport decoder limit. A later migration will apply these -payloads directly and acknowledge their exact revisions before removing -supervisor polling. At that point, the gateway will require a valid bootstrap -before marking a session ready. +Current-protocol supervisors apply stream-delivered configuration and report +both component outcomes and admission for the exact gateway-authored generation. +The gateway validates generation identity, persists admission, and returns that +durable state on the stream. The supervisor holds the workload boundary until +it receives an accepted acknowledgement. Rejection leaves the stream and +supervisor alive so a later complete replacement can repair the generation and +release the same workload. Compatibility protocol revisions continue using +polling. + +The gateway serializes construction per sandbox and component, and coalesces +repeated mutations into the latest full snapshot. An enqueue result means only +that the local stream queue accepted the message. A bounded scope fanout +scheduler coalesces repeated workspace and global changes, and semaphores sized +from the database pool bound delivery workers and snapshot builds. Fanout waits +for worker capacity before admitting each recipient, so a fleet-wide change +cannot create a fleet-sized task backlog or saturate the store and credential +backends. Snapshot construction has a deadline that starts once a build holds a +permit, and the gateway rejects encoded stream messages that approach the +transport decoder limit. ## Configuration Admission diff --git a/crates/openshell-server/src/compute/mod.rs b/crates/openshell-server/src/compute/mod.rs index 839bfae01..9c1cb48cf 100644 --- a/crates/openshell-server/src/compute/mod.rs +++ b/crates/openshell-server/src/compute/mod.rs @@ -3358,7 +3358,7 @@ impl ComputeRuntime { .await } - pub async fn supervisor_session_initialized( + pub async fn supervisor_session_admission( &self, sandbox_id: &str, instance_id: &str, @@ -3475,7 +3475,10 @@ impl ComputeRuntime { status.exit_code = None; if let Some(admission) = admission { status.configuration_admission = Some(admission.clone()); - status.configuration_activated = Some(true); + status.configuration_activated = Some( + admission.state + == i32::from(openshell_core::proto::ConfigurationAdmissionState::Accepted), + ); } sandbox.set_phase(SandboxPhase::Ready as i32); } else { @@ -10507,45 +10510,59 @@ mod tests { } #[tokio::test] - async fn supervisor_session_initialized_accepts_streamed_configuration_atomically() { + async fn supervisor_session_admission_preserves_repair_on_same_instance() { use openshell_core::proto::{ConfigurationAdmissionState, SandboxConfigurationAdmission}; let runtime = test_runtime(Arc::new(TestDriver::default())).await; - let mut sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); - sandbox - .status - .get_or_insert_default() - .configuration_admission = Some(SandboxConfigurationAdmission { - instance_id: "pending-fence".into(), - state: ConfigurationAdmissionState::Pending.into(), - ..Default::default() - }); + let sandbox = sandbox_record("sb-1", "sandbox-a", SandboxPhase::Provisioning); runtime.store.put_message(&sandbox).await.unwrap(); - - let admission = SandboxConfigurationAdmission { - instance_id: "test-generation".into(), - state: ConfigurationAdmissionState::Accepted.into(), + let mut admission = SandboxConfigurationAdmission { + instance_id: "configuration-1".into(), + state: ConfigurationAdmissionState::Rejected.into(), policy_version: 7, policy_hash: "policy-hash".into(), config_revision: 11, provider_env_revision: 13, - error: String::new(), + error: "invalid image policy".into(), }; + runtime - .supervisor_session_initialized("sb-1", "test-generation", &admission) + .supervisor_session_admission("sb-1", "supervisor-1", &admission) .await .unwrap(); - - let stored = runtime + let rejected = runtime .store .get_message::("sb-1") .await .unwrap() .unwrap(); - let status = stored.status.as_ref().unwrap(); - assert_eq!(status.configuration_admission.as_ref(), Some(&admission)); - assert_eq!(status.configuration_activated, Some(true)); - assert_eq!(stored.phase(), SandboxPhase::Ready as i32); + assert_eq!(rejected.phase(), SandboxPhase::Provisioning as i32); + assert_eq!( + rejected.status.as_ref().unwrap().configuration_activated, + Some(false) + ); + + admission.state = ConfigurationAdmissionState::Accepted.into(); + admission.error.clear(); + runtime + .supervisor_session_admission("sb-1", "supervisor-1", &admission) + .await + .unwrap(); + let repaired = runtime + .store + .get_message::("sb-1") + .await + .unwrap() + .unwrap(); + assert_eq!(repaired.phase(), SandboxPhase::Ready as i32); + assert_eq!( + repaired.status.as_ref().unwrap().main_process_instance_id, + "supervisor-1" + ); + assert_eq!( + repaired.status.as_ref().unwrap().configuration_activated, + Some(true) + ); } #[tokio::test] diff --git a/crates/openshell-server/src/storage_proto.rs b/crates/openshell-server/src/storage_proto.rs index 3e3a2c4df..f1f60da7b 100644 --- a/crates/openshell-server/src/storage_proto.rs +++ b/crates/openshell-server/src/storage_proto.rs @@ -162,7 +162,7 @@ mod tests { const STORAGE_V1_SCHEMA_SHA256: &str = "d68401809d8cea445c35233ef32412bbd041cb2ac5acaf368a0d0bf74d2ddf17"; const PUBLIC_RPC_SCHEMA_SHA256: &str = - "6c9b49049f0de12aecb68621b139553e35d6fa8fae2e35e4002c75e5e1bdbe60"; + "ab978d653fc8bff9f7dac633dd651b015fc73aecf3704d1a43f3b32e0143167f"; const DURABLE_SCHEMA_SHA256: &str = "965a8a09fa80168906a4f9cc6169d3623b98f5281c367695f2f2898230915dd0"; const PUBLIC_DURABLE_OVERLAP_SHA256: &str = diff --git a/crates/openshell-server/src/supervisor_session.rs b/crates/openshell-server/src/supervisor_session.rs index 7b8f7be7d..43594cbaf 100644 --- a/crates/openshell-server/src/supervisor_session.rs +++ b/crates/openshell-server/src/supervisor_session.rs @@ -21,9 +21,10 @@ use openshell_core::proto::{ ConfigApplyOutcome, ConfigBootstrap, ConfigComponent, ConfigComponentApplyResult, ConfigSnapshotRevision, ConfigUpdate, ConfigUpdateResult, GatewayMessage, PolicySource, ProviderReadinessObservation, RelayFrame, RelayInit, RelayOpen, ReportMainProcessExitRequest, - ReportMainProcessExitResponse, Sandbox, SandboxPhase, SessionAccepted, SessionRejected, - SshRelayTarget, StartupConfigCandidate, SupervisorMessage, config_snapshot_revision, - config_update, gateway_message, relay_open, startup_config_prepared, supervisor_message, + ReportMainProcessExitResponse, Sandbox, SandboxConfigurationAdmission, SandboxPhase, + SessionAccepted, SessionRejected, SshRelayTarget, StartupConfigCandidate, SupervisorMessage, + config_snapshot_revision, config_update, gateway_message, relay_open, startup_config_prepared, + supervisor_message, }; use openshell_core::proto::{ LEGACY_SUPERVISOR_PROTOCOL_REVISION, PREVIOUS_SUPERVISOR_PROTOCOL_REVISION, @@ -124,9 +125,83 @@ struct InFlightConfigUpdate { update_id: String, component_sequence: u64, revision: ConfigSnapshotRevision, + admission: Option, sent_at: Instant, } +struct CompletedConfigUpdate { + outcome: ConfigApplyOutcome, + admission: Option, +} + +fn expected_configuration_admission( + snapshot: &openshell_core::proto::SandboxConfigSnapshot, +) -> SandboxConfigurationAdmission { + use openshell_core::proto::ConfigurationAdmissionState; + + SandboxConfigurationAdmission { + instance_id: snapshot.configuration_instance_id.clone(), + state: if snapshot.configuration_admitted { + ConfigurationAdmissionState::Accepted.into() + } else { + ConfigurationAdmissionState::Rejected.into() + }, + policy_version: snapshot.version, + policy_hash: snapshot.policy_hash.clone(), + config_revision: snapshot.config_revision, + provider_env_revision: snapshot.provider_env_revision, + error: snapshot.configuration_error.clone(), + } +} + +fn validate_configuration_admission( + reported: Option<&SandboxConfigurationAdmission>, + expected: &SandboxConfigurationAdmission, + outcome: ConfigApplyOutcome, +) -> Result { + use openshell_core::proto::ConfigurationAdmissionState; + + let reported = reported.ok_or_else(|| { + Status::invalid_argument("sandbox configuration admission result is required") + })?; + if reported.instance_id != expected.instance_id + || reported.policy_version != expected.policy_version + || reported.policy_hash != expected.policy_hash + || reported.config_revision != expected.config_revision + || reported.provider_env_revision != expected.provider_env_revision + { + return Err(Status::invalid_argument( + "configuration admission does not match the delivered generation", + )); + } + let expected_state = ConfigurationAdmissionState::try_from(expected.state).unwrap_or_default(); + let reported_state = ConfigurationAdmissionState::try_from(reported.state).unwrap_or_default(); + if expected_state == ConfigurationAdmissionState::Rejected { + if reported_state != ConfigurationAdmissionState::Rejected { + return Err(Status::invalid_argument( + "rejected gateway configuration was reported as accepted", + )); + } + return Ok(expected.clone()); + } + if outcome_acknowledges_revision(outcome) + && reported_state == ConfigurationAdmissionState::Accepted + { + let mut admission = expected.clone(); + admission.error.clear(); + return Ok(admission); + } + if reported_state == ConfigurationAdmissionState::Rejected { + let mut admission = expected.clone(); + admission.state = ConfigurationAdmissionState::Rejected.into(); + admission.error = "effective configuration could not be activated".to_string(); + return Ok(admission); + } + Err(Status::invalid_argument( + "configuration admission is inconsistent with the apply result", + )) +} + /// Idempotency state for tool server endpoint-status reports from one live supervisor. #[derive(Clone, Debug, PartialEq, Eq)] pub(crate) struct EndpointReportCursor { @@ -184,8 +259,9 @@ fn build_config_update( ) -> (GatewayMessage, InFlightConfigUpdate) { state.sequence = state.sequence.saturating_add(1); let component_sequence = state.sequence; - let (component, revision) = match message { + let (component, revision, admission) = match message { SupervisorConfigMessage::SandboxConfig(snapshot) => { + let admission = expected_configuration_admission(&snapshot); let revision = ConfigSnapshotRevision { component: Some(config_snapshot_revision::Component::SandboxConfig( openshell_core::proto::SandboxConfigRevision { @@ -197,7 +273,11 @@ fn build_config_update( }, )), }; - (config_update::Component::SandboxConfig(*snapshot), revision) + ( + config_update::Component::SandboxConfig(*snapshot), + revision, + Some(admission), + ) } SupervisorConfigMessage::ProviderEnvironment(snapshot) => { let revision = ConfigSnapshotRevision { @@ -208,6 +288,7 @@ fn build_config_update( ( config_update::Component::ProviderEnvironment(snapshot), revision, + None, ) } }; @@ -224,6 +305,7 @@ fn build_config_update( update_id, component_sequence, revision, + admission, sent_at: Instant::now(), }, ) @@ -495,7 +577,7 @@ impl SupervisorSessionRegistry { sandbox_id: &str, session_id: &str, result: &ConfigUpdateResult, - ) -> Result<(), Status> { + ) -> Result { let component = result .result .as_ref() @@ -533,6 +615,10 @@ impl SupervisorSessionRegistry { )); } let outcome = validate_component_apply_result(component_result, &in_flight.revision)?; + let completed = CompletedConfigUpdate { + outcome, + admission: in_flight.admission.clone(), + }; if outcome_acknowledges_revision(outcome) { delivery_state.last_acknowledged_revision = Some(in_flight.revision.clone()); } @@ -541,7 +627,7 @@ impl SupervisorSessionRegistry { if delivery_state.last_acknowledged_revision.as_ref() == Some(&config_message_revision(&pending)) { - return Ok(()); + return Ok(completed); } let (message, next) = build_config_update(delivery_state, pending); match session.tx.try_send(message) { @@ -551,7 +637,7 @@ impl SupervisorSessionRegistry { } } } - Ok(()) + Ok(completed) } /// Record a successfully persisted bootstrap result in the live session's @@ -1288,18 +1374,10 @@ async fn accept_supervisor_session( outbound_tx: mpsc::Sender, mut inbound: tonic::Streaming, ) -> Result<(), Status> { - let bootstrap_admission = bootstrap.as_ref().and_then(|bootstrap| { - let snapshot = bootstrap.sandbox_config.as_ref()?; - Some(openshell_core::proto::SandboxConfigurationAdmission { - instance_id: instance_id.clone(), - state: openshell_core::proto::ConfigurationAdmissionState::Accepted.into(), - policy_version: snapshot.version, - policy_hash: snapshot.policy_hash.clone(), - config_revision: snapshot.config_revision, - provider_env_revision: snapshot.provider_env_revision, - error: String::new(), - }) - }); + let expected_bootstrap_admission = bootstrap + .as_ref() + .and_then(|bootstrap| bootstrap.sandbox_config.as_ref()) + .map(expected_configuration_admission); let expected_bootstrap_revisions = bootstrap .as_ref() .map(bootstrap_revision_fence) @@ -1389,7 +1467,7 @@ async fn accept_supervisor_session( } if !stream_applies_config - && !mark_supervisor_initialized(&state, &sandbox_id, &session_id, &instance_id, None).await + && !mark_supervisor_initialized(&state, &sandbox_id, &session_id, &instance_id).await { state .supervisor_sessions @@ -1436,7 +1514,7 @@ async fn accept_supervisor_session( &instance_id, stream_applies_config, &expected_bootstrap_revisions, - bootstrap_admission.as_ref(), + expected_bootstrap_admission.as_ref(), &session_tx, &mut inbound, shutdown_rx, @@ -1888,7 +1966,7 @@ async fn run_session_loop( instance_id: &str, stream_applies_config: bool, expected_bootstrap_revisions: &[(ConfigComponent, ConfigSnapshotRevision)], - bootstrap_admission: Option<&openshell_core::proto::SandboxConfigurationAdmission>, + expected_bootstrap_admission: Option<&SandboxConfigurationAdmission>, tx: &mpsc::Sender, inbound: &mut tonic::Streaming, mut shutdown_rx: oneshot::Receiver<()>, @@ -1910,7 +1988,7 @@ async fn run_session_loop( msg = inbound.message() => { match msg { Ok(Some(msg)) => { - let bootstrap_succeeded = match msg.payload.as_ref() { + let bootstrap_result = match msg.payload.as_ref() { Some(supervisor_message::Payload::ConfigBootstrapResult(result)) if stream_applies_config => { @@ -1918,7 +1996,7 @@ async fn run_session_loop( &result.results, expected_bootstrap_revisions, ) { - Ok(succeeded) => Some(succeeded), + Ok(succeeded) => Some((succeeded, result.admission.clone())), Err(error) => { warn!( sandbox_id, @@ -1936,29 +2014,45 @@ async fn run_session_loop( state, sandbox_id, session_id, + instance_id, stream_applies_config, + tx, msg, ).await; - match bootstrap_succeeded { - Some(true) => { - if !mark_supervisor_initialized( + if let Some((succeeded, reported_admission)) = bootstrap_result { + bootstrap_complete = true; + let Some(expected_admission) = expected_bootstrap_admission else { + warn!(sandbox_id, session_id, "supervisor bootstrap omitted expected admission"); + break; + }; + let outcome = if succeeded { + ConfigApplyOutcome::Applied + } else { + ConfigApplyOutcome::Unsupported + }; + let admission = match validate_configuration_admission( + reported_admission.as_ref(), + expected_admission, + outcome, + ) { + Ok(admission) => admission, + Err(error) => { + warn!(sandbox_id, session_id, error = %error, "invalid supervisor bootstrap admission"); + break; + } + }; + if !persist_and_ack_admission( state, sandbox_id, session_id, instance_id, - bootstrap_admission, + &admission, + tx, ) .await { break; } - bootstrap_complete = true; - } - Some(false) => { - warn!(sandbox_id, session_id, "supervisor configuration bootstrap failed"); - break; - } - None => {} } } Ok(None) => { @@ -2010,7 +2104,9 @@ async fn handle_supervisor_message( state: &Arc, sandbox_id: &str, session_id: &str, + instance_id: &str, stream_applies_config: bool, + tx: &mpsc::Sender, msg: SupervisorMessage, ) { match msg.payload { @@ -2049,18 +2145,21 @@ async fn handle_supervisor_message( ); } Some(supervisor_message::Payload::ConfigUpdateResult(result)) => { - if let Err(error) = state + let completed = match state .supervisor_sessions .complete_config_update(sandbox_id, session_id, &result) { - debug!( - sandbox_id, - session_id, - error = %error, - "ignored unmatched supervisor configuration result" - ); - return; - } + Ok(completed) => completed, + Err(error) => { + debug!( + sandbox_id, + session_id, + error = %error, + "ignored unmatched supervisor configuration result" + ); + return; + } + }; if let Some(result) = result.result.as_ref() && let Err(error) = record_component_apply_result(state, sandbox_id, result).await { @@ -2074,6 +2173,29 @@ async fn handle_supervisor_message( error = %error, "failed to persist supervisor configuration result" ); + return; + } + if let Some(expected_admission) = completed.admission.as_ref() { + let admission = match validate_configuration_admission( + result.admission.as_ref(), + expected_admission, + completed.outcome, + ) { + Ok(admission) => admission, + Err(error) => { + warn!(sandbox_id, session_id, error = %error, "invalid supervisor configuration admission"); + return; + } + }; + let _ = persist_and_ack_admission( + state, + sandbox_id, + session_id, + instance_id, + &admission, + tx, + ) + .await; } } Some(supervisor_message::Payload::ConfigBootstrapResult(result)) => { @@ -2190,7 +2312,6 @@ async fn mark_supervisor_initialized( sandbox_id: &str, session_id: &str, instance_id: &str, - admission: Option<&openshell_core::proto::SandboxConfigurationAdmission>, ) -> bool { if !state .supervisor_sessions @@ -2198,18 +2319,11 @@ async fn mark_supervisor_initialized( { return false; } - let result = if let Some(admission) = admission { - state - .compute - .supervisor_session_initialized(sandbox_id, instance_id, admission) - .await - } else { - state - .compute - .supervisor_session_connected(sandbox_id, instance_id) - .await - }; - if let Err(err) = result { + if let Err(err) = state + .compute + .supervisor_session_connected(sandbox_id, instance_id) + .await + { warn!( sandbox_id, session_id, @@ -2223,6 +2337,43 @@ async fn mark_supervisor_initialized( } } +async fn persist_and_ack_admission( + state: &Arc, + sandbox_id: &str, + session_id: &str, + instance_id: &str, + admission: &SandboxConfigurationAdmission, + tx: &mpsc::Sender, +) -> bool { + if !state + .supervisor_sessions + .is_current_session(sandbox_id, session_id) + { + return false; + } + if let Err(error) = state + .compute + .supervisor_session_admission(sandbox_id, instance_id, admission) + .await + { + warn!(sandbox_id, session_id, error = %error, "failed to persist supervisor configuration admission"); + return false; + } + if tx + .send(GatewayMessage { + payload: Some(gateway_message::Payload::ConfigurationAdmission( + admission.clone(), + )), + }) + .await + .is_err() + { + return false; + } + state.telemetry.sandbox_session_connected(sandbox_id); + true +} + async fn record_component_apply_result( state: &Arc, sandbox_id: &str, @@ -2387,7 +2538,24 @@ mod tests { })), }); - for original in std::iter::once(bootstrap).chain(updates) { + let admission = SandboxConfigurationAdmission { + instance_id: "configuration-1".into(), + state: openshell_core::proto::ConfigurationAdmissionState::Accepted.into(), + policy_version: 3, + policy_hash: "policy-hash".into(), + config_revision: 7, + provider_env_revision: 5, + error: String::new(), + }; + let admission_ack = GatewayMessage { + payload: Some(gateway_message::Payload::ConfigurationAdmission( + admission.clone(), + )), + }; + for original in std::iter::once(bootstrap) + .chain(updates) + .chain(std::iter::once(admission_ack)) + { let decoded = GatewayMessage::decode(original.encode_to_vec().as_slice()).unwrap(); assert_eq!(decoded, original); } @@ -2420,6 +2588,7 @@ mod tests { outcome: ConfigApplyOutcome::Applied.into(), failure: None, }), + admission: Some(admission), }, )), }; @@ -2434,6 +2603,38 @@ mod tests { assert!(validate_protocol_revision("sb-1", LEGACY_SUPERVISOR_PROTOCOL_REVISION).is_ok()); } + #[test] + fn configuration_admission_preserves_gateway_rejection_and_accepts_repair() { + use openshell_core::proto::ConfigurationAdmissionState; + + let mut expected = SandboxConfigurationAdmission { + instance_id: "configuration-1".into(), + state: ConfigurationAdmissionState::Rejected.into(), + policy_version: 3, + policy_hash: "policy-hash".into(), + config_revision: 7, + provider_env_revision: 5, + error: "invalid image policy".into(), + }; + let rejected = validate_configuration_admission( + Some(&expected), + &expected, + ConfigApplyOutcome::Applied, + ) + .unwrap(); + assert_eq!(rejected, expected); + + expected.state = ConfigurationAdmissionState::Accepted.into(); + expected.error.clear(); + let accepted = validate_configuration_admission( + Some(&expected), + &expected, + ConfigApplyOutcome::Applied, + ) + .unwrap(); + assert_eq!(accepted, expected); + } + #[test] fn bootstrap_requires_every_delivered_component_to_succeed() { let sandbox_revision = ConfigSnapshotRevision { @@ -2518,7 +2719,7 @@ mod tests { state.supervisor_sessions.register( "sb-bootstrap-ack".into(), "session-1".into(), - tx, + tx.clone(), shutdown_tx, ); let snapshot = ProviderEnvironmentSnapshot { @@ -2540,11 +2741,14 @@ mod tests { &state, "sb-bootstrap-ack", "session-1", + "instance-1", true, + &tx, SupervisorMessage { payload: Some(supervisor_message::Payload::ConfigBootstrapResult( ConfigBootstrapResult { results: vec![result], + admission: None, }, )), }, @@ -3183,6 +3387,7 @@ mod tests { outcome: ConfigApplyOutcome::Applied.into(), failure: None, }), + admission: None, }, ) .unwrap(); diff --git a/crates/openshell-supervisor-process/src/delegated.rs b/crates/openshell-supervisor-process/src/delegated.rs index ad319c37f..188fb3242 100644 --- a/crates/openshell-supervisor-process/src/delegated.rs +++ b/crates/openshell-supervisor-process/src/delegated.rs @@ -9,7 +9,8 @@ use std::time::Duration; use miette::Result; use openshell_isolation_interface::contract::{ - BoundaryExec, BoundaryLoopbackConnector, BoundaryProcess, + BackendError, BoundaryDuplexStream, BoundaryExec, BoundaryLoopbackConnector, BoundaryProcess, + LoopbackTarget, }; use openshell_ocsf::{ActivityId, AppLifecycleBuilder, SeverityId, StatusId, ocsf_emit}; @@ -27,6 +28,105 @@ pub struct BoundaryAccess { main_session: Option>, } +#[derive(Default)] +struct DeferredLoopbackConnector { + connector: tokio::sync::RwLock>>, + ready: tokio::sync::Notify, +} + +impl DeferredLoopbackConnector { + async fn install(&self, connector: Arc) { + *self.connector.write().await = Some(connector); + self.ready.notify_waiters(); + } +} + +#[async_trait::async_trait] +impl BoundaryLoopbackConnector for DeferredLoopbackConnector { + async fn connect(&self, target: LoopbackTarget) -> Result { + loop { + let notified = self.ready.notified(); + let connector = self.connector.read().await.clone(); + if let Some(connector) = connector { + return connector.connect(target).await; + } + notified.await; + } + } +} + +/// A supervisor stream that has reported configuration admission while the +/// boundary remains held in its confirmed, pre-workload state. +pub struct PrestartedSupervisorSession { + terminating: Arc, + task: Option>, + readiness: tokio::sync::watch::Receiver, + loopback: Arc, +} + +impl Drop for PrestartedSupervisorSession { + fn drop(&mut self) { + if let Some(task) = self.task.take() { + self.terminating.store(true, Ordering::Release); + task.abort(); + } + } +} + +/// Resume a prepared supervisor stream and report its bootstrap result before +/// workload activation. Relay requests wait until the running boundary installs +/// its loopback connector. +pub async fn start_prepared_supervisor_session( + prepared: crate::supervisor_session::PreparedSupervisorSession, + bootstrap_result: Option, + ssh_socket_path: Option<&str>, + config_apply_tx: tokio::sync::mpsc::Sender, + supervisor_session_updates: Option>>, +) -> Result { + let terminating = Arc::new(AtomicBool::new(false)); + let loopback = Arc::new(DeferredLoopbackConnector::default()); + let target = ssh_socket_path.map_or_else( + || std::path::PathBuf::from(openshell_core::container_paths::SSH_SOCKET_PATH), + std::path::PathBuf::from, + ); + let (task, mut readiness) = crate::supervisor_session::spawn_prepared( + prepared, + bootstrap_result, + target, + loopback.clone(), + None, + terminating.clone(), + config_apply_tx, + supervisor_session_updates, + ); + let ready = tokio::time::timeout( + crate::supervisor_session::SESSION_PREPARE_TIMEOUT, + readiness.wait_for(|ready| *ready), + ) + .await + .map(|result| result.map(|_| ())); + match ready { + Ok(Ok(())) => Ok(PrestartedSupervisorSession { + terminating, + task: Some(task), + readiness, + loopback, + }), + Ok(Err(_)) => { + task.abort(); + Err(miette::miette!( + "prepared supervisor session ended before bootstrap acknowledgement" + )) + } + Err(_) => { + task.abort(); + Err(miette::miette!( + "prepared supervisor session did not receive accepted configuration admission before the provisioning deadline" + )) + } + } +} + impl BoundaryAccess { /// Stable supervisor instance ID used for lifecycle reporting. #[must_use] @@ -88,59 +188,22 @@ pub async fn start_boundary_access( port_forward: Arc, agent: Arc, supervisor_session_updates: Option>>, - prepared_supervisor_session: Option, - prepared_bootstrap_result: Option, + mut prestarted_supervisor_session: Option, config_apply_tx: Option< tokio::sync::mpsc::Sender, >, ) -> Result { - let terminating = Arc::new(AtomicBool::new(false)); + if let Some(prestarted) = prestarted_supervisor_session.as_ref() { + prestarted.loopback.install(port_forward.clone()).await; + } + let terminating = prestarted_supervisor_session.as_ref().map_or_else( + || Arc::new(AtomicBool::new(false)), + |prestarted| prestarted.terminating.clone(), + ); let Some(ssh_socket_path) = ssh_socket_path.map(std::path::PathBuf::from) else { - let (session_task, session_readiness) = match (prepared_supervisor_session, config_apply_tx) - { - (Some(prepared), Some(config_apply_tx)) => { - // Configuration admission is carried by the supervisor stream, - // even when this sandbox does not expose SSH. Keep the stream - // alive and use the canonical socket path only as the target for - // any unexpected relay request; no listener is started here. - let (task, mut accepted) = crate::supervisor_session::spawn_prepared( - prepared, - prepared_bootstrap_result, - std::path::PathBuf::from(openshell_core::container_paths::SSH_SOCKET_PATH), - port_forward, - None, - terminating.clone(), - config_apply_tx, - supervisor_session_updates, - ); - let accepted_result = tokio::time::timeout( - Duration::from_secs(10), - accepted.wait_for(|ready| *ready), - ) - .await - .map(|result| result.map(|_| ())); - match accepted_result { - Ok(Ok(())) => (Some(task), Some(accepted)), - Ok(Err(_)) => { - task.abort(); - return Err(miette::miette!( - "prepared supervisor session ended before bootstrap acknowledgement" - )); - } - Err(_) => { - task.abort(); - return Err(miette::miette!( - "prepared supervisor session did not become ready within 10 seconds" - )); - } - } - } - (Some(_), None) => { - return Err(miette::miette!( - "prepared supervisor session requires a configuration apply channel" - )); - } - (None, _) => (None, None), + let (session_task, session_readiness) = match prestarted_supervisor_session.as_mut() { + Some(prestarted) => (prestarted.task.take(), Some(prestarted.readiness.clone())), + None => (None, None), }; return Ok(BoundaryAccess { instance_id, @@ -205,78 +268,47 @@ pub async fn start_boundary_access( } } - let (session_task, session_readiness) = match ( - prepared_supervisor_session, - config_apply_tx.clone(), - openshell_endpoint, - sandbox_id, - ) { - (Some(prepared), Some(config_apply_tx), _, _) => { - let (task, mut accepted) = crate::supervisor_session::spawn_prepared( - prepared, - prepared_bootstrap_result, - ssh_socket_path, - port_forward, - None, - terminating.clone(), - config_apply_tx, - supervisor_session_updates, - ); - let accepted_result = - tokio::time::timeout(Duration::from_secs(10), accepted.wait_for(|ready| *ready)) - .await - .map(|result| result.map(|_| ())); - match accepted_result { - Ok(Ok(())) => (Some(task), Some(accepted)), - Ok(Err(_)) => { - task.abort(); - return Err(miette::miette!( - "prepared supervisor session ended before bootstrap acknowledgement" - )); - } - Err(_) => { - task.abort(); - return Err(miette::miette!( - "prepared supervisor session did not become ready within 10 seconds" - )); + let (session_task, session_readiness) = match prestarted_supervisor_session.as_mut() { + Some(prestarted) => (prestarted.task.take(), Some(prestarted.readiness.clone())), + None => match (openshell_endpoint, sandbox_id) { + (Some(endpoint), Some(id)) => { + let (task, mut accepted) = crate::supervisor_session::spawn_with_readiness( + endpoint.to_string(), + id.to_string(), + ssh_socket_path, + port_forward, + None, + terminating.clone(), + crate::supervisor_session::SessionRuntimeContext { + instance_id: instance_id.clone(), + session_id_updates: supervisor_session_updates, + config_apply_tx, + }, + ); + let accepted_result = tokio::time::timeout( + Duration::from_secs(10), + accepted.wait_for(|ready| *ready), + ) + .await + .map(|result| result.map(|_| ())); + match accepted_result { + Ok(Ok(())) => (Some(task), Some(accepted)), + Ok(Err(_)) => { + task.abort(); + return Err(miette::miette!( + "supervisor session ended before gateway acceptance" + )); + } + Err(_) => { + task.abort(); + return Err(miette::miette!( + "gateway did not accept supervisor session within 10 seconds" + )); + } } } - } - (None, _, Some(endpoint), Some(id)) => { - let (task, mut accepted) = crate::supervisor_session::spawn_with_readiness( - endpoint.to_string(), - id.to_string(), - ssh_socket_path, - port_forward, - None, - terminating.clone(), - crate::supervisor_session::SessionRuntimeContext { - instance_id: instance_id.clone(), - session_id_updates: supervisor_session_updates, - config_apply_tx, - }, - ); - let accepted_result = - tokio::time::timeout(Duration::from_secs(10), accepted.wait_for(|ready| *ready)) - .await - .map(|result| result.map(|_| ())); - match accepted_result { - Ok(Ok(())) => (Some(task), Some(accepted)), - Ok(Err(_)) => { - task.abort(); - return Err(miette::miette!( - "supervisor session ended before gateway acceptance" - )); - } - Err(_) => { - task.abort(); - return Err(miette::miette!( - "gateway did not accept supervisor session within 10 seconds" - )); - } - } - } - _ => (None, None), + _ => (None, None), + }, }; Ok(BoundaryAccess { diff --git a/crates/openshell-supervisor-process/src/supervisor_session.rs b/crates/openshell-supervisor-process/src/supervisor_session.rs index 2acde43b5..0c06a9afa 100644 --- a/crates/openshell-supervisor-process/src/supervisor_session.rs +++ b/crates/openshell-supervisor-process/src/supervisor_session.rs @@ -47,7 +47,7 @@ const MAX_BACKOFF: Duration = Duration::from_secs(30); const CONFIG_APPLY_TIMEOUT: Duration = Duration::from_mins(1); // The gateway may hold the initial stream open while an operator repairs an // invalid image policy. Keep this above the five-minute provisioning window. -const SESSION_PREPARE_TIMEOUT: Duration = Duration::from_mins(6); +pub const SESSION_PREPARE_TIMEOUT: Duration = Duration::from_mins(6); type StartupPolicyPreparer = Box Result, String> + Send>; @@ -734,7 +734,9 @@ async fn run_prepared_session( .map_err(|_| "failed to queue configuration bootstrap result")?; } let config_sequences = Arc::new(Mutex::new(ConfigSequenceWatermarks::default())); - config.ready_tx.send_replace(true); + if prepared.protocol_revision != SUPERVISOR_PROTOCOL_REVISION { + config.ready_tx.send_replace(true); + } // Main loop: receive gateway messages + send heartbeats. let mut heartbeat_interval = @@ -762,6 +764,7 @@ async fn run_prepared_session( terminating: &config.terminating, config_apply_tx: config.config_apply_tx.as_ref(), config_sequences: &config_sequences, + ready_tx: &config.ready_tx, }; handle_gateway_message( &msg, @@ -824,6 +827,7 @@ async fn apply_bootstrap( ) }) .collect(), + admission: None, }; }; let (response, receiver) = tokio::sync::oneshot::channel(); @@ -837,12 +841,14 @@ async fn apply_bootstrap( { return ConfigBootstrapResult { results: Vec::new(), + admission: None, }; } match tokio::time::timeout(CONFIG_APPLY_TIMEOUT, receiver).await { Ok(Ok(result)) => result, _ => ConfigBootstrapResult { results: Vec::new(), + admission: None, }, } } @@ -930,6 +936,7 @@ struct GatewayMessageContext<'a> { terminating: &'a Arc, config_apply_tx: Option<&'a mpsc::Sender>, config_sequences: &'a Arc>, + ready_tx: &'a watch::Sender, } fn handle_gateway_message(msg: &GatewayMessage, context: &GatewayMessageContext<'_>) { @@ -937,6 +944,11 @@ fn handle_gateway_message(msg: &GatewayMessage, context: &GatewayMessageContext< Some(gateway_message::Payload::Heartbeat(_)) => { // Gateway heartbeat — nothing to do. } + Some(gateway_message::Payload::ConfigurationAdmission(admission)) => { + let accepted = admission.state + == i32::from(openshell_core::proto::ConfigurationAdmissionState::Accepted); + context.ready_tx.send_replace(accepted); + } Some(gateway_message::Payload::ConfigUpdate(update)) => { let update = update.clone(); let tx = context.tx.clone(); @@ -971,6 +983,7 @@ fn handle_gateway_message(msg: &GatewayMessage, context: &GatewayMessageContext< result: Some(failed_component_result( component, revision, outcome, code, message, )), + admission: None, }; let result = if invalid_update { fallback( diff --git a/crates/openshell-supervisor/src/lib.rs b/crates/openshell-supervisor/src/lib.rs index 3bced922a..5fff44782 100644 --- a/crates/openshell-supervisor/src/lib.rs +++ b/crates/openshell-supervisor/src/lib.rs @@ -809,6 +809,7 @@ pub async fn run_sandbox( let mut prepared_bootstrap_result = stream_bootstrap.as_ref().map(|bootstrap| { use openshell_core::proto::{ConfigApplyOutcome, ConfigBootstrapResult, ConfigComponent}; let mut results = Vec::with_capacity(2); + let mut admission = None; if let Some(snapshot) = bootstrap.provider_environment.as_ref() { let revision = provider_config_revision(snapshot.provider_env_revision); let outcome = if provider_bootstrap_degraded { @@ -827,22 +828,31 @@ pub async fn run_sandbox( if let Some(snapshot) = bootstrap.sandbox_config.as_ref() { let settings: openshell_core::grpc_client::SettingsPollResult = snapshot.clone().into(); let revision = sandbox_config_revision(&settings); - let outcome = if loaded_policy_origin.allows_gateway_policy_reload() { + let outcome = if !snapshot.configuration_admitted { + ConfigApplyOutcome::FailedClosed + } else if loaded_policy_origin.allows_gateway_policy_reload() { ConfigApplyOutcome::Applied } else { ConfigApplyOutcome::RetainedLocalOverride }; let applied_revision = - (outcome != ConfigApplyOutcome::RetainedLocalOverride).then_some(revision.clone()); + matches!(outcome, ConfigApplyOutcome::Applied).then_some(revision.clone()); results.push(config_apply_result( ConfigComponent::SandboxConfig, revision, applied_revision, outcome, - None, + (!snapshot.configuration_admitted).then(|| { + ( + "configuration_rejected", + snapshot.configuration_error.clone(), + true, + ) + }), )); + admission = Some(streamed_configuration_admission(snapshot, true)); } - ConfigBootstrapResult { results } + ConfigBootstrapResult { results, admission } }); if credential_gating_unavailable( @@ -1131,8 +1141,12 @@ pub async fn run_sandbox( let poll_pid = entrypoint_pid.clone(); let poll_provider_credentials = provider_credentials.clone(); let poll_policy_local = networking.as_ref().map(|n| n.policy_local_ctx.clone()); - let poll_endpoint_policy = loaded_policy_origin - .allows_gateway_policy_reload() + let initial_stream_configuration_admitted = stream_bootstrap + .as_ref() + .and_then(|bootstrap| bootstrap.sandbox_config.as_ref()) + .is_none_or(|snapshot| snapshot.configuration_admitted); + let poll_endpoint_policy = (loaded_policy_origin.allows_gateway_policy_reload() + && initial_stream_configuration_admitted) .then(|| retained_proto.clone()) .flatten(); let poll_interval_secs: u64 = std::env::var("OPENSHELL_POLICY_POLL_INTERVAL_SECS") @@ -1172,9 +1186,10 @@ pub async fn run_sandbox( supervisor_session_id: supervisor_session_id.clone(), }; + let wait_for_workload = !uses_stream_configuration; tokio::spawn(async move { let mut workload_started = workload_started_rx; - if workload_started.wait_for(|started| *started).await.is_err() { + if wait_for_workload && workload_started.wait_for(|started| *started).await.is_err() { return; } if let Err(e) = run_policy_poll_loop(poll_ctx).await { @@ -1203,6 +1218,24 @@ pub async fn run_sandbox( }; tokio::pin!(proxy_exited); + // Report the stream-delivered bootstrap result while the boundary remains + // confirmed but held. Docker and other admission-aware drivers do not + // release the workload until the gateway accepts this result, so delaying + // the stream until after `start_agent` would deadlock activation. + let prestarted_supervisor_session = match prepared_supervisor_session.take() { + Some(prepared) => Some( + openshell_supervisor_process::delegated::start_prepared_supervisor_session( + prepared, + prepared_bootstrap_result.take(), + ssh_socket_path.as_deref(), + config_apply_tx.clone(), + Some(supervisor_session_updates.clone()), + ) + .await?, + ), + None => None, + }; + let (confirmed, backend_name) = remote_ready; let exit_code = { let running = confirmed @@ -1226,8 +1259,7 @@ pub async fn run_sandbox( running.loopback_connector(), agent.clone(), Some(supervisor_session_updates), - prepared_supervisor_session, - prepared_bootstrap_result.take(), + prestarted_supervisor_session, Some(config_apply_tx), ) .await?; @@ -2499,17 +2531,32 @@ async fn load_policy( } } + let configuration_rejected = stream_bootstrap && !snapshot.configuration_admitted; let mut loaded_policy_revision = policy_bound_to_snapshot.then(|| LoadedPolicyRevision::from_snapshot(&snapshot)); + if configuration_rejected { + ocsf_emit!( + ConfigStateChangeBuilder::new(ocsf_ctx()) + .severity(SeverityId::High) + .status(StatusId::Failure) + .state(StateId::Other, "configuration_error") + .message(snapshot.configuration_error.clone()) + .build() + ); + proto_policy = openshell_policy::restrictive_default_policy(); + } + // Build OPA engine from baked-in rules + typed proto data. // In cluster mode, proxy networking is always enabled so OPA is // always required for allow/deny decisions. // The initial load uses pid=0 (no symlink resolution) because the // container hasn't started yet. After the entrypoint spawns, the // engine is rebuilt with the real PID for symlink resolution. - info!("Creating OPA engine from proto policy data"); - let mut has_last_valid_policy = true; + if !configuration_rejected { + info!("Creating OPA engine from proto policy data"); + } + let mut has_last_valid_policy = !configuration_rejected; let engine = match OpaEngine::from_proto(&proto_policy) { Ok(engine) => Arc::new(engine), Err(e) => { @@ -2550,13 +2597,18 @@ async fn load_policy( install_builtin_middleware_registry(&engine).await?; - let middleware_services = snapshot.supervisor_middleware_services.clone(); + let middleware_services = if configuration_rejected { + Vec::new() + } else { + snapshot.supervisor_middleware_services.clone() + }; let middleware_registry_status = if middleware_services.is_empty() { MiddlewareRegistryStatus::Synchronized } else if let Err(error) = grpc_retry("Middleware connect", || { let middleware_services = middleware_services.clone(); let extension_credentials = extension_credentials.clone(); - let extension_authentication_enabled = snapshot.extension_authentication_enabled; + let extension_authentication_enabled = + snapshot.extension_authentication_enabled && !configuration_rejected; async move { let credentials = if extension_authentication_enabled { openshell_core::grpc_client::CachedOpenShellClient::connect_with_credentials( @@ -2624,8 +2676,8 @@ async fn load_policy( revision: loaded_policy_revision, has_last_valid_policy, }, - agent_proposals_enabled_from_settings(&snapshot.settings), - snapshot.extension_authentication_enabled, + !configuration_rejected && agent_proposals_enabled_from_settings(&snapshot.settings), + snapshot.extension_authentication_enabled && !configuration_rejected, )); } @@ -3950,6 +4002,48 @@ fn config_apply_result( } } +fn config_apply_result_activates( + result: &openshell_core::proto::ConfigComponentApplyResult, +) -> bool { + use openshell_core::proto::ConfigApplyOutcome; + + matches!( + ConfigApplyOutcome::try_from(result.outcome), + Ok(ConfigApplyOutcome::Applied + | ConfigApplyOutcome::IgnoredDuplicate + | ConfigApplyOutcome::RetainedLocalOverride + | ConfigApplyOutcome::Degraded) + ) +} + +fn streamed_configuration_admission( + snapshot: &openshell_core::proto::SandboxConfigSnapshot, + runtime_applied: bool, +) -> openshell_core::proto::SandboxConfigurationAdmission { + use openshell_core::proto::{ConfigurationAdmissionState, SandboxConfigurationAdmission}; + + let accepted = snapshot.configuration_admitted && runtime_applied; + SandboxConfigurationAdmission { + instance_id: snapshot.configuration_instance_id.clone(), + state: if accepted { + ConfigurationAdmissionState::Accepted.into() + } else { + ConfigurationAdmissionState::Rejected.into() + }, + policy_version: snapshot.version, + policy_hash: snapshot.policy_hash.clone(), + config_revision: snapshot.config_revision, + provider_env_revision: snapshot.provider_env_revision, + error: if accepted { + String::new() + } else if snapshot.configuration_error.is_empty() { + "effective configuration could not be activated".to_string() + } else { + snapshot.configuration_error.clone() + }, + } +} + #[allow(clippy::too_many_arguments)] async fn apply_stream_config_request( ctx: &PolicyPollLoopContext, @@ -3980,6 +4074,7 @@ async fn apply_stream_config_request( response, } => { let mut results = Vec::with_capacity(2); + let mut admission = None; if let Some(snapshot) = bootstrap.provider_environment { let result = apply_stream_provider_snapshot(ctx, snapshot, current_provider_env_revision); @@ -3999,32 +4094,38 @@ async fn apply_stream_config_request( results.push(result); } if let Some(snapshot) = bootstrap.sandbox_config { - results.push( - apply_stream_sandbox_snapshot( - ctx, - client, - snapshot.into(), - current_config_revision, - current_stream_sandbox_revision, - current_policy_version, - current_policy_hash, - current_endpoint_policy, - current_middleware_services, - current_extension_authentication_enabled, - middleware_registry_status, - current_settings, - reloads_gateway_policy, - has_last_valid_policy, - ) - .await, - ); + let expected_admission = snapshot.clone(); + let result = apply_stream_sandbox_snapshot( + ctx, + client, + snapshot.into(), + current_config_revision, + current_stream_sandbox_revision, + current_policy_version, + current_policy_hash, + current_endpoint_policy, + current_middleware_services, + current_extension_authentication_enabled, + middleware_registry_status, + current_settings, + reloads_gateway_policy, + has_last_valid_policy, + ) + .await; + admission = Some(streamed_configuration_admission( + &expected_admission, + config_apply_result_activates(&result), + )); + results.push(result); } - let _ = response.send(ConfigBootstrapResult { results }); + let _ = response.send(ConfigBootstrapResult { results, admission }); } ConfigApplyRequest::Update { update, response } => { + let mut admission = None; let result = match update.component { Some(config_update::Component::SandboxConfig(snapshot)) => { - apply_stream_sandbox_snapshot( + let expected_admission = snapshot.clone(); + let result = apply_stream_sandbox_snapshot( ctx, client, snapshot.into(), @@ -4040,7 +4141,12 @@ async fn apply_stream_config_request( reloads_gateway_policy, has_last_valid_policy, ) - .await + .await; + admission = Some(streamed_configuration_admission( + &expected_admission, + config_apply_result_activates(&result), + )); + result } Some(config_update::Component::ProviderEnvironment(snapshot)) => { let result = apply_stream_provider_snapshot( @@ -4079,6 +4185,7 @@ async fn apply_stream_config_request( update_id: update.update_id, component_sequence: update.component_sequence, result: Some(result), + admission, }); } } @@ -4443,24 +4550,29 @@ async fn run_policy_poll_loop_with_client( .map_or(0, |snapshot| snapshot.config_revision); let mut current_stream_sandbox_revision = initial_stream_snapshot .as_ref() + .filter(|snapshot| snapshot.configuration_admitted) .map(sandbox_config_revision) .filter(|_| ctx.loaded_policy_origin.allows_gateway_policy_reload()); let mut current_provider_env_revision: u64 = ctx.provider_credentials.snapshot().revision; let mut current_policy_version: u32 = initial_stream_snapshot .as_ref() + .filter(|snapshot| snapshot.configuration_admitted) .map_or(0, |snapshot| snapshot.version); let mut current_policy_hash = initial_stream_snapshot .as_ref() + .filter(|snapshot| snapshot.configuration_admitted) .map_or_else(String::new, |snapshot| snapshot.policy_hash.clone()); let mut current_policy_generation = None; let mut current_endpoint_policy = ctx.endpoint_policy.clone(); let mut current_middleware_services = initial_stream_snapshot .as_ref() + .filter(|snapshot| snapshot.configuration_admitted) .map_or_else(Vec::new, |snapshot| { snapshot.supervisor_middleware_services.clone() }); let mut current_extension_authentication_enabled = initial_stream_snapshot .as_ref() + .filter(|snapshot| snapshot.configuration_admitted) .map_or(ctx.extension_authentication_enabled, |snapshot| { snapshot.extension_authentication_enabled }); @@ -4470,6 +4582,7 @@ async fn run_policy_poll_loop_with_client( openshell_core::proto::EffectiveSetting, > = initial_stream_snapshot .as_ref() + .filter(|snapshot| snapshot.configuration_admitted) .map_or_else(std::collections::HashMap::new, |snapshot| { snapshot.settings.clone() }); @@ -4494,7 +4607,9 @@ async fn run_policy_poll_loop_with_client( // reconciled below instead of being recorded as already applied. if let Some(snapshot) = initial_stream_snapshot.as_ref() { let _ = ctx.workspace_tx.send(snapshot.workspace.clone()); - if let Some(generation) = initial_generation { + if snapshot.configuration_admitted + && let Some(generation) = initial_generation + { ctx.provider_readiness.policy_activated( &EnvironmentIdentity::from_settings(snapshot), snapshot.config_revision, @@ -4503,13 +4618,15 @@ async fn run_policy_poll_loop_with_client( current_policy_generation = Some(generation); } apply_ocsf_json_setting(&ctx.ocsf_enabled, &snapshot.settings); - endpoint_status::reset( - ctx.endpoint_observation_tx.as_ref(), - current_endpoint_policy.as_ref(), - ¤t_policy_hash, - ctx.provider_credentials.snapshot().revision, - ) - .await; + if snapshot.configuration_admitted { + endpoint_status::reset( + ctx.endpoint_observation_tx.as_ref(), + current_endpoint_policy.as_ref(), + ¤t_policy_hash, + ctx.provider_credentials.snapshot().revision, + ) + .await; + } } else { match client.poll_settings(&ctx.sandbox).await { Ok(result) => { diff --git a/proto/openshell.proto b/proto/openshell.proto index cfd760d59..45218def0 100644 --- a/proto/openshell.proto +++ b/proto/openshell.proto @@ -2875,6 +2875,9 @@ message GatewayMessage { RelayClose relay_close = 5; ConfigUpdate config_update = 6; StartupConfigCandidate startup_config_candidate = 7; + // Durable gateway acknowledgement of streamed configuration admission. + // The supervisor must not launch the workload until this is ACCEPTED. + SandboxConfigurationAdmission configuration_admission = 8; } } @@ -2994,11 +2997,16 @@ message ConfigUpdateResult { // Echoes ConfigUpdate.component_sequence. uint64 component_sequence = 2; ConfigComponentApplyResult result = 3; + // Admission state for a sandbox-configuration update. Omitted for provider + // environment updates and compatibility supervisors. + SandboxConfigurationAdmission admission = 4; } // Aggregate application result for the SessionAccepted bootstrap. message ConfigBootstrapResult { repeated ConfigComponentApplyResult results = 1; + // Admission state of the complete gateway-authored bootstrap generation. + SandboxConfigurationAdmission admission = 2; } // Gateway rejects the supervisor session.