fix(server): keep an operator's startup policy repair over a stale supervisor write

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

@@ -2661,6 +2661,7 @@ async fn persist_existing_policy_projection(
     annotations: &HashMap<String, String>,
     current_annotations: &HashMap<String, String>,
     backfill_policy: Option<&ProtoSandboxPolicy>,
+    replace_startup_baseline: bool,
 ) -> Result<HashMap<String, String>, Status> {
     let annotations_match = annotations
         .iter()
@@ -2678,7 +2679,7 @@ async fn persist_existing_policy_projection(
                 crate::policy_store::permits_initial_static_policy_repair(sandbox);
             if let Some(policy) = backfill_policy.as_ref()
                 && let Some(spec) = sandbox.spec.as_mut()
-                && (spec.policy.is_none() || startup_blocked)
+                && (spec.policy.is_none() || (startup_blocked && replace_startup_baseline))
             {
                 spec.policy = Some(policy.clone());
             }
@@ -4234,6 +4235,7 @@ async fn handle_update_config_inner(
                 &req.annotations,
                 &response_annotations,
                 None,
+                false,
             )
             .await?
         };
@@ -4382,6 +4384,7 @@ async fn handle_update_config_inner(
                     &req.annotations,
                     &response_annotations,
                     backfill_policy.as_ref(),
+                    startup_blocked && !sandbox_caller,
                 )
                 .await?;
                 if backfill_policy.is_some() {
@@ -4411,6 +4414,7 @@ async fn handle_update_config_inner(
                 expected_resource_version: req.expected_resource_version,
                 annotations: req.annotations.clone(),
                 backfill_policy: backfill_policy.clone(),
+                replace_startup_baseline: startup_blocked && !sandbox_caller,
             };
 
             match state.store.put_policy_revision_atomic(&write).await {
@@ -7333,6 +7337,7 @@ async fn apply_merge_operations_with_retry(
                     expected_resource_version: context.expected_resource_version,
                     annotations: context.annotations.clone(),
                     backfill_policy: None,
+                    replace_startup_baseline: false,
                 })
                 .await
                 .map(Some)

crates/openshell-server/src/persistence/tests.rs

@@ -1379,6 +1379,7 @@ async fn policy_atomic_write_commits_revision_provenance_and_sandbox_projection(
             expected_resource_version: current_version,
             annotations: provenance.clone(),
             backfill_policy: Some(policy.clone()),
+            replace_startup_baseline: false,
         })
         .await
         .unwrap();
@@ -1448,6 +1449,7 @@ async fn policy_atomic_write_rolls_back_sandbox_when_revision_insert_conflicts()
             expected_resource_version: before_version,
             annotations: StdHashMap::from([("signature".to_string(), "new".to_string())]),
             backfill_policy: Some(policy),
+            replace_startup_baseline: false,
         })
         .await
         .unwrap_err();
@@ -1493,6 +1495,7 @@ async fn policy_atomic_write_persists_workspace() {
             expected_resource_version: current_version,
             annotations: StdHashMap::new(),
             backfill_policy: Some(policy),
+            replace_startup_baseline: false,
         })
         .await
         .unwrap();

crates/openshell-server/src/policy_store.rs

