feat(config): track durable completion only for streamed apply

Keep push delivery opt-in through Stage 3 so the default flip stays a
small follow-up.

- In poll mode, finish each configuration operation as untracked when it
  commits and answer the update without an operation, including for
  WAIT_FOR_COMPLETION.
- In push mode, finish pending operations when a build finds that the
  sandbox's supervisor session does not apply streamed configuration.
- Fall back to polling policy load status in `policy set --wait` and
  `policy update --wait` when the gateway returns no completion operation.

Signed-off-by: Piotr Mlocek <pmlocek@nvidia.com>

crates/openshell-cli/src/run.rs

@@ -152,6 +152,77 @@ fn report_config_update_operation(
     }
 }
 
+/// Wait for a policy revision by polling its load status. Gateways that do not
+/// track completion, such as those that keep configuration polling, return no
+/// completion operation for a waited update.
+async fn wait_for_policy_load(
+    client: &mut crate::tls::GrpcClient,
+    name: &str,
+    workspace: &str,
+    version: u32,
+    timeout_secs: u64,
+) -> Result<i32> {
+    let deadline = Instant::now() + Duration::from_secs(timeout_secs);
+    loop {
+        if Instant::now() > deadline {
+            eprintln!(
+                "{} Timeout waiting for policy version {} to load",
+                "✗".red().bold(),
+                version
+            );
+            return Ok(POLICY_WAIT_TIMEOUT_EXIT_CODE);
+        }
+
+        tokio::time::sleep(Duration::from_secs(1)).await;
+
+        let inner = client
+            .get_sandbox_policy_status(GetSandboxPolicyStatusRequest {
+                sandbox: name.to_string(),
+                workspace_scope: Some(openshell_core::proto::workspace_selector(
+                    workspace.to_string(),
+                )),
+                version,
+                global: false,
+            })
+            .await
+            .into_diagnostic()?
+            .into_inner();
+        let Some(rev) = &inner.revision else {
+            continue;
+        };
+        match PolicyStatus::try_from(rev.status).unwrap_or(PolicyStatus::Unspecified) {
+            PolicyStatus::Loaded => {
+                eprintln!(
+                    "{} Policy version {} loaded (active version: {})",
+                    "✓".green().bold(),
+                    rev.version,
+                    inner.active_version
+                );
+                return Ok(0);
+            }
+            PolicyStatus::Failed => {
+                eprintln!(
+                    "{} Policy version {} failed to load: {}",
+                    "✗".red().bold(),
+                    rev.version,
+                    rev.load_error
+                );
+                return Ok(1);
+            }
+            PolicyStatus::Superseded => {
+                eprintln!(
+                    "{} Policy version {} was superseded (active version: {})",
+                    "⚠".yellow().bold(),
+                    rev.version,
+                    inner.active_version
+                );
+                return Ok(0);
+            }
+            _ => {}
+        }
+    }
+}
+
 fn proto_timestamp_ms(timestamp: Option<&prost_types::Timestamp>) -> i64 {
     timestamp
         .and_then(|value| openshell_core::time::timestamp_to_millis(value).ok())
@@ -5562,6 +5633,10 @@ pub async fn sandbox_policy_set(
         return Ok(0);
     }
 
+    if resp.operation.is_none() {
+        return wait_for_policy_load(&mut client, name, workspace, resp.version, timeout_secs)
+            .await;
+    }
     report_config_update_operation(resp.operation.as_ref(), resp.version)?;
     Ok(0)
 }
@@ -5692,6 +5767,10 @@ pub async fn sandbox_policy_update(
         return Ok(0);
     }
 
+    if response.operation.is_none() {
+        return wait_for_policy_load(&mut client, name, workspace, response.version, timeout_secs)
+            .await;
+    }
     report_config_update_operation(response.operation.as_ref(), response.version)?;
     Ok(0)
 }

crates/openshell-server/src/config_delivery/mod.rs

@@ -311,7 +311,7 @@ fn bootstrap_revisions_match(bootstrap: &ConfigBootstrap) -> bool {
         })
 }
 