@@ -37,6 +37,10 @@ pub struct AtomicPolicyRevisionWrite {
     /// Populate the create-time baseline, or replace it while startup admission
     /// is blocked and no workload has consumed the static restrictions.
     pub backfill_policy: Option<ProtoSandboxPolicy>,
+    /// Allow `backfill_policy` to replace an existing baseline while startup
+    /// admission is blocked. Only a management repair may do this; a
+    /// sandbox-originated write must conflict instead of overwriting it.
+    pub replace_startup_baseline: bool,
 }
 
 pub fn policy_record_for_atomic_write(
@@ -108,7 +112,7 @@ pub fn project_policy_revision_onto_sandbox(
                 changed = true;
             }
             Some(current) if current == backfill_policy => {}
-            Some(_) if startup_blocked => {
+            Some(_) if startup_blocked && write.replace_startup_baseline => {
                 spec.policy = Some(backfill_policy.clone());
                 changed = true;
             }
@@ -710,7 +714,7 @@ mod tests {
             .unwrap()
             .read_write
             .push("/new-static-path".to_string());
-        let write = AtomicPolicyRevisionWrite {
+        let mut write = AtomicPolicyRevisionWrite {
             id: "revision".to_string(),
             sandbox_id: "sandbox".to_string(),
             workspace: "default".to_string(),
@@ -721,6 +725,7 @@ mod tests {
             expected_resource_version: 0,
             annotations: HashMap::new(),
             backfill_policy: Some(replacement.clone()),
+            replace_startup_baseline: true,
         };
         for activated in [None, Some(false), Some(true)] {
             for state in [Admission::Pending, Admission::Rejected, Admission::Accepted] {
@@ -753,5 +758,27 @@ mod tests {
                 }
             }
         }
+        // A sandbox-originated write built from a stale read must not replace an
+        // operator's repair, even while startup admission is still blocked.
+        write.replace_startup_baseline = false;
+        let blocked = Sandbox {
+            spec: Some(SandboxSpec {
+                policy: Some(baseline),
+                ..Default::default()
+            }),
+            status: Some(SandboxStatus {
+                configuration_activated: Some(false),
+                configuration_admission: Some(SandboxConfigurationAdmission {
+                    state: Admission::Pending.into(),
+                    ..Default::default()
+                }),
+                ..Default::default()
+            }),
+            ..Default::default()
+        };
+        assert!(matches!(
+            project_policy_revision_onto_sandbox(&write, &blocked.encode_to_vec(), 1),
+            Err(PersistenceError::Conflict { .. })
+        ));
     }
 }

crates/openshell-supervisor/src/lib.rs

@@ -2062,6 +2062,36 @@ where
     ))
 }
 