-fn push_enabled(state: &ServerState) -> bool {
+pub fn push_enabled(state: &ServerState) -> bool {
     state.config.config_delivery_mode == ConfigDeliveryMode::Push
 }
 
@@ -583,9 +583,30 @@ async fn run_build(state: &Arc<ServerState>, ticket: &BuildTicket) -> BuildResul
         OwnerCheck::Current => {
             let disposition = state
                 .supervisor_config_router()
-                .deliver(&key.sandbox_id, session_id, message, require_acknowledgement)
+                .deliver(
+                    &key.sandbox_id,
+                    session_id,
+                    message,
+                    require_acknowledgement,
+                )
                 .await;
             record_delivery(key.component, disposition.metric_label());
+            // A session that keeps polling never reports a result for the
+            // pending operations bound to this snapshot.
+            if require_acknowledgement
+                && !state
+                    .supervisor_sessions
+                    .session_applies_config(&key.sandbox_id, session_id)
+                && let Err(error) =
+                    crate::config_update_operation::finish_pending_untracked(state, &key.sandbox_id)
+                        .await
+            {
+                warn!(
+                    sandbox_id = %key.sandbox_id,
+                    error = %error,
+                    "failed to finish configuration operations for a polling supervisor"
+                );
+            }
             false
         }
         OwnerCheck::Remote => {

crates/openshell-server/src/config_update_operation.rs

@@ -472,6 +472,38 @@ async fn finish(
     Ok(())
 }
 
+/// Completion of streamed configuration is not tracked here: the gateway keeps
+/// configuration polling, or the sandbox's supervisor does not apply streamed
+/// snapshots.
+const UNTRACKED_COMPLETION: &str = "the sandbox supervisor polls for configuration";
+
+/// Finish an operation whose completion the gateway cannot observe.
+pub async fn finish_untracked(state: &ServerState, operation_id: &str) -> Result<(), Status> {
+    finish(
+        state,
+        operation_id,
+        ConfigUpdateOperationState::Inactive,
+        ConfigApplyOutcome::Unspecified,
+        UNTRACKED_COMPLETION,
+    )
+    .await
+}
+
+/// Finish every pending operation for a sandbox whose supervisor polls.
+pub async fn finish_pending_untracked(state: &ServerState, sandbox_id: &str) -> Result<(), Status> {
+    let records = state
+        .store
+        .list_pending_config_operations_for_scope(sandbox_id)
+        .await
+        .map_err(|error| Status::internal(format!("list pending operations failed: {error}")))?;
+    for record in records {
+        if let Some(operation) = record.operation.as_ref() {
+            finish_untracked(state, &operation.operation_id).await?;
+        }
+    }
+    Ok(())
+}
+
 async fn finish_if_target_matches(
     state: &ServerState,
     operation_id: &str,

crates/openshell-server/src/grpc/mod.rs

@@ -1020,6 +1020,14 @@ pub mod test_support {
         }
     }
 
+    /// Switch a test state to push delivery, which tracks durable completion.
+    pub fn enable_push_delivery(state: &mut Arc<ServerState>) {
+        Arc::get_mut(state)
+            .expect("test state is not shared yet")
+            .config
+            .config_delivery_mode = openshell_core::config::ConfigDeliveryMode::Push;
+    }
+
     /// Configuration stream features advertised by a test supervisor hello.
     #[derive(Clone, Copy, Debug, PartialEq, Eq)]
     pub enum StreamFeatures {

crates/openshell-server/src/grpc/mutation_replay/ordinary/tests.rs

@@ -1163,7 +1163,8 @@ async fn config_completion_timeout_preserves_replayable_committed_operation() {
         ConfigUpdateConsistency, ConfigUpdateOperationState, Sandbox, SandboxPhase, SandboxSpec,
         SandboxStatus, SettingValue, UpdateConfigRequest, setting_value,
     };
-    let (_directory, state) = protected_state().await;
+    let (_directory, mut state) = protected_state().await;
+    crate::grpc::test_support::enable_push_delivery(&mut state);
     state
         .store
         .put_message(&Sandbox {

crates/openshell-server/src/grpc/policy.rs

@@ -153,6 +153,33 @@ async fn finish_config_update_operation(
     consistency: ConfigUpdateConsistency,
     timeout: std::time::Duration,
 ) -> Result<Response<UpdateConfigResponse>, Status> {
+    // Boxed so the many call sites in the update handler stay small.
+    Box::pin(finish_config_update_operation_inner(
+        state,
+        operation_id,
+        consistency,
+        timeout,
+    ))
+    .await
+}
+
+async fn finish_config_update_operation_inner(
+    state: &Arc<ServerState>,
+    operation_id: &str,
+    consistency: ConfigUpdateConsistency,
+    timeout: std::time::Duration,
+) -> Result<Response<UpdateConfigResponse>, Status> {
+    if !crate::config_delivery::push_enabled(state) {
+        // Polling supervisors report no apply results, so completion is not
+        // tracked. Respond as a commit-only update without an operation.
+        config_update_operation::finish_untracked(state, operation_id).await?;
+        let record = config_update_operation::get_record(state, operation_id)
+            .await?
+            .ok_or_else(|| Status::internal("committed update operation disappeared"))?;
+        let mut response = config_update_operation::response_from_record(&record)?;
+        response.operation = None;
+        return Ok(Response::new(response));
+    }
     config_update_operation::reconcile_one(state, operation_id).await?;
     if consistency == ConfigUpdateConsistency::WaitForCompletion {
         config_update_operation::wait_for_terminal(state, operation_id, timeout).await?;
@@ -13270,6 +13297,108 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn poll_mode_waited_update_returns_without_completion_operation() {
+        let state = test_server_state().await;
+        let sandbox = test_sandbox(
+            "sb-poll-wait",
+            "poll-wait",
+            ProtoSandboxPolicy::default(),
+            Vec::new(),
+        );
+        state.store.put_message(&sandbox).await.unwrap();
+
+        // A polling supervisor reports no apply result, so the gateway answers
+        // a waited update as committed instead of waiting for completion.
+        let response = tokio::time::timeout(
+            std::time::Duration::from_secs(2),
+            handle_update_config(
+                &state,
+                with_user(Request::new(UpdateConfigRequest {
+                    sandbox: sandbox.object_name().to_string(),
+                    setting_key: "ocsf_json_enabled".to_string(),
+                    setting_value: Some(SettingValue {
+                        value: Some(setting_value::Value::BoolValue(true)),
+                    }),
+                    workspace_scope: Some(openshell_core::proto::workspace_selector("default")),
+                    consistency: ConfigUpdateConsistency::WaitForCompletion.into(),
+                    idempotency_key: "poll-wait".to_string(),
+                    ..Default::default()
+                })),
+            ),
+        )
+        .await
+        .expect("poll mode must not wait for completion")
+        .unwrap()
+        .into_inner();
+        assert!(response.operation.is_none());
+        assert!(
+            state
+                .store
+                .list_pending_config_operations_for_scope(sandbox.object_id())
+                .await
+                .unwrap()
+                .is_empty(),
+            "poll mode leaves no pending operation behind"
+        );
+    }
+
+    #[tokio::test]
+    async fn push_mode_operation_finishes_for_a_polling_supervisor() {
+        let mut state = test_server_state().await;
+        crate::grpc::test_support::enable_push_delivery(&mut state);
+        let sandbox = test_sandbox(
+            "sb-shadow-operation",
+            "shadow-operation",
+            ProtoSandboxPolicy::default(),
+            Vec::new(),
+        );
+        state.store.put_message(&sandbox).await.unwrap();
+        // A snapshot-only session keeps polling and never reports results.
+        let _session =
+            crate::config_delivery::register_test_push_session(&state, &sandbox, "session-1");
+        crate::supervisor_owner::SupervisorOwnerIndex::new(
+            Arc::clone(&state.store),
+            crate::supervisor_owner::OWNER_TTL,
+        )
+        .publish(
+            "sb-shadow-operation",
+            "session-1",
+            "test-supervisor",
+            1,
+            &state.replica_id,
+            "local://test",
+        )
+        .await
+        .unwrap();
+
+        let response = handle_update_config(
+            &state,
+            with_user(Request::new(UpdateConfigRequest {
+                sandbox: sandbox.object_name().to_string(),
+                setting_key: "ocsf_json_enabled".to_string(),
+                setting_value: Some(SettingValue {
+                    value: Some(setting_value::Value::BoolValue(true)),
+                }),
+                workspace_scope: Some(openshell_core::proto::workspace_selector("default")),
+                consistency: ConfigUpdateConsistency::WaitForCompletion.into(),
+                wait_timeout: openshell_core::time::duration_from_std(
+                    std::time::Duration::from_secs(5),
+                )
+                .ok(),
+                ..Default::default()
+            })),
+        )
+        .await
+        .unwrap()
+        .into_inner();
+        let operation = response.operation.expect("push mode tracks the operation");
+        assert_eq!(
+            openshell_core::proto::ConfigUpdateOperationState::try_from(operation.state).unwrap(),
+            openshell_core::proto::ConfigUpdateOperationState::Inactive
+        );
+    }
+
     #[tokio::test]
     async fn committed_policy_update_publishes_complete_snapshot() {
         let mut state = test_server_state().await;
@@ -23099,7 +23228,8 @@ mod tests {
 
     #[tokio::test]
     async fn update_config_same_policy_hash_with_new_provenance_creates_revision() {
-        let state = test_server_state().await;
+        let mut state = test_server_state().await;
+        crate::grpc::test_support::enable_push_delivery(&mut state);
         let policy = test_policy_with_rule("sandbox_only", "sandbox.example.com");
         let hash = deterministic_policy_hash(&policy);
         let sandbox = test_sandbox("sb-same-hash", "same-hash", policy.clone(), Vec::new());
@@ -24599,7 +24729,8 @@ mod tests {
 
     #[tokio::test]
     async fn stopped_sandbox_setting_update_with_default_version_commits_with_annotations() {
-        let state = test_server_state().await;
+        let mut state = test_server_state().await;
+        crate::grpc::test_support::enable_push_delivery(&mut state);
         let mut sandbox = test_sandbox(
             "sb-inactive-operation",
             "inactive-operation",
@@ -24704,7 +24835,8 @@ mod tests {
 
     #[tokio::test]
     async fn settings_operation_without_policy_history_applies_and_wakes_local_waiter() {
-        let state = test_server_state().await;
+        let mut state = test_server_state().await;
+        crate::grpc::test_support::enable_push_delivery(&mut state);
         let sandbox = test_sandbox(
             "sb-operation-wake",
             "operation-wake",
@@ -24784,7 +24916,8 @@ mod tests {
 
     #[tokio::test]
     async fn operation_waiter_recovers_when_notification_is_missed() {
-        let state = test_server_state().await;
+        let mut state = test_server_state().await;
+        crate::grpc::test_support::enable_push_delivery(&mut state);
         let sandbox = test_sandbox(
             "sb-operation-poll",
             "operation-poll",

crates/openshell-server/src/storage_proto.rs

@@ -230,7 +230,7 @@ mod tests {
     // SandboxProvisioning in both closures. Old rows decode false and empty;
     // decoding or deadline updates cannot claim an existing attempt.
     const PUBLIC_RPC_SCHEMA_SHA256: &str =
-        "f769d9effbb594fed3110470a580c623b3211b4b027ff64fdb3a64260d6d80f6";
+        "ae08da01c8ce68c29a5b298555f99721de74a0325134ce01b1be8b6ef5f5516f";
     const DURABLE_SCHEMA_SHA256: &str =
         "8fa5153f96896035f9ffc0b9ef753a74028c81738383fa8b7ab084bd7e978f97";
     const PUBLIC_DURABLE_OVERLAP_SHA256: &str =
@@ -715,7 +715,7 @@ mod tests {
                 overlap_hash.as_str(),
             ),
             (
-                (327, 28),
+                (329, 29),
                 (94, 21),
                 (81, 21),
                 PUBLIC_RPC_SCHEMA_SHA256,

crates/openshell-server/src/supervisor_session.rs

@@ -1141,6 +1141,18 @@ impl SupervisorSessionRegistry {
         true
     }
 
+    /// True when this exact session applies and acknowledges streamed
+    /// configuration.
+    pub fn session_applies_config(&self, sandbox_id: &str, session_id: &str) -> bool {
+        self.sessions
+            .lock()
+            .unwrap()
+            .get(sandbox_id)
+            .is_some_and(|session| {
+                session.session_id == session_id && session.config_sequences.acknowledged
+            })
+    }
+
     pub fn is_current_session(&self, sandbox_id: &str, session_id: &str) -> bool {
         self.sessions
             .lock()
@@ -6816,9 +6828,12 @@ mod tests {
         };
         let message = SupervisorConfigMessage::SandboxConfig(Box::new(snapshot.clone()));
         assert_eq!(
-            state
-                .supervisor_sessions
-                .deliver_config(sandbox_id, "session-1", message.clone(), false),
+            state.supervisor_sessions.deliver_config(
+                sandbox_id,
+                "session-1",
+                message.clone(),
+                false
+            ),
             DeliveryDisposition::Queued
         );
         let Some(gateway_message::Payload::ConfigUpdate(update)) =
@@ -6911,9 +6926,12 @@ mod tests {
         };
         let message = SupervisorConfigMessage::SandboxConfig(Box::new(snapshot.clone()));
         assert_eq!(
-            state
-                .supervisor_sessions
-                .deliver_config(sandbox_id, "session-1", message.clone(), false),
+            state.supervisor_sessions.deliver_config(
+                sandbox_id,
+                "session-1",
+                message.clone(),
+                false
+            ),
             DeliveryDisposition::Queued
         );
         let Some(gateway_message::Payload::ConfigUpdate(update)) =

crates/openshell-supervisor/src/lib.rs

@@ -11614,5 +11614,4 @@ network_policies:
         assert_eq!(duplicate, result);
         assert_eq!(ctx.opa_engine.current_generation(), rejected_generation);
     }
-
 }

sdk/typescript/src/client.test.ts

@@ -28,9 +28,9 @@ import {
   ConfigUpdateConsistency,
   ConfigUpdateOperationState,
   OpenShell,
+  ServiceAuthorizationMode as ProtoServiceAuthorizationMode,
   SandboxPhase,
   SandboxRestartPolicy,
-  ServiceAuthorizationMode as ProtoServiceAuthorizationMode,
   ServiceStatus,
 } from './gen/openshell_pb.js';
 import { PolicySource, SettingScope } from './gen/sandbox_pb.js';