+/// Retry transport failures while preserving the snapshot used by the request.
+/// `None` means the gateway returned `ABORTED`: the caller must back off and
+/// fetch fresh configuration instead of resending this snapshot's request.
+async fn grpc_retry_snapshot<T, F, Fut>(op_name: &str, f: F) -> Result<Option<T>>
+where
+    T: Send,
+    F: Fn() -> Fut + Send + Sync,
+    Fut: Future<Output = Result<T>> + Send,
+{
+    grpc_retry(op_name, || async {
+        match f().await {
+            Ok(value) => Ok(Some(value)),
+            Err(error) => {
+                let mut source: Option<&dyn std::error::Error> = Some(error.as_ref());
+                while let Some(cause) = source {
+                    if let Some(status) = cause.downcast_ref::<tonic::Status>() {
+                        if status.code() == tonic::Code::Aborted {
+                            return Ok(None);
+                        }
+                        break;
+                    }
+                    source = cause.source();
+                }
+                Err(error)
+            }
+        }
+    })
+    .await
+}
+
 #[tonic::async_trait]
 trait StartupGateway: Send + Sync {
     async fn snapshot(
@@ -2429,12 +2459,19 @@ async fn load_policy_with_gateway(
                 // Sync and re-fetch over a single connection to avoid extra
                 // TLS handshakes.
                 let ws = snapshot.workspace.clone();
-                let synced = grpc_retry("Image policy synchronization", || {
+                let synced = grpc_retry_snapshot("Image policy synchronization", || {
                     gateway.sync(sandbox, &discovered, &ws)
                 })
                 .await;
                 snapshot = match synced {
-                    Ok(synced) => synced,
+                    Ok(Some(synced)) => synced,
+                    Ok(None) => {
+                        tokio::time::sleep(Duration::from_secs(
+                            1u64 << reconciliation_attempts.min(2),
+                        ))
+                        .await;
+                        continue;
+                    }
                     Err(error) => {
                         // The gateway stored nothing, so report the rejection
                         // against the snapshot this upload was built from and
@@ -2489,12 +2526,19 @@ async fn load_policy_with_gateway(
             let sync_policy = proto_sync_payload_for_enriched_policy(&proto_policy, enriched)
                 .filter(|_| snapshot.policy_source == openshell_core::proto::PolicySource::Sandbox);
             if let Some(sync_policy) = sync_policy {
-                let synced = grpc_retry("Enriched policy synchronization", || {
+                let synced = grpc_retry_snapshot("Enriched policy synchronization", || {
                     gateway.sync(sandbox, &sync_policy, &snapshot.workspace)
                 })
                 .await;
                 let canonical = match synced {
-                    Ok(canonical) => canonical,
+                    Ok(Some(canonical)) => canonical,
+                    Ok(None) => {
+                        tokio::time::sleep(Duration::from_secs(
+                            1u64 << reconciliation_attempts.min(2),
+                        ))
+                        .await;
+                        continue;
+                    }
                     Err(error) => {
                         // The stored policy is unchanged, so report the
                         // rejection against the snapshot it was read from and
@@ -6577,6 +6621,176 @@ network_policies:
         assert_eq!(bundle.2, Some(repaired));
     }
 
+    async fn assert_startup_write_aborted_refetches(image_upload: bool) {
+        use openshell_core::proto::{ConfigurationAdmissionState, PolicySource};
+        // Inject ABORTED while the fixture changes the desired policy. This
+        // checks the supervisor's response to a reported conflict; it does not
+        // show that the gateway rejects an unfenced stale policy write.
+        let operator_policy = proto_policy_fixture();
+        let (calls, mut observed) = tokio::sync::mpsc::unbounded_channel();
+        let gateway = WriteRefusingStartupGateway {
+            desired: Arc::new(std::sync::Mutex::new(if image_upload {
+                unset_policy_snapshot()
+            } else {
+                settings_poll_result(Some(proto_tcp_policy_fixture()), 1, PolicySource::Sandbox)
+            })),
+            refusals: Arc::new(std::sync::Mutex::new(std::collections::VecDeque::from([
+                SyncRefusal::new(
+                    tonic::Code::Aborted,
+                    "persist policy revision failed due to concurrent modification",
+                ),
+            ]))),
+            repair_after_refusal: Some(settings_poll_result(
+                Some(operator_policy.clone()),
+                2,
+                PolicySource::Sandbox,
+            )),
+            rejection_report_errors: Arc::default(),
+            calls,
+        };
+        let started = tokio::time::Instant::now();
+        let policy = load_startup_test_policy(&gateway)
+            .await
+            .expect("startup accepts the operator's policy");
+        let mut trace = Vec::new();
+        while let Ok(call) = observed.try_recv() {
+            trace.push(call);
+        }
+        let refused = trace
+            .iter()
+            .position(|call| *call == StartupCall::Sync)
+            .expect("startup attempts the policy write");
+        assert_eq!(
+            trace.get(refused + 1),
+            Some(&StartupCall::Snapshot),
+            "after ABORTED, startup must fetch a fresh snapshot instead of resending the stale payload; calls: {trace:?}"
+        );
+        assert_eq!(
+            trace
+                .iter()
+                .filter(|call| **call == StartupCall::Sync)
+                .count(),
+            1,
+            "startup must not resend the payload built from the stale read; calls: {trace:?}"
+        );
+        assert!(
+            !trace.iter().any(|call| matches!(
+                call,
+                StartupCall::Report(report)
+                    if report.state == ConfigurationAdmissionState::Rejected
+            )),
+            "a stale write is not a configuration error; calls: {trace:?}"
+        );
+        assert_eq!(
+            policy, operator_policy,
+            "startup must install the operator's policy, not the stale write-back"
+        );
+        assert!(started.elapsed() >= Duration::from_secs(1));
+        assert_eq!(gateway.desired.lock().unwrap().version, 2);
+        let accepted = trace.last().expect("startup reports acceptance");
+        assert!(matches!(accepted, StartupCall::Report(report)
+            if report.state == ConfigurationAdmissionState::Accepted
+                && report.generation == Some((2, "hash-v2".to_string(), 200, 0))));
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn startup_conflict_writes_refetch_instead_of_resending() {
+        for image_upload in [true, false] {
+            Box::pin(assert_startup_write_aborted_refetches(image_upload)).await;
+        }
+    }
+
+    /// Load through the production startup loop, retaining a timeout so a
+    /// mistaken conflict retry cannot leave a test waiting for repair forever.
+    async fn load_startup_test_policy(
+        gateway: &impl StartupGateway,
+    ) -> Result<openshell_core::proto::SandboxPolicy> {
+        let bundle = timeout(
+            Duration::from_mins(1),
+            load_policy_with_gateway(
+                Some("sandbox-id".to_string()),
+                Some("sandbox".to_string()),
+                Some("http://unused.invalid".to_string()),
+                None,
+                None,
+                &openshell_extension_core::ExtensionCredentialStore::new(),
+                LocalPolicyIdentity::Required,
+                None,
+                Some(ImagePolicyDiscovery::Policy(Box::new(
+                    proto_tcp_policy_fixture(),
+                ))),
+                gateway,
+            ),
+        )
+        .await
+        .expect("startup must finish within its retry budget")?;
+        Ok(bundle
+            .2
+            .expect("gateway startup returns its accepted policy"))
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn startup_conflict_write_transport_errors_are_retried() {
+        use openshell_core::proto::PolicySource;
+        for image_upload in [true, false] {
+            let (calls, mut observed) = tokio::sync::mpsc::unbounded_channel();
+            let gateway = WriteRefusingStartupGateway {
+                desired: Arc::new(std::sync::Mutex::new(if image_upload {
+                    unset_policy_snapshot()
+                } else {
+                    settings_poll_result(Some(proto_tcp_policy_fixture()), 1, PolicySource::Sandbox)
+                })),
+                refusals: Arc::new(std::sync::Mutex::new(std::collections::VecDeque::from([
+                    SyncRefusal::new(tonic::Code::Unavailable, "transport failed"),
+                ]))),
+                repair_after_refusal: None,
+                rejection_report_errors: Arc::default(),
+                calls,
+            };
+            load_startup_test_policy(&gateway)
+                .await
+                .expect("startup retries a transport failure");
+            let syncs = std::iter::from_fn(|| observed.try_recv().ok())
+                .filter(|call| *call == StartupCall::Sync)
+                .count();
+            assert_eq!(syncs, 2, "the same write is retried once");
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn startup_conflict_writes_keep_reconciliation_budget() {
+        use openshell_core::proto::PolicySource;
+        for image_upload in [true, false] {
+            let (calls, mut observed) = tokio::sync::mpsc::unbounded_channel();
+            let gateway = WriteRefusingStartupGateway {
+                desired: Arc::new(std::sync::Mutex::new(if image_upload {
+                    unset_policy_snapshot()
+                } else {
+                    settings_poll_result(Some(proto_tcp_policy_fixture()), 1, PolicySource::Sandbox)
+                })),
+                refusals: Arc::new(std::sync::Mutex::new(std::collections::VecDeque::from(
+                    vec![SyncRefusal::new(tonic::Code::Aborted, "configuration keeps changing"); 5],
+                ))),
+                repair_after_refusal: None,
+                rejection_report_errors: Arc::default(),
+                calls,
+            };
+            let error = load_startup_test_policy(&gateway)
+                .await
+                .expect_err("conflicts must have a bounded retry budget");
+            assert!(
+                error
+                    .to_string()
+                    .contains("did not stabilize after 5 attempts"),
+                "{error}"
+            );
+            let syncs = std::iter::from_fn(|| observed.try_recv().ok())
+                .filter(|call| *call == StartupCall::Sync)
+                .count();
+            assert_eq!(syncs, 5, "each conflict spends one attempt, no resends");
+        }
+    }
+
     #[tokio::test(start_paused = true)]
     async fn startup_policy_write_refusal_image_repair_before_report_refetches() {
         Box::pin(assert_startup_repair_before_rejection_refetches(