From 50cda8effe833c1dc6fd97caa9aeeb2a8339df5e Mon Sep 17 00:00:00 2001
From: Kyle Zheng <kyzheng@nvidia.com>
Date: Tue, 7 Jul 2026 12:06:36 -0700
Subject: [PATCH 01/10] fix(sandbox): acknowledge initial policy revision
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The supervisor loaded and enforced a sandbox-scoped policy but never told
the gateway which revision it loaded. The policy poll loop seeded itself
with the initial revision's hash on its first poll, so `policy_changed`
was never true for that revision and `ReportPolicyStatus(LOADED)` — which
only ran in the hot-reload branch — was never called. The revision stayed
`Pending` and `current_policy_version` stayed 0 even though the sandbox was
`Ready` and the policy was effective. This was most visible with sparse
policies that get baseline-enriched into a new revision during startup.

After the OPA engine is constructed, report the exact sandbox revision the
supervisor loaded as LOADED, and seed the poll loop from that revision so
it is not re-reported. Report FAILED with the original construction error
if engine construction or conversion fails. Only sandbox-sourced revisions
(version > 0) whose canonical content matches the loaded policy are
acknowledged; global and local-file policies are untouched. Delivery uses
the shared bounded retry, is non-fatal on transient failure, and a pending
initial acknowledgement is delivered before any newer revision so policy
history is never reordered.

Signed-off-by: Kyle Zheng <kyzheng@nvidia.com>
---
 architecture/sandbox.md             |  21 ++
 crates/openshell-sandbox/src/lib.rs | 331 +++++++++++++++++++++++++++-
 2 files changed, 349 insertions(+), 3 deletions(-)

diff --git a/architecture/sandbox.md b/architecture/sandbox.md
index 580d8f96d..0de0b7d2a 100644
--- a/architecture/sandbox.md
+++ b/architecture/sandbox.md
@@ -171,6 +171,27 @@ the structured 403 and authors the narrowest rule. Mechanistically mapping L7
 would either over-broaden rules or require path-templating logic that rots
 quickly.
 
+## Policy Revision Acknowledgement
+
+When the supervisor loads a sandbox-scoped policy from the gateway, it
+acknowledges the exact revision it constructed. After the OPA engine is built
+successfully, the supervisor reports that revision as `LOADED`, which advances
+`SandboxStatus.current_policy_version` and moves the revision out of `Pending`.
+If policy construction fails, it reports the same revision as `FAILED` with the
+original construction error.
+
+This holds even when the initial policy is enriched with baseline paths during
+startup: the enriched revision the supervisor synced back to the gateway is the
+revision it acknowledges, so a successfully constructed initial policy never
+remains `Pending`. The acknowledgement is delivered with bounded retries and is
+non-fatal: a transient delivery failure does not fail an otherwise healthy
+sandbox, and the pending acknowledgement is retried before any newer revision is
+processed so policy history is never reordered.
+
+Only sandbox-scoped revisions (`PolicySource::Sandbox`, version greater than
+zero) are acknowledged. Global policies and local-file development policies do
+not use the sandbox revision API and produce no acknowledgement.
+
 ## Failure Behavior
 
 - If gateway config polling fails, the sandbox keeps its last-known-good policy.
diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs
index 53b1eba58..56deb0f3b 100644
--- a/crates/openshell-sandbox/src/lib.rs
+++ b/crates/openshell-sandbox/src/lib.rs
@@ -418,6 +418,7 @@ pub async fn run_sandbox(
             endpoint: poll_endpoint,
             sandbox_id: poll_id,
             opa_engine: poll_engine,
+            loaded_policy: retained_proto.clone(),
             entrypoint_pid: poll_pid,
             interval_secs: poll_interval_secs,
             ocsf_enabled: poll_ocsf_enabled,
@@ -1476,9 +1477,21 @@ async fn load_policy(
         // 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 opa_engine = Some(Arc::new(OpaEngine::from_proto(&proto_policy)?));
+        let opa_engine = match OpaEngine::from_proto(&proto_policy) {
+            Ok(engine) => Some(Arc::new(engine)),
+            Err(e) => {
+                report_initial_policy_failure(endpoint, id, &proto_policy, &e).await;
+                return Err(e);
+            }
+        };
 
-        let policy = SandboxPolicy::try_from(proto_policy.clone())?;
+        let policy = match SandboxPolicy::try_from(proto_policy.clone()) {
+            Ok(policy) => policy,
+            Err(e) => {
+                report_initial_policy_failure(endpoint, id, &proto_policy, &e).await;
+                return Err(e);
+            }
+        };
         return Ok((policy, opa_engine, Some(proto_policy)));
     }
 
@@ -1598,6 +1611,138 @@ fn discover_policy_from_path(path: &std::path::Path) -> openshell_core::proto::S
     }
 }
 
+/// A sandbox-scoped policy revision that was constructed successfully at
+/// startup and must be acknowledged to the gateway exactly once.
+///
+/// The supervisor loads the initial policy, builds the OPA engine, and then
+/// reports the exact revision as loaded. Without this, an enriched initial
+/// revision stays `Pending` forever because the poll loop seeds itself with
+/// that revision's hash and never treats it as a change.
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct InitialPolicyAck {
+    version: u32,
+    policy_hash: String,
+    config_revision: u64,
+}
+
+/// Determine whether the initially loaded policy corresponds to an
+/// authoritative sandbox-scoped revision that must be acknowledged.
+///
+/// Returns `Some` only for sandbox-sourced revisions (version > 0) whose
+/// canonical gateway content structurally matches the policy the supervisor
+/// loaded into the OPA engine. Global policies, local-file development
+/// policies, version zero, and content that cannot be matched yield `None`,
+/// so those paths never emit a sandbox-revision acknowledgement.
+fn initial_policy_ack_candidate(
+    loaded: Option<&openshell_core::proto::SandboxPolicy>,
+    canonical: &openshell_core::grpc_client::SettingsPollResult,
+) -> Option<InitialPolicyAck> {
+    let loaded = loaded?;
+    if canonical.policy_source != openshell_core::proto::PolicySource::Sandbox {
+        return None;
+    }
+    if canonical.version == 0 {
+        return None;
+    }
+    let canonical_policy = canonical.policy.as_ref()?;
+    if !policies_structurally_match(loaded, canonical_policy) {
+        return None;
+    }
+    Some(InitialPolicyAck {
+        version: canonical.version,
+        policy_hash: canonical.policy_hash.clone(),
+        config_revision: canonical.config_revision,
+    })
+}
+
+/// Compare two sandbox policies for enforcement-equivalent content.
+///
+/// The policy `version` field and provider-managed rule names are ignored:
+/// the gateway strips provider rule names on store and re-composes them on
+/// read, so comparing them directly would spuriously reject a revision the
+/// supervisor actually loaded.
+fn policies_structurally_match(
+    a: &openshell_core::proto::SandboxPolicy,
+    b: &openshell_core::proto::SandboxPolicy,
+) -> bool {
+    let mut a = a.clone();
+    let mut b = b.clone();
+    a.version = 0;
+    b.version = 0;
+    strip_proto_provider_policy_entries(&mut a);
+    strip_proto_provider_policy_entries(&mut b);
+    a == b
+}
+
+/// Report a successfully constructed initial policy revision as loaded,
+/// using the shared bounded gRPC retry. Returns whether delivery succeeded;
+/// a transient failure is non-fatal and retried later by the poll loop.
+async fn report_initial_policy_ack(
+    client: &openshell_core::grpc_client::CachedOpenShellClient,
+    sandbox_id: &str,
+    ack: &InitialPolicyAck,
+) -> bool {
+    grpc_retry("Initial policy acknowledgement", || {
+        let client = client.clone();
+        async move {
+            client
+                .report_policy_status(sandbox_id, ack.version, true, "")
+                .await
+        }
+    })
+    .await
+    .inspect_err(|e| {
+        warn!(error = %e, version = ack.version, "Failed to acknowledge initial policy revision");
+    })
+    .is_ok()
+}
+
+/// Best-effort `FAILED` acknowledgement when initial policy construction or
+/// conversion fails.
+///
+/// Re-fetches the authoritative sandbox revision so the exact version is
+/// reported, and preserves the original construction error as the reported
+/// message. A delivery failure here is swallowed so it can never mask the
+/// original construction error returned to the caller.
+async fn report_initial_policy_failure(
+    endpoint: &str,
+    sandbox_id: &str,
+    loaded: &openshell_core::proto::SandboxPolicy,
+    error: &miette::Report,
+) {
+    let client = match openshell_core::grpc_client::CachedOpenShellClient::connect(endpoint).await {
+        Ok(client) => client,
+        Err(e) => {
+            warn!(error = %e, "Failed to connect to report initial policy failure");
+            return;
+        }
+    };
+    let canonical = match client.poll_settings(sandbox_id).await {
+        Ok(result) => result,
+        Err(e) => {
+            warn!(error = %e, "Failed to fetch config to report initial policy failure");
+            return;
+        }
+    };
+    let Some(candidate) = initial_policy_ack_candidate(Some(loaded), &canonical) else {
+        return;
+    };
+    let message = error.to_string();
+    if let Err(e) = grpc_retry("Initial policy failure report", || {
+        let client = client.clone();
+        let message = message.clone();
+        async move {
+            client
+                .report_policy_status(sandbox_id, candidate.version, false, &message)
+                .await
+        }
+    })
+    .await
+    {
+        warn!(error = %e, version = candidate.version, "Failed to report initial policy failure");
+    }
+}
+
 /// Background loop that polls the server for policy updates.
 ///
 /// When a new version is detected, attempts to reload the OPA engine via
@@ -1610,6 +1755,10 @@ struct PolicyPollLoopContext {
     endpoint: String,
     sandbox_id: String,
     opa_engine: Arc<OpaEngine>,
+    /// The policy the supervisor loaded into the OPA engine, used to identify
+    /// and acknowledge the initial sandbox-scoped revision. `None` for
+    /// local-file mode, where no gateway acknowledgement applies.
+    loaded_policy: Option<openshell_core::proto::SandboxPolicy>,
     entrypoint_pid: Arc<AtomicU32>,
     interval_secs: u64,
     ocsf_enabled: Arc<std::sync::atomic::AtomicBool>,
@@ -1631,12 +1780,41 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
         openshell_core::proto::EffectiveSetting,
     > = std::collections::HashMap::new();
 
-    // Initialize revision from the first poll.
+    // A sandbox-scoped initial revision that was built successfully but whose
+    // acknowledgement has not yet been delivered. Held so a transient
+    // status-report failure never fails a healthy sandbox, and so a newer
+    // revision cannot be acknowledged ahead of the one actually loaded.
+    let mut pending_initial_ack: Option<InitialPolicyAck> = None;
+
+    // Initialize revision from the first poll and acknowledge the initial
+    // policy revision the supervisor actually loaded. Seeding the loop from
+    // that revision is what prevents it from being re-reported on later polls.
     match client.poll_settings(&ctx.sandbox_id).await {
         Ok(result) => {
             apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings);
             current_config_revision = result.config_revision;
             current_policy_hash = result.policy_hash.clone();
+
+            if let Some(candidate) =
+                initial_policy_ack_candidate(ctx.loaded_policy.as_ref(), &result)
+            {
+                if report_initial_policy_ack(&client, &ctx.sandbox_id, &candidate).await {
+                    ocsf_emit!(ConfigStateChangeBuilder::new(ocsf_ctx())
+                        .severity(SeverityId::Informational)
+                        .status(StatusId::Success)
+                        .state(StateId::Enabled, "loaded")
+                        .unmapped("version", serde_json::json!(candidate.version))
+                        .unmapped("policy_hash", serde_json::json!(&candidate.policy_hash))
+                        .message(format!(
+                            "Acknowledged initial policy revision as loaded [version:{}]",
+                            candidate.version
+                        ))
+                        .build());
+                } else {
+                    pending_initial_ack = Some(candidate);
+                }
+            }
+
             current_settings = result.settings;
             debug!(
                 config_revision = current_config_revision,
@@ -1652,6 +1830,17 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
     loop {
         tokio::time::sleep(interval).await;
 
+        // Deliver any pending initial acknowledgement before processing newer
+        // revisions, so the revision actually loaded is acknowledged first and
+        // policy history is never reordered.
+        if let Some(candidate) = pending_initial_ack.clone() {
+            if report_initial_policy_ack(&client, &ctx.sandbox_id, &candidate).await {
+                pending_initial_ack = None;
+            } else {
+                continue;
+            }
+        }
+
         let result = match client.poll_settings(&ctx.sandbox_id).await {
             Ok(r) => r,
             Err(e) => {
@@ -2087,4 +2276,140 @@ filesystem_policy:
         let local_policy = SandboxPolicy::try_from(proto).expect("conversion should succeed");
         assert!(matches!(local_policy.network.mode, NetworkMode::Proxy));
     }
+
+    // ---- Initial policy acknowledgement tests ----
+
+    fn proto_policy_fixture() -> openshell_core::proto::SandboxPolicy {
+        openshell_policy::restrictive_default_policy()
+    }
+
+    fn settings_poll_result(
+        policy: Option<openshell_core::proto::SandboxPolicy>,
+        version: u32,
+        source: openshell_core::proto::PolicySource,
+    ) -> openshell_core::grpc_client::SettingsPollResult {
+        openshell_core::grpc_client::SettingsPollResult {
+            policy,
+            version,
+            policy_hash: format!("hash-v{version}"),
+            config_revision: u64::from(version) * 100,
+            policy_source: source,
+            settings: std::collections::HashMap::new(),
+            global_policy_version: 0,
+            provider_env_revision: 0,
+        }
+    }
+
+    #[test]
+    fn initial_ack_candidate_matches_sandbox_revision() {
+        let loaded = proto_policy_fixture();
+        let canonical = settings_poll_result(
+            Some(loaded.clone()),
+            2,
+            openshell_core::proto::PolicySource::Sandbox,
+        );
+
+        let ack = initial_policy_ack_candidate(Some(&loaded), &canonical)
+            .expect("sandbox-sourced matching revision should be acknowledged");
+
+        assert_eq!(ack.version, 2);
+        assert_eq!(ack.policy_hash, "hash-v2");
+        assert_eq!(ack.config_revision, 200);
+    }
+
+    #[test]
+    fn initial_ack_candidate_ignores_global_policy() {
+        let loaded = proto_policy_fixture();
+        let canonical = settings_poll_result(
+            Some(loaded.clone()),
+            1,
+            openshell_core::proto::PolicySource::Global,
+        );
+
+        assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
+    }
+
+    #[test]
+    fn initial_ack_candidate_ignores_version_zero() {
+        let loaded = proto_policy_fixture();
+        let canonical = settings_poll_result(
+            Some(loaded.clone()),
+            0,
+            openshell_core::proto::PolicySource::Sandbox,
+        );
+
+        assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
+    }
+
+    #[test]
+    fn initial_ack_candidate_ignores_local_file_mode() {
+        // Local-file mode retains no proto policy, so there is nothing to
+        // acknowledge to the gateway.
+        let canonical = settings_poll_result(
+            Some(proto_policy_fixture()),
+            1,
+            openshell_core::proto::PolicySource::Sandbox,
+        );
+
+        assert!(initial_policy_ack_candidate(None, &canonical).is_none());
+    }
+
+    #[test]
+    fn initial_ack_candidate_rejects_missing_canonical_policy() {
+        let loaded = proto_policy_fixture();
+        let canonical =
+            settings_poll_result(None, 2, openshell_core::proto::PolicySource::Sandbox);
+
+        assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
+    }
+
+    #[test]
+    fn initial_ack_candidate_rejects_mismatched_content() {
+        let loaded = proto_policy_fixture();
+        let mut different = proto_policy_fixture();
+        different.network_policies.insert(
+            "user_rule".to_string(),
+            openshell_core::proto::NetworkPolicyRule::default(),
+        );
+        let canonical =
+            settings_poll_result(Some(different), 2, openshell_core::proto::PolicySource::Sandbox);
+
+        assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
+    }
+
+    #[test]
+    fn structural_match_ignores_version_field() {
+        let mut a = proto_policy_fixture();
+        let mut b = proto_policy_fixture();
+        a.version = 1;
+        b.version = 2;
+
+        assert!(policies_structurally_match(&a, &b));
+    }
+
+    #[test]
+    fn structural_match_ignores_provider_rule_names() {
+        // The gateway strips provider rule names on store and re-composes them
+        // on read, so a policy differing only by provider rules still matches.
+        let base = proto_policy_fixture();
+        let mut with_provider = base.clone();
+        with_provider.network_policies.insert(
+            "_provider_work_github".to_string(),
+            openshell_core::proto::NetworkPolicyRule::default(),
+        );
+
+        assert!(policies_structurally_match(&base, &with_provider));
+    }
+
+    #[test]
+    fn structural_match_detects_user_rule_differences() {
+        let base = proto_policy_fixture();
+        let mut changed = base.clone();
+        changed.network_policies.insert(
+            "user_rule".to_string(),
+            openshell_core::proto::NetworkPolicyRule::default(),
+        );
+
+        assert!(!policies_structurally_match(&base, &changed));
+    }
 }

From 8173942536a1a37cff91b9cbaffe1991797902ca Mon Sep 17 00:00:00 2001
From: Kyle Zheng <kyzheng@nvidia.com>
Date: Tue, 7 Jul 2026 12:06:46 -0700
Subject: [PATCH 02/10] feat(python): expose sandbox labels and selectors

The gateway protobuf and CLI already support request-level sandbox labels
(`CreateSandboxRequest.name`/`labels`) and selector-based listing
(`ListSandboxesRequest.label_selector`), but the public Python SDK dropped
them, so Python-created sandboxes could not be found via
`openshell sandbox list --selector ...`.

Add optional, source-compatible `name`/`labels` to `SandboxClient.create`,
`create_session`, and the high-level `Sandbox`, and `label_selector` to
`list`/`list_ids`. `SandboxRef` now carries the gateway labels as an
immutable mapping (default empty, so `SandboxRef(id, name, status)` still
works). Caller-provided label mappings are copied. Attaching the high-level
`Sandbox` to an existing sandbox rejects `name`/`labels` since creation
metadata cannot change on attach. Template labels remain a separate concept.
No protobuf changes are required.

Signed-off-by: Kyle Zheng <kyzheng@nvidia.com>
---
 docs/sandboxes/manage-sandboxes.mdx |  15 ++
 e2e/python/test_sandbox_api.py      |  55 +++++++
 python/openshell/sandbox.py         |  65 +++++++-
 python/openshell/sandbox_test.py    | 227 ++++++++++++++++++++++++++++
 4 files changed, 354 insertions(+), 8 deletions(-)

diff --git a/docs/sandboxes/manage-sandboxes.mdx b/docs/sandboxes/manage-sandboxes.mdx
index fdbf497f5..8d47754c5 100644
--- a/docs/sandboxes/manage-sandboxes.mdx
+++ b/docs/sandboxes/manage-sandboxes.mdx
@@ -221,6 +221,21 @@ openshell sandbox list --selector env=dev
 openshell sandbox list --selector env=dev,team=platform
 ```
 
+The Python SDK accepts the same gateway labels and selectors. Labels passed to
+`create` are stored on the gateway sandbox object and are returned on
+`SandboxRef.labels`, so selector-based listing finds Python-created sandboxes:
+
+```python
+from openshell import SandboxClient
+
+with SandboxClient.from_active_cluster() as client:
+    sandbox = client.create(name="deep-research-1", labels={"env": "dev", "team": "platform"})
+    assert sandbox.labels["team"] == "platform"
+
+    matches = client.list(label_selector="env=dev,team=platform")
+    assert sandbox.id in {s.id for s in matches}
+```
+
 ## Expose Long Running Services
 
 Service forwarding makes a long-running process inside a sandbox reachable through a gateway-managed URL. Use it for development servers, notebooks, dashboards, or other services that keep listening after the sandbox starts. Run the service on loopback inside the sandbox, expose its port, then open the URL printed by OpenShell.
diff --git a/e2e/python/test_sandbox_api.py b/e2e/python/test_sandbox_api.py
index 80a7d5d6b..929728d7c 100644
--- a/e2e/python/test_sandbox_api.py
+++ b/e2e/python/test_sandbox_api.py
@@ -58,3 +58,58 @@ def read(self, path: str) -> str:
         )
         assert verify_file.exit_code == 0
         assert verify_file.stdout.strip() == "ok"
+
+
+def test_sandbox_labels_and_selectors(sandbox_client: SandboxClient) -> None:
+    import contextlib
+    import uuid
+
+    suffix = uuid.uuid4().hex[:8]
+    job_a = f"aiq-labels-a-{suffix}"
+    job_b = f"aiq-labels-b-{suffix}"
+    group_selector = f"aiq-test={suffix}"
+    primary_selector = f"aiq-test={suffix},role=primary"
+
+    created: list[str] = []
+    try:
+        ref_a = sandbox_client.create(
+            name=job_a, labels={"aiq-test": suffix, "role": "primary"}
+        )
+        created.append(ref_a.name)
+        ref_b = sandbox_client.create(
+            name=job_b, labels={"aiq-test": suffix, "role": "secondary"}
+        )
+        created.append(ref_b.name)
+
+        # Labels round-trip through create and get.
+        assert ref_a.labels["role"] == "primary"
+        assert dict(sandbox_client.get(job_a).labels)["role"] == "primary"
+        assert dict(sandbox_client.get(job_b).labels)["role"] == "secondary"
+
+        # A specific selector filters to exactly the primary sandbox.
+        assert {
+            s.name for s in sandbox_client.list(label_selector=primary_selector)
+        } == {job_a}
+        # The shared group label returns both.
+        assert {s.name for s in sandbox_client.list(label_selector=group_selector)} == {
+            job_a,
+            job_b,
+        }
+
+        # Deleting one removes only it from selector results.
+        assert sandbox_client.delete(job_a)
+        sandbox_client.wait_deleted(job_a)
+        created.remove(job_a)
+        assert {s.name for s in sandbox_client.list(label_selector=group_selector)} == {
+            job_b
+        }
+
+        # Final deletion leaves no matching sandboxes.
+        assert sandbox_client.delete(job_b)
+        sandbox_client.wait_deleted(job_b)
+        created.remove(job_b)
+        assert not sandbox_client.list(label_selector=group_selector)
+    finally:
+        for name in created:
+            with contextlib.suppress(Exception):
+                sandbox_client.delete(name)
diff --git a/python/openshell/sandbox.py b/python/openshell/sandbox.py
index 3b64089de..07b9675c0 100644
--- a/python/openshell/sandbox.py
+++ b/python/openshell/sandbox.py
@@ -13,7 +13,8 @@
 import threading
 import time
 from collections import namedtuple
-from dataclasses import dataclass
+from dataclasses import dataclass, field
+from types import MappingProxyType
 from typing import TYPE_CHECKING, cast
 from urllib.parse import urlparse
 
@@ -135,6 +136,7 @@ class SandboxRef:
     id: str
     name: str
     status: SandboxStatusRef
+    labels: Mapping[str, str] = field(default_factory=lambda: MappingProxyType({}))
 
     @property
     def phase(self) -> int:
@@ -397,10 +399,16 @@ def create(
         self,
         *,
         spec: openshell_pb2.SandboxSpec | None = None,
+        name: str | None = None,
+        labels: Mapping[str, str] | None = None,
     ) -> SandboxRef:
         request_spec = spec if spec is not None else _default_spec()
         response = self._stub.CreateSandbox(
-            openshell_pb2.CreateSandboxRequest(spec=request_spec),
+            openshell_pb2.CreateSandboxRequest(
+                spec=request_spec,
+                name=name or "",
+                labels=dict(labels) if labels else {},
+            ),
             timeout=self._timeout,
         )
         sandbox_ref = _sandbox_ref(response.sandbox)
@@ -412,8 +420,10 @@ def create_session(
         self,
         *,
         spec: openshell_pb2.SandboxSpec | None = None,
+        name: str | None = None,
+        labels: Mapping[str, str] | None = None,
     ) -> SandboxSession:
-        return SandboxSession(self, self.create(spec=spec))
+        return SandboxSession(self, self.create(spec=spec, name=name, labels=labels))
 
     def get(self, sandbox_name: str) -> SandboxRef:
         response = self._stub.GetSandbox(
@@ -425,15 +435,36 @@ def get(self, sandbox_name: str) -> SandboxRef:
     def get_session(self, sandbox_name: str) -> SandboxSession:
         return SandboxSession(self, self.get(sandbox_name))
 
-    def list(self, *, limit: int = 100, offset: int = 0) -> builtins.list[SandboxRef]:
+    def list(
+        self,
+        *,
+        limit: int = 100,
+        offset: int = 0,
+        label_selector: str | None = None,
+    ) -> builtins.list[SandboxRef]:
         response = self._stub.ListSandboxes(
-            openshell_pb2.ListSandboxesRequest(limit=limit, offset=offset),
+            openshell_pb2.ListSandboxesRequest(
+                limit=limit,
+                offset=offset,
+                label_selector=label_selector or "",
+            ),
             timeout=self._timeout,
         )
         return [_sandbox_ref(item) for item in response.sandboxes]
 
-    def list_ids(self, *, limit: int = 100, offset: int = 0) -> builtins.list[str]:
-        return [item.id for item in self.list(limit=limit, offset=offset)]
+    def list_ids(
+        self,
+        *,
+        limit: int = 100,
+        offset: int = 0,
+        label_selector: str | None = None,
+    ) -> builtins.list[str]:
+        return [
+            item.id
+            for item in self.list(
+                limit=limit, offset=offset, label_selector=label_selector
+            )
+        ]
 
     def delete(self, sandbox_name: str) -> bool:
         response = self._stub.DeleteSandbox(
@@ -646,6 +677,8 @@ def __init__(
         sandbox: str | SandboxRef | None = None,
         delete_on_exit: bool = True,
         spec: openshell_pb2.SandboxSpec | None = None,
+        name: str | None = None,
+        labels: Mapping[str, str] | None = None,
         timeout: float = 30.0,
         ready_timeout_seconds: float = 120.0,
         auto_refresh: bool = True,
@@ -665,6 +698,8 @@ def __init__(
         self._sandbox_input = sandbox
         self._delete_on_exit = delete_on_exit
         self._spec = spec
+        self._name = name
+        self._labels = labels
         self._timeout = timeout
         self._ready_timeout_seconds = ready_timeout_seconds
         self._auto_refresh = auto_refresh
@@ -686,6 +721,15 @@ def sandbox(self) -> SandboxRef:
         return self._session.sandbox
 
     def __enter__(self) -> Sandbox:
+        # Creation metadata cannot be applied when attaching to an existing
+        # sandbox; reject it before opening a connection.
+        if self._sandbox_input is not None and (
+            self._name is not None or self._labels is not None
+        ):
+            raise SandboxError(
+                "name and labels cannot be set when attaching to an existing sandbox"
+            )
+
         client = SandboxClient.from_active_cluster(
             cluster=self._cluster,
             timeout=self._timeout,
@@ -696,7 +740,9 @@ def __enter__(self) -> Sandbox:
         self._client = client
 
         if self._sandbox_input is None:
-            self._session = client.create_session(spec=self._spec)
+            self._session = client.create_session(
+                spec=self._spec, name=self._name, labels=self._labels
+            )
         elif isinstance(self._sandbox_input, SandboxRef):
             self._session = SandboxSession(client, self._sandbox_input)
         else:
@@ -813,6 +859,9 @@ def _sandbox_ref(sandbox: openshell_pb2.Sandbox) -> SandboxRef:
             phase=status.phase if status else 0,
             current_policy_version=status.current_policy_version if status else 0,
         ),
+        labels=MappingProxyType(
+            dict(sandbox.metadata.labels) if sandbox.metadata else {}
+        ),
     )
 
 
diff --git a/python/openshell/sandbox_test.py b/python/openshell/sandbox_test.py
index 82e86eefc..2e9c9b275 100644
--- a/python/openshell/sandbox_test.py
+++ b/python/openshell/sandbox_test.py
@@ -8,8 +8,11 @@
 import threading
 import time
 from pathlib import Path
+from types import SimpleNamespace
 from typing import Any, cast
 
+import pytest
+
 from openshell._proto import openshell_pb2
 from openshell.sandbox import (
     _PYTHON_CLOUDPICKLE_BOOTSTRAP,
@@ -18,6 +21,8 @@
     Sandbox,
     SandboxClient,
     SandboxError,
+    SandboxRef,
+    SandboxStatusRef,
     TlsConfig,
     _BearerAuthInterceptor,
     _load_cluster_bearer_token,
@@ -25,6 +30,7 @@
     _normalize_bearer,
     _OidcRefresher,
     _read_oidc_token_bundle,
+    _sandbox_ref,
 )
 
 
@@ -1413,3 +1419,224 @@ def test_from_active_cluster_reads_utf8_bytes_from_active_gateway_and_metadata(
         assert client._endpoint == "tést.example:8080"
     finally:
         client.close()
+
+
+# ---- Sandbox label / selector API tests ----
+
+
+def _make_sandbox_proto(
+    id_: str,
+    name: str,
+    labels: dict[str, str] | None = None,
+    phase: int = openshell_pb2.SANDBOX_PHASE_READY,
+    version: int = 0,
+) -> openshell_pb2.Sandbox:
+    sandbox = openshell_pb2.Sandbox()
+    sandbox.metadata.id = id_
+    sandbox.metadata.name = name
+    for key, value in (labels or {}).items():
+        sandbox.metadata.labels[key] = value
+    sandbox.status.phase = phase
+    sandbox.status.current_policy_version = version
+    return sandbox
+
+
+class _FakeSandboxStub:
+    def __init__(self, listed: list[openshell_pb2.Sandbox] | None = None) -> None:
+        self.create_request: openshell_pb2.CreateSandboxRequest | None = None
+        self.list_request: openshell_pb2.ListSandboxesRequest | None = None
+        self._listed = listed or []
+
+    def CreateSandbox(
+        self,
+        request: openshell_pb2.CreateSandboxRequest,
+        timeout: float | None = None,
+    ) -> Any:
+        self.create_request = request
+        _ = timeout
+        return SimpleNamespace(
+            sandbox=_make_sandbox_proto(
+                "sandbox-1", request.name or "generated", dict(request.labels)
+            )
+        )
+
+    def ListSandboxes(
+        self,
+        request: openshell_pb2.ListSandboxesRequest,
+        timeout: float | None = None,
+    ) -> Any:
+        self.list_request = request
+        _ = timeout
+        return SimpleNamespace(sandboxes=list(self._listed))
+
+
+class _RecordingHighLevelClient:
+    """A stand-in for SandboxClient used to observe high-level forwarding."""
+
+    def __init__(self) -> None:
+        self.create_kwargs: dict[str, Any] | None = None
+
+    def create_session(
+        self,
+        *,
+        spec: Any = None,
+        name: str | None = None,
+        labels: Any = None,
+    ) -> Any:
+        self.create_kwargs = {"spec": spec, "name": name, "labels": labels}
+        return SimpleNamespace(sandbox=SimpleNamespace(name=name or "generated"))
+
+    def wait_ready(self, name: str, *, timeout_seconds: float = 300.0) -> SandboxRef:
+        _ = timeout_seconds
+        return SandboxRef(
+            id="sandbox-1",
+            name=name,
+            status=SandboxStatusRef(phase=2, current_policy_version=0),
+        )
+
+
+def test_create_forwards_name_and_labels() -> None:
+    stub = _FakeSandboxStub()
+    client = _client_with_fake_stub(stub)
+
+    ref = client.create(name="job-1", labels={"aiq": "deep-research"})
+
+    assert stub.create_request is not None
+    assert stub.create_request.name == "job-1"
+    assert dict(stub.create_request.labels) == {"aiq": "deep-research"}
+    assert dict(ref.labels) == {"aiq": "deep-research"}
+
+
+def test_create_without_args_sends_empty_metadata() -> None:
+    stub = _FakeSandboxStub()
+    client = _client_with_fake_stub(stub)
+
+    client.create()
+
+    assert stub.create_request is not None
+    assert stub.create_request.name == ""
+    assert dict(stub.create_request.labels) == {}
+
+
+def test_create_copies_caller_labels() -> None:
+    stub = _FakeSandboxStub()
+    client = _client_with_fake_stub(stub)
+
+    caller_labels = {"aiq": "deep-research"}
+    client.create(labels=caller_labels)
+    caller_labels["aiq"] = "mutated"
+
+    assert stub.create_request is not None
+    assert dict(stub.create_request.labels) == {"aiq": "deep-research"}
+
+
+def test_create_session_forwards_name_and_labels() -> None:
+    stub = _FakeSandboxStub()
+    client = _client_with_fake_stub(stub)
+
+    session = client.create_session(name="job-2", labels={"team": "aiq"})
+
+    assert stub.create_request is not None
+    assert stub.create_request.name == "job-2"
+    assert dict(stub.create_request.labels) == {"team": "aiq"}
+    assert session.sandbox.name == "job-2"
+
+
+def test_list_forwards_label_selector() -> None:
+    stub = _FakeSandboxStub()
+    client = _client_with_fake_stub(stub)
+
+    client.list(label_selector="aiq=deep-research")
+
+    assert stub.list_request is not None
+    assert stub.list_request.label_selector == "aiq=deep-research"
+
+
+def test_list_without_selector_sends_empty_string() -> None:
+    stub = _FakeSandboxStub()
+    client = _client_with_fake_stub(stub)
+
+    client.list()
+
+    assert stub.list_request is not None
+    assert stub.list_request.label_selector == ""
+
+
+def test_list_ids_forwards_label_selector() -> None:
+    stub = _FakeSandboxStub(listed=[_make_sandbox_proto("sandbox-1", "job-1")])
+    client = _client_with_fake_stub(stub)
+
+    ids = client.list_ids(label_selector="aiq=deep-research")
+
+    assert stub.list_request is not None
+    assert stub.list_request.label_selector == "aiq=deep-research"
+    assert ids == ["sandbox-1"]
+
+
+def test_sandbox_ref_retains_gateway_labels() -> None:
+    proto = _make_sandbox_proto(
+        "sandbox-1", "job-1", {"aiq": "deep-research", "env": "dev"}
+    )
+
+    ref = _sandbox_ref(proto)
+
+    assert dict(ref.labels) == {"aiq": "deep-research", "env": "dev"}
+
+
+def test_returned_labels_are_immutable() -> None:
+    proto = _make_sandbox_proto("sandbox-1", "job-1", {"aiq": "deep-research"})
+    ref = _sandbox_ref(proto)
+
+    with pytest.raises(TypeError):
+        ref.labels["mutated"] = "nope"  # type: ignore[index]
+
+
+def test_direct_sandbox_ref_construction_defaults_labels() -> None:
+    ref = SandboxRef(
+        id="sandbox-1",
+        name="job-1",
+        status=SandboxStatusRef(phase=2, current_policy_version=0),
+    )
+
+    assert dict(ref.labels) == {}
+
+
+def test_high_level_creation_forwards_name_and_labels(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    recording = _RecordingHighLevelClient()
+    monkeypatch.setattr(
+        SandboxClient,
+        "from_active_cluster",
+        classmethod(lambda _cls, **_kwargs: recording),
+    )
+
+    sandbox = Sandbox(
+        name="job-1", labels={"aiq": "deep-research"}, delete_on_exit=False
+    )
+    sandbox.__enter__()
+
+    assert recording.create_kwargs == {
+        "spec": None,
+        "name": "job-1",
+        "labels": {"aiq": "deep-research"},
+    }
+
+
+def test_high_level_attach_rejects_name() -> None:
+    sandbox = Sandbox(sandbox="existing-sandbox", name="job-1")
+
+    with pytest.raises(SandboxError):
+        sandbox.__enter__()
+
+
+def test_high_level_attach_rejects_labels() -> None:
+    ref = SandboxRef(
+        id="sandbox-1",
+        name="existing",
+        status=SandboxStatusRef(phase=2, current_policy_version=0),
+    )
+    sandbox = Sandbox(sandbox=ref, labels={"aiq": "deep-research"})
+
+    with pytest.raises(SandboxError):
+        sandbox.__enter__()

From 9b7acc6f3b5560ff2e4ff341e686515ae0c87689 Mon Sep 17 00:00:00 2001
From: Kyle Zheng <kyzheng@nvidia.com>
Date: Tue, 7 Jul 2026 16:36:23 -0700
Subject: [PATCH 03/10] fix(python): keep SandboxRef hashable and copy
 high-level labels
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Excluding the new immutable `labels` field from SandboxRef equality/hash
(`compare=False`) preserves the original (id, name, status) identity and keeps
the frozen dataclass hashable — a MappingProxyType field would otherwise make
`hash(SandboxRef(...))` raise. Also defensively copy caller-provided labels in
the high-level `Sandbox` so later caller mutation cannot change what is sent.

Signed-off-by: Kyle Zheng <kyzheng@nvidia.com>
---
 python/openshell/sandbox.py      |  9 +++++++--
 python/openshell/sandbox_test.py | 11 +++++++++++
 2 files changed, 18 insertions(+), 2 deletions(-)

diff --git a/python/openshell/sandbox.py b/python/openshell/sandbox.py
index 07b9675c0..805cbfa2f 100644
--- a/python/openshell/sandbox.py
+++ b/python/openshell/sandbox.py
@@ -136,7 +136,11 @@ class SandboxRef:
     id: str
     name: str
     status: SandboxStatusRef
-    labels: Mapping[str, str] = field(default_factory=lambda: MappingProxyType({}))
+    # Excluded from equality/hash so SandboxRef stays hashable (MappingProxyType
+    # is unhashable) and keeps its original (id, name, status) identity.
+    labels: Mapping[str, str] = field(
+        default_factory=lambda: MappingProxyType({}), compare=False
+    )
 
     @property
     def phase(self) -> int:
@@ -699,7 +703,8 @@ def __init__(
         self._delete_on_exit = delete_on_exit
         self._spec = spec
         self._name = name
-        self._labels = labels
+        # Copy so later caller mutation cannot change what gets sent on enter.
+        self._labels = dict(labels) if labels is not None else None
         self._timeout = timeout
         self._ready_timeout_seconds = ready_timeout_seconds
         self._auto_refresh = auto_refresh
diff --git a/python/openshell/sandbox_test.py b/python/openshell/sandbox_test.py
index 2e9c9b275..7703ffd4a 100644
--- a/python/openshell/sandbox_test.py
+++ b/python/openshell/sandbox_test.py
@@ -1601,6 +1601,17 @@ def test_direct_sandbox_ref_construction_defaults_labels() -> None:
     assert dict(ref.labels) == {}
 
 
+def test_sandbox_ref_stays_hashable_with_labels_excluded_from_identity() -> None:
+    ref_a = _sandbox_ref(_make_sandbox_proto("sandbox-1", "job-1", {"aiq": "a"}))
+    ref_b = _sandbox_ref(_make_sandbox_proto("sandbox-1", "job-1", {"aiq": "b"}))
+
+    # Frozen dataclass must remain hashable despite the immutable labels field.
+    assert hash(ref_a) == hash(ref_b)
+    # Labels are excluded from identity: same (id, name, status) compares equal.
+    assert ref_a == ref_b
+    assert {ref_a, ref_b} == {ref_a}
+
+
 def test_high_level_creation_forwards_name_and_labels(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

From 2ae3416285937612456cdc2be6d924b9d0d2779b Mon Sep 17 00:00:00 2001
From: Kyle Zheng <kyzheng@nvidia.com>
Date: Tue, 7 Jul 2026 20:58:43 -0700
Subject: [PATCH 04/10] fix(sandbox): bound initial-policy-ack retries

The poll loop retried a pending initial acknowledgement before processing any
newer revision, but retried unconditionally forever. A permanently
undeliverable ack (e.g. the revision was superseded before it could be
reported) would then stall all later policy hot-reloads and provider-env
refreshes. Cap the retries; after the bound, give up and resume normal polling
so the loop cannot livelock on a stuck acknowledgement.

Signed-off-by: Kyle Zheng <kyzheng@nvidia.com>
---
 crates/openshell-sandbox/src/lib.rs | 27 +++++++++++++++++++++++++--
 1 file changed, 25 insertions(+), 2 deletions(-)

diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs
index 56deb0f3b..87e37fd99 100644
--- a/crates/openshell-sandbox/src/lib.rs
+++ b/crates/openshell-sandbox/src/lib.rs
@@ -1766,6 +1766,11 @@ struct PolicyPollLoopContext {
     policy_local_ctx: Option<Arc<openshell_supervisor_network::policy_local::PolicyLocalContext>>,
 }
 
+/// Maximum attempts to deliver a pending initial policy acknowledgement before
+/// giving up and resuming normal polling, so a permanently undeliverable ack
+/// cannot stall later hot-reloads.
+const MAX_PENDING_ACK_ATTEMPTS: u32 = 6;
+
 async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
     use openshell_core::grpc_client::CachedOpenShellClient;
     use openshell_core::proto::PolicySource;
@@ -1785,6 +1790,10 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
     // status-report failure never fails a healthy sandbox, and so a newer
     // revision cannot be acknowledged ahead of the one actually loaded.
     let mut pending_initial_ack: Option<InitialPolicyAck> = None;
+    // Bounded so a permanently-undeliverable initial acknowledgement (e.g. the
+    // revision was superseded before it could be reported) cannot block newer
+    // policy revisions and provider-env refreshes forever.
+    let mut pending_ack_attempts: u32 = 0;
 
     // Initialize revision from the first poll and acknowledge the initial
     // policy revision the supervisor actually loaded. Seeding the loop from
@@ -1832,12 +1841,26 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
 
         // Deliver any pending initial acknowledgement before processing newer
         // revisions, so the revision actually loaded is acknowledged first and
-        // policy history is never reordered.
+        // policy history is never reordered. Bounded: after a fixed number of
+        // failed attempts, give up and resume normal polling so a permanently
+        // undeliverable ack cannot stall later hot-reloads.
         if let Some(candidate) = pending_initial_ack.clone() {
             if report_initial_policy_ack(&client, &ctx.sandbox_id, &candidate).await {
                 pending_initial_ack = None;
+                pending_ack_attempts = 0;
             } else {
-                continue;
+                pending_ack_attempts += 1;
+                if pending_ack_attempts >= MAX_PENDING_ACK_ATTEMPTS {
+                    warn!(
+                        version = candidate.version,
+                        attempts = pending_ack_attempts,
+                        "Giving up on pending initial policy acknowledgement; resuming normal polling"
+                    );
+                    pending_initial_ack = None;
+                    pending_ack_attempts = 0;
+                } else {
+                    continue;
+                }
             }
         }
 

From e7c379074f6c95c928964e14bfece4d26861efd2 Mon Sep 17 00:00:00 2001
From: Kyle Zheng <kyzheng@nvidia.com>
Date: Tue, 7 Jul 2026 20:58:43 -0700
Subject: [PATCH 05/10] test(sandbox): add sparse-policy revision-2
 acknowledgement e2e

Regression for #2159: create a sandbox with the network-only policy-advisor
fixture, which the supervisor enriches with baseline filesystem paths during
startup (creating revision 2, superseding revision 1). Assert the effective
policy reaches revision 2 and no revision remains Pending once the supervisor
acknowledges the load. Adds SandboxGuard::create_keep_with_args to create a
kept sandbox with an initial --policy.

Signed-off-by: Kyle Zheng <kyzheng@nvidia.com>
---
 e2e/rust/src/harness/sandbox.rs      | 27 +++++++++--
 e2e/rust/tests/live_policy_update.rs | 72 ++++++++++++++++++++++++++++
 2 files changed, 94 insertions(+), 5 deletions(-)

diff --git a/e2e/rust/src/harness/sandbox.rs b/e2e/rust/src/harness/sandbox.rs
index af71a38ee..43fa4ee39 100644
--- a/e2e/rust/src/harness/sandbox.rs
+++ b/e2e/rust/src/harness/sandbox.rs
@@ -126,13 +126,30 @@ impl SandboxGuard {
     pub async fn create_keep(
         command: &[&str],
         ready_marker: &str,
+    ) -> Result<Self, String> {
+        Self::create_keep_with_args(&[], command, ready_marker).await
+    }
+
+    /// Like [`SandboxGuard::create_keep`], but forwards extra flags to
+    /// `sandbox create` (e.g. `--policy <file>`, `--name <n>`) before the
+    /// `-- <command>` separator.
+    ///
+    /// # Errors
+    ///
+    /// Returns an error if the process exits prematurely, the ready marker is
+    /// not seen within [`SANDBOX_READY_TIMEOUT`], or the sandbox name cannot
+    /// be parsed.
+    pub async fn create_keep_with_args(
+        create_args: &[&str],
+        command: &[&str],
+        ready_marker: &str,
     ) -> Result<Self, String> {
         let mut cmd = openshell_cmd();
-        cmd.arg("sandbox")
-            .arg("create")
-            .arg("--keep")
-            .arg("--")
-            .args(command);
+        cmd.arg("sandbox").arg("create").arg("--keep");
+        for arg in create_args {
+            cmd.arg(arg);
+        }
+        cmd.arg("--").args(command);
         cmd.stdout(Stdio::piped()).stderr(Stdio::piped());
 
         let mut child = cmd
diff --git a/e2e/rust/tests/live_policy_update.rs b/e2e/rust/tests/live_policy_update.rs
index c60b29548..ec7e534bd 100644
--- a/e2e/rust/tests/live_policy_update.rs
+++ b/e2e/rust/tests/live_policy_update.rs
@@ -421,3 +421,75 @@ async fn live_policy_update_from_empty_network_policies() {
 
     guard.cleanup().await;
 }
+
+/// Regression for #2159: a sparse initial policy that the supervisor enriches
+/// with baseline filesystem paths during startup must have its resulting
+/// revision acknowledged as loaded, not left `Pending`.
+///
+/// Reproduction (matches the maintainer's triage): create a sandbox with the
+/// network-only `examples/policy-advisor/sandbox-policy.yaml`. The supervisor
+/// adds baseline filesystem paths, syncs the enriched policy back to the
+/// gateway (creating revision 2, superseding revision 1), builds the OPA engine
+/// and then acknowledges revision 2 as loaded. Before the fix, revision 2
+/// stayed `Pending` even though the sandbox was `Ready` and the policy was
+/// effective.
+///
+/// NOTE: This exercises the Docker-backed supervisor built from this branch.
+/// The exact `policy list` status wording ("Loaded"/"Superseded") may differ by
+/// CLI version; the assertions below key on the effective version reaching 2 and
+/// no revision remaining `Pending` once the acknowledgement lands.
+#[tokio::test]
+async fn initial_sparse_policy_is_acknowledged_as_loaded() {
+    // Repo-relative path to the sparse network-only policy fixture.
+    let sparse_policy = concat!(
+        env!("CARGO_MANIFEST_DIR"),
+        "/../../examples/policy-advisor/sandbox-policy.yaml"
+    );
+
+    let mut guard = SandboxGuard::create_keep_with_args(
+        &["--name", "e2e-2159-sparse-enrich", "--policy", sparse_policy, "--no-tty"],
+        &["sh", "-c", "echo Ready && sleep infinity"],
+        "Ready",
+    )
+    .await
+    .expect("create keep sandbox with sparse policy");
+
+    // The enriched revision (2) is synced during startup; the acknowledgement
+    // (LOADED) is delivered by the supervisor's poll loop shortly after Ready.
+    // Poll until the effective policy is version 2 and no revision is Pending.
+    let mut acknowledged = false;
+    let mut last_list = String::new();
+    for _ in 0..30 {
+        let get = run_cli(&["policy", "get", &guard.name]).await;
+        let version = get.success.then(|| extract_version(&get.output)).flatten();
+
+        let list = run_cli(&["policy", "list", &guard.name]).await;
+        last_list = list.output.clone();
+        let pending = list.output.to_lowercase().contains("pending");
+
+        if version == Some(2) && list.success && !pending {
+            acknowledged = true;
+            break;
+        }
+        tokio::time::sleep(std::time::Duration::from_secs(2)).await;
+    }
+
+    assert!(
+        acknowledged,
+        "enriched initial policy should reach revision 2 with no Pending revision.\n\
+         last `policy list` output:\n{last_list}"
+    );
+
+    // Both the superseded original (1) and the loaded enriched revision (2)
+    // must appear in the revision history.
+    assert!(
+        list_output_contains_version(&last_list, 2),
+        "policy list should contain revision 2:\n{last_list}"
+    );
+    assert!(
+        list_output_contains_version(&last_list, 1),
+        "policy list should contain revision 1:\n{last_list}"
+    );
+
+    guard.cleanup().await;
+}

From 8bc38fafd4591a64a326a91a7b050583264bea66 Mon Sep 17 00:00:00 2001
From: Kyle Zheng <kyzheng@nvidia.com>
Date: Wed, 8 Jul 2026 09:35:09 -0700
Subject: [PATCH 06/10] fix(ci): correct sandbox checks

Signed-off-by: Kyle Zheng <kyzheng@nvidia.com>
---
 crates/openshell-sandbox/src/lib.rs | 34 ++++++++++++++++-------------
 python/openshell/sandbox_test.py    |  4 ++--
 2 files changed, 21 insertions(+), 17 deletions(-)

diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs
index 87e37fd99..c87eeb838 100644
--- a/crates/openshell-sandbox/src/lib.rs
+++ b/crates/openshell-sandbox/src/lib.rs
@@ -1808,17 +1808,19 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
                 initial_policy_ack_candidate(ctx.loaded_policy.as_ref(), &result)
             {
                 if report_initial_policy_ack(&client, &ctx.sandbox_id, &candidate).await {
-                    ocsf_emit!(ConfigStateChangeBuilder::new(ocsf_ctx())
-                        .severity(SeverityId::Informational)
-                        .status(StatusId::Success)
-                        .state(StateId::Enabled, "loaded")
-                        .unmapped("version", serde_json::json!(candidate.version))
-                        .unmapped("policy_hash", serde_json::json!(&candidate.policy_hash))
-                        .message(format!(
-                            "Acknowledged initial policy revision as loaded [version:{}]",
-                            candidate.version
-                        ))
-                        .build());
+                    ocsf_emit!(
+                        ConfigStateChangeBuilder::new(ocsf_ctx())
+                            .severity(SeverityId::Informational)
+                            .status(StatusId::Success)
+                            .state(StateId::Enabled, "loaded")
+                            .unmapped("version", serde_json::json!(candidate.version))
+                            .unmapped("policy_hash", serde_json::json!(&candidate.policy_hash))
+                            .message(format!(
+                                "Acknowledged initial policy revision as loaded [version:{}]",
+                                candidate.version
+                            ))
+                            .build()
+                    );
                 } else {
                     pending_initial_ack = Some(candidate);
                 }
@@ -2380,8 +2382,7 @@ filesystem_policy:
     #[test]
     fn initial_ack_candidate_rejects_missing_canonical_policy() {
         let loaded = proto_policy_fixture();
-        let canonical =
-            settings_poll_result(None, 2, openshell_core::proto::PolicySource::Sandbox);
+        let canonical = settings_poll_result(None, 2, openshell_core::proto::PolicySource::Sandbox);
 
         assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
     }
@@ -2394,8 +2395,11 @@ filesystem_policy:
             "user_rule".to_string(),
             openshell_core::proto::NetworkPolicyRule::default(),
         );
-        let canonical =
-            settings_poll_result(Some(different), 2, openshell_core::proto::PolicySource::Sandbox);
+        let canonical = settings_poll_result(
+            Some(different),
+            2,
+            openshell_core::proto::PolicySource::Sandbox,
+        );
 
         assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
     }
diff --git a/python/openshell/sandbox_test.py b/python/openshell/sandbox_test.py
index 7703ffd4a..72850b108 100644
--- a/python/openshell/sandbox_test.py
+++ b/python/openshell/sandbox_test.py
@@ -66,7 +66,7 @@ class _Response:
         return _Response()
 
 
-def _client_with_fake_stub(stub: _FakeStub) -> SandboxClient:
+def _client_with_fake_stub(stub: object) -> SandboxClient:
     client = cast("SandboxClient", object.__new__(SandboxClient))
     client._timeout = 30.0
     client._stub = cast("Any", stub)
@@ -1428,7 +1428,7 @@ def _make_sandbox_proto(
     id_: str,
     name: str,
     labels: dict[str, str] | None = None,
-    phase: int = openshell_pb2.SANDBOX_PHASE_READY,
+    phase: openshell_pb2.SandboxPhase = openshell_pb2.SANDBOX_PHASE_READY,
     version: int = 0,
 ) -> openshell_pb2.Sandbox:
     sandbox = openshell_pb2.Sandbox()

From 74bfbe4996192259b056c726be603f441e8eabb4 Mon Sep 17 00:00:00 2001
From: Kyle Zheng <kyzheng@nvidia.com>
Date: Wed, 8 Jul 2026 10:14:56 -0700
Subject: [PATCH 07/10] test(cli): serialize mTLS environment access

Signed-off-by: Kyle Zheng <kyzheng@nvidia.com>
---
 crates/openshell-cli/tests/mtls_integration.rs | 2 ++
 1 file changed, 2 insertions(+)

diff --git a/crates/openshell-cli/tests/mtls_integration.rs b/crates/openshell-cli/tests/mtls_integration.rs
index 28a4e6c9c..3f51dd604 100644
--- a/crates/openshell-cli/tests/mtls_integration.rs
+++ b/crates/openshell-cli/tests/mtls_integration.rs
@@ -668,6 +668,7 @@ async fn gateway_add_mtls_loopback_explicit_name_does_not_fallback_to_openshell_
 
 #[tokio::test]
 async fn cli_connects_with_client_cert() {
+    let _env = EnvVarGuard::set(&[]);
     install_rustls_provider();
 
     let (ca, ca_key) = build_ca();
@@ -741,6 +742,7 @@ async fn run_server_no_client_auth(
 
 #[tokio::test]
 async fn cli_connects_with_gateway_insecure() {
+    let _env = EnvVarGuard::set(&[]);
     install_rustls_provider();
 
     let (ca, ca_key) = build_ca();

From 2ab114a5d51f3d0caeeebf502de9d4447a1ddd23 Mon Sep 17 00:00:00 2001
From: Kyle Zheng <kyzheng@nvidia.com>
Date: Wed, 8 Jul 2026 13:32:32 -0700
Subject: [PATCH 08/10] fix(sandbox): address policy review feedback

Signed-off-by: Kyle Zheng <kyzheng@nvidia.com>
---
 architecture/sandbox.md             |  12 +-
 crates/openshell-sandbox/src/lib.rs | 321 ++++++++++++++++++----------
 python/openshell/sandbox.py         |  43 +++-
 python/openshell/sandbox_test.py    |  40 ++++
 4 files changed, 285 insertions(+), 131 deletions(-)

diff --git a/architecture/sandbox.md b/architecture/sandbox.md
index 0de0b7d2a..d9d7a215f 100644
--- a/architecture/sandbox.md
+++ b/architecture/sandbox.md
@@ -183,10 +183,14 @@ original construction error.
 This holds even when the initial policy is enriched with baseline paths during
 startup: the enriched revision the supervisor synced back to the gateway is the
 revision it acknowledges, so a successfully constructed initial policy never
-remains `Pending`. The acknowledgement is delivered with bounded retries and is
-non-fatal: a transient delivery failure does not fail an otherwise healthy
-sandbox, and the pending acknowledgement is retried before any newer revision is
-processed so policy history is never reordered.
+remains `Pending`. If the first poll returns a different revision, the supervisor
+processes it through the normal reload path instead of treating it as already
+loaded.
+
+Policy status delivery uses a FIFO background worker with bounded retries. This
+preserves report order without blocking policy polling, enforcement, settings,
+or provider refreshes when the status endpoint is unavailable. Delivery failure
+is non-fatal and logged after the retry bound is exhausted.
 
 Only sandbox-scoped revisions (`PolicySource::Sandbox`, version greater than
 zero) are acknowledged. Global policies and local-file development policies do
diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs
index c87eeb838..5b988cde6 100644
--- a/crates/openshell-sandbox/src/lib.rs
+++ b/crates/openshell-sandbox/src/lib.rs
@@ -1625,6 +1625,51 @@ struct InitialPolicyAck {
     config_revision: u64,
 }
 
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct PolicyStatusUpdate {
+    version: u32,
+    loaded: bool,
+    error: String,
+    initial_policy_hash: Option<String>,
+}
+
+const POLICY_STATUS_QUEUE_CAPACITY: usize = 32;
+
+impl PolicyStatusUpdate {
+    fn initial_loaded(ack: &InitialPolicyAck) -> Self {
+        Self {
+            version: ack.version,
+            loaded: true,
+            error: String::new(),
+            initial_policy_hash: Some(ack.policy_hash.clone()),
+        }
+    }
+
+    fn loaded(version: u32) -> Self {
+        Self {
+            version,
+            loaded: true,
+            error: String::new(),
+            initial_policy_hash: None,
+        }
+    }
+
+    fn failed(version: u32, error: String) -> Self {
+        Self {
+            version,
+            loaded: false,
+            error,
+            initial_policy_hash: None,
+        }
+    }
+}
+
+#[derive(Clone, Debug, PartialEq, Eq)]
+enum InitialPollDisposition {
+    Acknowledge(InitialPolicyAck),
+    Reconcile,
+}
+
 /// Determine whether the initially loaded policy corresponds to an
 /// authoritative sandbox-scoped revision that must be acknowledged.
 ///
@@ -1655,6 +1700,16 @@ fn initial_policy_ack_candidate(
     })
 }
 
+fn initial_poll_disposition(
+    loaded: Option<&openshell_core::proto::SandboxPolicy>,
+    canonical: &openshell_core::grpc_client::SettingsPollResult,
+) -> InitialPollDisposition {
+    initial_policy_ack_candidate(loaded, canonical).map_or(
+        InitialPollDisposition::Reconcile,
+        InitialPollDisposition::Acknowledge,
+    )
+}
+
 /// Compare two sandbox policies for enforcement-equivalent content.
 ///
 /// The policy `version` field and provider-managed rule names are ignored:
@@ -1674,27 +1729,76 @@ fn policies_structurally_match(
     a == b
 }
 
-/// Report a successfully constructed initial policy revision as loaded,
-/// using the shared bounded gRPC retry. Returns whether delivery succeeded;
-/// a transient failure is non-fatal and retried later by the poll loop.
-async fn report_initial_policy_ack(
-    client: &openshell_core::grpc_client::CachedOpenShellClient,
-    sandbox_id: &str,
-    ack: &InitialPolicyAck,
-) -> bool {
-    grpc_retry("Initial policy acknowledgement", || {
-        let client = client.clone();
-        async move {
-            client
-                .report_policy_status(sandbox_id, ack.version, true, "")
-                .await
+/// Deliver policy status updates independently from policy reconciliation.
+///
+/// The channel is FIFO, so a delayed older status can never arrive after a
+/// newer status and move the gateway's active version backward. Delivery uses
+/// the existing bounded retry, but failures never delay policy enforcement.
+async fn run_policy_status_reporter(
+    client: openshell_core::grpc_client::CachedOpenShellClient,
+    sandbox_id: String,
+    mut updates: tokio::sync::mpsc::Receiver<PolicyStatusUpdate>,
+) {
+    while let Some(update) = updates.recv().await {
+        let operation = if update.initial_policy_hash.is_some() {
+            "Initial policy acknowledgement"
+        } else {
+            "Policy status report"
+        };
+        let version = update.version;
+        let loaded = update.loaded;
+        let result = grpc_retry(operation, || {
+            let sandbox_id = sandbox_id.clone();
+            let error = update.error.clone();
+            let client = client.clone();
+            async move {
+                client
+                    .report_policy_status(&sandbox_id, version, loaded, &error)
+                    .await
+            }
+        })
+        .await;
+
+        if let Err(error) = result {
+            warn!(
+                %error,
+                version = update.version,
+                loaded = update.loaded,
+                "Failed to deliver policy status update; continuing enforcement"
+            );
+            continue;
         }
-    })
-    .await
-    .inspect_err(|e| {
-        warn!(error = %e, version = ack.version, "Failed to acknowledge initial policy revision");
-    })
-    .is_ok()
+
+        if let Some(policy_hash) = update.initial_policy_hash {
+            ocsf_emit!(
+                ConfigStateChangeBuilder::new(ocsf_ctx())
+                    .severity(SeverityId::Informational)
+                    .status(StatusId::Success)
+                    .state(StateId::Enabled, "loaded")
+                    .unmapped("version", serde_json::json!(update.version))
+                    .unmapped("policy_hash", serde_json::json!(policy_hash))
+                    .message(format!(
+                        "Acknowledged initial policy revision as loaded [version:{}]",
+                        update.version
+                    ))
+                    .build()
+            );
+        }
+    }
+}
+
+fn enqueue_policy_status(
+    sender: &tokio::sync::mpsc::Sender<PolicyStatusUpdate>,
+    update: PolicyStatusUpdate,
+) {
+    let version = update.version;
+    if let Err(error) = sender.try_send(update) {
+        warn!(
+            %error,
+            version,
+            "Policy status reporter unavailable; dropping update"
+        );
+    }
 }
 
 /// Best-effort `FAILED` acknowledgement when initial policy construction or
@@ -1766,17 +1870,19 @@ struct PolicyPollLoopContext {
     policy_local_ctx: Option<Arc<openshell_supervisor_network::policy_local::PolicyLocalContext>>,
 }
 
-/// Maximum attempts to deliver a pending initial policy acknowledgement before
-/// giving up and resuming normal polling, so a permanently undeliverable ack
-/// cannot stall later hot-reloads.
-const MAX_PENDING_ACK_ATTEMPTS: u32 = 6;
-
 async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
     use openshell_core::grpc_client::CachedOpenShellClient;
     use openshell_core::proto::PolicySource;
     use std::sync::atomic::Ordering;
 
     let client = CachedOpenShellClient::connect(&ctx.endpoint).await?;
+    let (status_sender, status_receiver) = tokio::sync::mpsc::channel(POLICY_STATUS_QUEUE_CAPACITY);
+    tokio::spawn(run_policy_status_reporter(
+        client.clone(),
+        ctx.sandbox_id.clone(),
+        status_receiver,
+    ));
+
     let mut current_config_revision: u64 = 0;
     let mut current_provider_env_revision: u64 = ctx.provider_credentials.snapshot().revision;
     let mut current_policy_hash = String::new();
@@ -1785,53 +1891,32 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
         openshell_core::proto::EffectiveSetting,
     > = std::collections::HashMap::new();
 
-    // A sandbox-scoped initial revision that was built successfully but whose
-    // acknowledgement has not yet been delivered. Held so a transient
-    // status-report failure never fails a healthy sandbox, and so a newer
-    // revision cannot be acknowledged ahead of the one actually loaded.
-    let mut pending_initial_ack: Option<InitialPolicyAck> = None;
-    // Bounded so a permanently-undeliverable initial acknowledgement (e.g. the
-    // revision was superseded before it could be reported) cannot block newer
-    // policy revisions and provider-env refreshes forever.
-    let mut pending_ack_attempts: u32 = 0;
+    // A first poll that does not match the policy already loaded into OPA must
+    // pass through the normal reconciliation path immediately. It must never
+    // seed the applied-state trackers before OPA actually loads it.
+    let mut pending_result = None;
 
     // Initialize revision from the first poll and acknowledge the initial
-    // policy revision the supervisor actually loaded. Seeding the loop from
-    // that revision is what prevents it from being re-reported on later polls.
+    // policy revision the supervisor actually loaded. A mismatched result is
+    // reconciled below instead of being recorded as already applied.
     match client.poll_settings(&ctx.sandbox_id).await {
-        Ok(result) => {
-            apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings);
-            current_config_revision = result.config_revision;
-            current_policy_hash = result.policy_hash.clone();
-
-            if let Some(candidate) =
-                initial_policy_ack_candidate(ctx.loaded_policy.as_ref(), &result)
-            {
-                if report_initial_policy_ack(&client, &ctx.sandbox_id, &candidate).await {
-                    ocsf_emit!(
-                        ConfigStateChangeBuilder::new(ocsf_ctx())
-                            .severity(SeverityId::Informational)
-                            .status(StatusId::Success)
-                            .state(StateId::Enabled, "loaded")
-                            .unmapped("version", serde_json::json!(candidate.version))
-                            .unmapped("policy_hash", serde_json::json!(&candidate.policy_hash))
-                            .message(format!(
-                                "Acknowledged initial policy revision as loaded [version:{}]",
-                                candidate.version
-                            ))
-                            .build()
-                    );
-                } else {
-                    pending_initial_ack = Some(candidate);
-                }
+        Ok(result) => match initial_poll_disposition(ctx.loaded_policy.as_ref(), &result) {
+            InitialPollDisposition::Acknowledge(candidate) => {
+                apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings);
+                current_config_revision = candidate.config_revision;
+                current_policy_hash.clone_from(&candidate.policy_hash);
+                current_settings = result.settings;
+                enqueue_policy_status(
+                    &status_sender,
+                    PolicyStatusUpdate::initial_loaded(&candidate),
+                );
+                debug!(
+                    config_revision = current_config_revision,
+                    "Settings poll: initial policy matches loaded revision"
+                );
             }
-
-            current_settings = result.settings;
-            debug!(
-                config_revision = current_config_revision,
-                "Settings poll: initial config revision"
-            );
-        }
+            InitialPollDisposition::Reconcile => pending_result = Some(result),
+        },
         Err(e) => {
             warn!(error = %e, "Settings poll: failed to fetch initial version, will retry");
         }
@@ -1839,39 +1924,17 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
 
     let interval = Duration::from_secs(ctx.interval_secs);
     loop {
-        tokio::time::sleep(interval).await;
-
-        // Deliver any pending initial acknowledgement before processing newer
-        // revisions, so the revision actually loaded is acknowledged first and
-        // policy history is never reordered. Bounded: after a fixed number of
-        // failed attempts, give up and resume normal polling so a permanently
-        // undeliverable ack cannot stall later hot-reloads.
-        if let Some(candidate) = pending_initial_ack.clone() {
-            if report_initial_policy_ack(&client, &ctx.sandbox_id, &candidate).await {
-                pending_initial_ack = None;
-                pending_ack_attempts = 0;
-            } else {
-                pending_ack_attempts += 1;
-                if pending_ack_attempts >= MAX_PENDING_ACK_ATTEMPTS {
-                    warn!(
-                        version = candidate.version,
-                        attempts = pending_ack_attempts,
-                        "Giving up on pending initial policy acknowledgement; resuming normal polling"
-                    );
-                    pending_initial_ack = None;
-                    pending_ack_attempts = 0;
-                } else {
+        let result = if let Some(result) = pending_result.take() {
+            result
+        } else {
+            tokio::time::sleep(interval).await;
+            match client.poll_settings(&ctx.sandbox_id).await {
+                Ok(result) => result,
+                Err(e) => {
+                    debug!(error = %e, "Settings poll: server unreachable, will retry");
                     continue;
                 }
             }
-        }
-
-        let result = match client.poll_settings(&ctx.sandbox_id).await {
-            Ok(r) => r,
-            Err(e) => {
-                debug!(error = %e, "Settings poll: server unreachable, will retry");
-                continue;
-            }
         };
 
         let provider_env_changed = result.provider_env_revision != current_provider_env_revision;
@@ -1986,13 +2049,11 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
                                 .build()
                         );
                     }
-                    if result.version > 0
-                        && result.policy_source == PolicySource::Sandbox
-                        && let Err(e) = client
-                            .report_policy_status(&ctx.sandbox_id, result.version, true, "")
-                            .await
-                    {
-                        warn!(error = %e, "Failed to report policy load success");
+                    if result.version > 0 && result.policy_source == PolicySource::Sandbox {
+                        enqueue_policy_status(
+                            &status_sender,
+                            PolicyStatusUpdate::loaded(result.version),
+                        );
                     }
                 }
                 Err(e) => {
@@ -2007,18 +2068,11 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
                             result.version
                         ))
                         .build());
-                    if result.version > 0
-                        && result.policy_source == PolicySource::Sandbox
-                        && let Err(report_err) = client
-                            .report_policy_status(
-                                &ctx.sandbox_id,
-                                result.version,
-                                false,
-                                &e.to_string(),
-                            )
-                            .await
-                    {
-                        warn!(error = %report_err, "Failed to report policy load failure");
+                    if result.version > 0 && result.policy_source == PolicySource::Sandbox {
+                        enqueue_policy_status(
+                            &status_sender,
+                            PolicyStatusUpdate::failed(result.version, e.to_string()),
+                        );
                     }
                 }
             }
@@ -2404,6 +2458,39 @@ filesystem_policy:
         assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
     }
 
+    #[test]
+    fn initial_poll_reconciles_policy_that_was_not_loaded() {
+        let loaded = proto_policy_fixture();
+        let mut newer = proto_policy_fixture();
+        newer.network_policies.insert(
+            "new_deny_rule".to_string(),
+            openshell_core::proto::NetworkPolicyRule::default(),
+        );
+        let canonical =
+            settings_poll_result(Some(newer), 2, openshell_core::proto::PolicySource::Sandbox);
+
+        assert_eq!(
+            initial_poll_disposition(Some(&loaded), &canonical),
+            InitialPollDisposition::Reconcile
+        );
+    }
+
+    #[test]
+    fn policy_status_queue_preserves_revision_order() {
+        let (sender, mut receiver) = tokio::sync::mpsc::channel(2);
+        enqueue_policy_status(&sender, PolicyStatusUpdate::loaded(1));
+        enqueue_policy_status(
+            &sender,
+            PolicyStatusUpdate::failed(2, "invalid policy".to_string()),
+        );
+
+        assert_eq!(receiver.try_recv().unwrap(), PolicyStatusUpdate::loaded(1));
+        assert_eq!(
+            receiver.try_recv().unwrap(),
+            PolicyStatusUpdate::failed(2, "invalid policy".to_string())
+        );
+    }
+
     #[test]
     fn structural_match_ignores_version_field() {
         let mut a = proto_policy_fixture();
diff --git a/python/openshell/sandbox.py b/python/openshell/sandbox.py
index 805cbfa2f..9b00ebfb7 100644
--- a/python/openshell/sandbox.py
+++ b/python/openshell/sandbox.py
@@ -14,8 +14,7 @@
 import time
 from collections import namedtuple
 from dataclasses import dataclass, field
-from types import MappingProxyType
-from typing import TYPE_CHECKING, cast
+from typing import TYPE_CHECKING, Any, Never, SupportsIndex, cast
 from urllib.parse import urlparse
 
 import grpc
@@ -131,16 +130,42 @@ class SandboxStatusRef:
     current_policy_version: int
 
 
+class _ImmutableLabels(dict[str, str]):
+    """A read-only, copy- and pickle-safe label mapping."""
+
+    def _deny_mutation(self, *args: object, **kwargs: object) -> Never:
+        del args, kwargs
+        raise TypeError("sandbox labels are immutable")
+
+    __setitem__ = _deny_mutation
+    __delitem__ = _deny_mutation
+    clear = _deny_mutation
+    pop = _deny_mutation
+    popitem = _deny_mutation
+    setdefault = _deny_mutation
+    update = _deny_mutation
+    __ior__ = _deny_mutation
+
+    def __deepcopy__(self, memo: dict[int, object]) -> _ImmutableLabels:
+        del memo
+        return type(self)(self)
+
+    def __reduce_ex__(self, protocol: SupportsIndex, /) -> str | tuple[Any, ...]:
+        del protocol
+        return type(self), (dict(self),)
+
+
 @dataclass(frozen=True)
 class SandboxRef:
     id: str
     name: str
     status: SandboxStatusRef
-    # Excluded from equality/hash so SandboxRef stays hashable (MappingProxyType
-    # is unhashable) and keeps its original (id, name, status) identity.
-    labels: Mapping[str, str] = field(
-        default_factory=lambda: MappingProxyType({}), compare=False
-    )
+    # Excluded from equality/hash to preserve the original identity while the
+    # immutable mapping remains safe for deepcopy, pickle, and asdict.
+    labels: Mapping[str, str] = field(default_factory=_ImmutableLabels, compare=False)
+
+    def __post_init__(self) -> None:
+        object.__setattr__(self, "labels", _ImmutableLabels(self.labels))
 
     @property
     def phase(self) -> int:
@@ -864,9 +889,7 @@ def _sandbox_ref(sandbox: openshell_pb2.Sandbox) -> SandboxRef:
             phase=status.phase if status else 0,
             current_policy_version=status.current_policy_version if status else 0,
         ),
-        labels=MappingProxyType(
-            dict(sandbox.metadata.labels) if sandbox.metadata else {}
-        ),
+        labels=sandbox.metadata.labels if sandbox.metadata else {},
     )
 
 
diff --git a/python/openshell/sandbox_test.py b/python/openshell/sandbox_test.py
index 72850b108..700a9d1ed 100644
--- a/python/openshell/sandbox_test.py
+++ b/python/openshell/sandbox_test.py
@@ -5,8 +5,11 @@
 
 import json
 import os
+import pickle
 import threading
 import time
+from copy import deepcopy
+from dataclasses import asdict
 from pathlib import Path
 from types import SimpleNamespace
 from typing import Any, cast
@@ -1612,6 +1615,43 @@ def test_sandbox_ref_stays_hashable_with_labels_excluded_from_identity() -> None
     assert {ref_a, ref_b} == {ref_a}
 
 
+def test_sandbox_ref_labels_support_standard_serialization() -> None:
+    ref = _sandbox_ref(
+        _make_sandbox_proto("sandbox-1", "job-1", {"aiq": "deep-research"})
+    )
+
+    assert asdict(ref)["labels"] == {"aiq": "deep-research"}
+    assert dict(deepcopy(ref).labels) == {"aiq": "deep-research"}
+    assert dict(pickle.loads(pickle.dumps(ref)).labels) == {"aiq": "deep-research"}
+
+
+def test_default_sandbox_ref_labels_support_standard_serialization() -> None:
+    ref = SandboxRef(
+        id="sandbox-1",
+        name="job-1",
+        status=SandboxStatusRef(phase=2, current_policy_version=0),
+    )
+
+    assert asdict(ref)["labels"] == {}
+    assert dict(deepcopy(ref).labels) == {}
+    assert dict(pickle.loads(pickle.dumps(ref)).labels) == {}
+
+
+def test_direct_sandbox_ref_copies_and_freezes_labels() -> None:
+    labels = {"aiq": "deep-research"}
+    ref = SandboxRef(
+        id="sandbox-1",
+        name="job-1",
+        status=SandboxStatusRef(phase=2, current_policy_version=0),
+        labels=labels,
+    )
+    labels["aiq"] = "mutated"
+
+    assert dict(ref.labels) == {"aiq": "deep-research"}
+    with pytest.raises(TypeError):
+        ref.labels["mutated"] = "nope"  # type: ignore[index]
+
+
 def test_high_level_creation_forwards_name_and_labels(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:

From 5089665327caf449dc640f3d792294a7cffbaa56 Mon Sep 17 00:00:00 2001
From: Kyle Zheng <kyzheng@nvidia.com>
Date: Wed, 8 Jul 2026 16:00:18 -0700
Subject: [PATCH 09/10] fix(sandbox): preserve exact policy acknowledgements

Signed-off-by: Kyle Zheng <kyzheng@nvidia.com>
---
 architecture/sandbox.md                  |  22 +-
 crates/openshell-core/src/grpc_client.rs |  73 ++--
 crates/openshell-sandbox/src/lib.rs      | 402 ++++++++++++-----------
 3 files changed, 273 insertions(+), 224 deletions(-)

diff --git a/architecture/sandbox.md b/architecture/sandbox.md
index d9d7a215f..7d90043ed 100644
--- a/architecture/sandbox.md
+++ b/architecture/sandbox.md
@@ -173,12 +173,14 @@ quickly.
 
 ## Policy Revision Acknowledgement
 
-When the supervisor loads a sandbox-scoped policy from the gateway, it
-acknowledges the exact revision it constructed. After the OPA engine is built
-successfully, the supervisor reports that revision as `LOADED`, which advances
+When the supervisor loads a sandbox-scoped policy from the gateway, it retains
+the version, hash, source, and configuration revision returned with that exact
+policy snapshot. After the OPA engine is built successfully, the supervisor
+reports that revision as `LOADED`, which advances
 `SandboxStatus.current_policy_version` and moves the revision out of `Pending`.
-If policy construction fails, it reports the same revision as `FAILED` with the
-original construction error.
+If policy construction fails, it reports the captured revision as `FAILED` with
+the original construction error. It never infers revision identity by comparing
+policy structure.
 
 This holds even when the initial policy is enriched with baseline paths during
 startup: the enriched revision the supervisor synced back to the gateway is the
@@ -187,10 +189,12 @@ remains `Pending`. If the first poll returns a different revision, the superviso
 processes it through the normal reload path instead of treating it as already
 loaded.
 
-Policy status delivery uses a FIFO background worker with bounded retries. This
-preserves report order without blocking policy polling, enforcement, settings,
-or provider refreshes when the status endpoint is unavailable. Delivery failure
-is non-fatal and logged after the retry bound is exhausted.
+Policy status delivery uses a FIFO background worker. Retryable delivery
+failures retain the ordered update and retry with capped exponential backoff;
+terminal errors are logged and discarded. The outbox is nonblocking and does
+not discard updates because of a fixed queue capacity, so status endpoint
+outages cannot block policy polling, enforcement, settings, or provider
+refreshes and cannot permanently lose the initial acknowledgement.
 
 Only sandbox-scoped revisions (`PolicySource::Sandbox`, version greater than
 zero) are acknowledged. Global policies and local-file development policies do
diff --git a/crates/openshell-core/src/grpc_client.rs b/crates/openshell-core/src/grpc_client.rs
index 96158a1d1..57f6bdd81 100644
--- a/crates/openshell-core/src/grpc_client.rs
+++ b/crates/openshell-core/src/grpc_client.rs
@@ -573,11 +573,23 @@ pub async fn fetch_policy(endpoint: &str, sandbox_id: &str) -> Result<Option<Pro
     fetch_policy_with_client(&mut client, sandbox_id).await
 }
 
-/// Fetch sandbox policy using an existing client connection.
-async fn fetch_policy_with_client(
+/// Fetch the authoritative policy and revision metadata in one response.
+///
+/// Callers that must acknowledge the exact revision they loaded should retain
+/// this snapshot instead of re-fetching metadata after policy construction.
+pub async fn fetch_settings_snapshot(
+    endpoint: &str,
+    sandbox_id: &str,
+) -> Result<SettingsPollResult> {
+    debug!(endpoint = %endpoint, sandbox_id = %sandbox_id, "Connecting to fetch OpenShell settings snapshot");
+    let mut client = connect(endpoint).await?;
+    fetch_settings_snapshot_with_client(&mut client, sandbox_id).await
+}
+
+async fn fetch_settings_snapshot_with_client(
     client: &mut OpenShellClient<AuthedChannel>,
     sandbox_id: &str,
-) -> Result<Option<ProtoSandboxPolicy>> {
+) -> Result<SettingsPollResult> {
     let response = client
         .get_sandbox_config(GetSandboxConfigRequest {
             sandbox_id: sandbox_id.to_string(),
@@ -585,14 +597,22 @@ async fn fetch_policy_with_client(
         .await
         .into_diagnostic()?;
 
-    let inner = response.into_inner();
+    Ok(settings_poll_result(response.into_inner()))
+}
+
+/// Fetch sandbox policy using an existing client connection.
+async fn fetch_policy_with_client(
+    client: &mut OpenShellClient<AuthedChannel>,
+    sandbox_id: &str,
+) -> Result<Option<ProtoSandboxPolicy>> {
+    let snapshot = fetch_settings_snapshot_with_client(client, sandbox_id).await?;
 
     // version 0 with no policy means the sandbox was created without one.
-    if inner.version == 0 && inner.policy.is_none() {
+    if snapshot.version == 0 && snapshot.policy.is_none() {
         return Ok(None);
     }
 
-    Ok(Some(inner.policy.ok_or_else(|| {
+    Ok(Some(snapshot.policy.ok_or_else(|| {
         miette::miette!("Server returned non-zero version but empty policy")
     })?))
 }
@@ -661,6 +681,18 @@ pub async fn sync_policy(endpoint: &str, sandbox: &str, policy: &ProtoSandboxPol
     sync_policy_with_client(&mut client, sandbox, policy).await
 }
 
+/// Sync an enriched policy and return the authoritative revision snapshot.
+pub async fn sync_policy_and_fetch_snapshot(
+    endpoint: &str,
+    sandbox_id: &str,
+    sandbox: &str,
+    policy: &ProtoSandboxPolicy,
+) -> Result<SettingsPollResult> {
+    let mut client = connect(endpoint).await?;
+    sync_policy_with_client(&mut client, sandbox, policy).await?;
+    fetch_settings_snapshot_with_client(&mut client, sandbox_id).await
+}
+
 /// Fetch provider environment variables for a sandbox from `OpenShell` server via gRPC.
 ///
 /// Returns a map of environment variable names to values derived from provider
@@ -700,6 +732,7 @@ pub struct CachedOpenShellClient {
 }
 
 /// Settings poll result returned by [`CachedOpenShellClient::poll_settings`].
+#[derive(Clone, Debug)]
 pub struct SettingsPollResult {
     pub policy: Option<ProtoSandboxPolicy>,
     pub version: u32,
@@ -713,6 +746,20 @@ pub struct SettingsPollResult {
     pub provider_env_revision: u64,
 }
 
+fn settings_poll_result(inner: crate::proto::GetSandboxConfigResponse) -> SettingsPollResult {
+    SettingsPollResult {
+        policy: inner.policy,
+        version: inner.version,
+        policy_hash: inner.policy_hash,
+        config_revision: inner.config_revision,
+        policy_source: PolicySource::try_from(inner.policy_source)
+            .unwrap_or(PolicySource::Unspecified),
+        settings: inner.settings,
+        global_policy_version: inner.global_policy_version,
+        provider_env_revision: inner.provider_env_revision,
+    }
+}
+
 pub struct ProviderEnvironmentResult {
     pub environment: HashMap<String, String>,
     pub provider_env_revision: u64,
@@ -743,19 +790,7 @@ impl CachedOpenShellClient {
             .await
             .into_diagnostic()?;
 
-        let inner = response.into_inner();
-
-        Ok(SettingsPollResult {
-            policy: inner.policy,
-            version: inner.version,
-            policy_hash: inner.policy_hash,
-            config_revision: inner.config_revision,
-            policy_source: PolicySource::try_from(inner.policy_source)
-                .unwrap_or(PolicySource::Unspecified),
-            settings: inner.settings,
-            global_policy_version: inner.global_policy_version,
-            provider_env_revision: inner.provider_env_revision,
-        })
+        Ok(settings_poll_result(response.into_inner()))
     }
 
     /// Submit denial summaries and/or agent-authored proposals for policy analysis.
diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs
index 5b988cde6..6f5951f9f 100644
--- a/crates/openshell-sandbox/src/lib.rs
+++ b/crates/openshell-sandbox/src/lib.rs
@@ -128,7 +128,7 @@ pub async fn run_sandbox(
     // Load policy and initialize OPA engine
     let openshell_endpoint_for_proxy = openshell_endpoint.clone();
     let sandbox_name_for_agg = sandbox.clone();
-    let (mut policy, opa_engine, retained_proto) = load_policy(
+    let (mut policy, opa_engine, retained_proto, loaded_policy_revision) = load_policy(
         sandbox_id.clone(),
         sandbox,
         openshell_endpoint.clone(),
@@ -418,7 +418,7 @@ pub async fn run_sandbox(
             endpoint: poll_endpoint,
             sandbox_id: poll_id,
             opa_engine: poll_engine,
-            loaded_policy: retained_proto.clone(),
+            loaded_policy_revision,
             entrypoint_pid: poll_pid,
             interval_secs: poll_interval_secs,
             ocsf_enabled: poll_ocsf_enabled,
@@ -1371,6 +1371,7 @@ async fn load_policy(
     SandboxPolicy,
     Option<Arc<OpaEngine>>,
     Option<openshell_core::proto::SandboxPolicy>,
+    Option<LoadedPolicyRevision>,
 )> {
     // File mode: load OPA engine from rego rules + YAML data (dev override)
     if let (Some(policy_file), Some(data_file)) = (&policy_rules, &policy_data) {
@@ -1400,7 +1401,7 @@ async fn load_policy(
             process: config.process,
         };
         enrich_sandbox_baseline_paths(&mut policy);
-        return Ok((policy, Some(Arc::new(engine)), None));
+        return Ok((policy, Some(Arc::new(engine)), None, None));
     }
 
     // gRPC mode: fetch typed proto policy, construct OPA engine from baked rules + proto data
@@ -1410,12 +1411,12 @@ async fn load_policy(
             endpoint = %endpoint,
             "Fetching sandbox policy via gRPC"
         );
-        let proto_policy = grpc_retry("Policy fetch", || {
-            openshell_core::grpc_client::fetch_policy(endpoint, id)
+        let mut snapshot = grpc_retry("Policy fetch", || {
+            openshell_core::grpc_client::fetch_settings_snapshot(endpoint, id)
         })
         .await?;
 
-        let mut proto_policy = if let Some(p) = proto_policy {
+        let mut proto_policy = if let Some(p) = snapshot.policy.clone() {
             p
         } else {
             // No policy configured on the server. Discover from disk or
@@ -1443,33 +1444,68 @@ async fn load_policy(
 
             // Sync and re-fetch over a single connection to avoid extra
             // TLS handshakes.
-            grpc_retry("Policy discovery sync", || {
-                openshell_core::grpc_client::discover_and_sync_policy(
+            snapshot = grpc_retry("Policy discovery sync", || {
+                openshell_core::grpc_client::sync_policy_and_fetch_snapshot(
                     endpoint,
                     id,
                     sandbox,
                     &discovered,
                 )
             })
-            .await?
+            .await?;
+            snapshot.policy.clone().ok_or_else(|| {
+                miette::miette!("Server still returned no policy after sync — this is a bug")
+            })?
         };
 
+        // True only while `snapshot` describes the exact policy that will be
+        // constructed below. If enrichment cannot be synced and re-fetched,
+        // the policy remains enforceable but cannot be acknowledged by
+        // inferred structural equality.
+        let mut policy_bound_to_snapshot = true;
+
         // Ensure baseline filesystem paths are present for proxy-mode
         // sandboxes.  If the policy was enriched, sync the updated version
         // back to the gateway so users can see the effective policy.
         let enriched = enrich_proto_baseline_paths(&mut proto_policy);
         let sync_policy = proto_sync_payload_for_enriched_policy(&proto_policy, enriched);
-        if let Some(sync_policy) = sync_policy
-            && let Some(sandbox_name) = sandbox.as_deref()
-            && let Err(e) =
-                openshell_core::grpc_client::sync_policy(endpoint, sandbox_name, &sync_policy).await
-        {
-            warn!(
-                error = %e,
-                "Failed to sync enriched policy back to gateway (non-fatal)"
-            );
+        if let Some(sync_policy) = sync_policy {
+            if let Some(sandbox_name) = sandbox.as_deref() {
+                match openshell_core::grpc_client::sync_policy_and_fetch_snapshot(
+                    endpoint,
+                    id,
+                    sandbox_name,
+                    &sync_policy,
+                )
+                .await
+                {
+                    Ok(canonical) => {
+                        if let Some(policy) = canonical.policy.clone() {
+                            proto_policy = policy;
+                            snapshot = canonical;
+                        } else {
+                            policy_bound_to_snapshot = false;
+                            warn!(
+                                "Gateway returned no policy after enrichment sync; initial revision will be reconciled"
+                            );
+                        }
+                    }
+                    Err(e) => {
+                        policy_bound_to_snapshot = false;
+                        warn!(
+                            error = %e,
+                            "Failed to sync enriched policy back to gateway; initial revision will be reconciled"
+                        );
+                    }
+                }
+            } else {
+                policy_bound_to_snapshot = false;
+            }
         }
 
+        let loaded_policy_revision =
+            policy_bound_to_snapshot.then(|| LoadedPolicyRevision::from_snapshot(&snapshot));
+
         // 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.
@@ -1480,7 +1516,8 @@ async fn load_policy(
         let opa_engine = match OpaEngine::from_proto(&proto_policy) {
             Ok(engine) => Some(Arc::new(engine)),
             Err(e) => {
-                report_initial_policy_failure(endpoint, id, &proto_policy, &e).await;
+                report_initial_policy_failure(endpoint, id, loaded_policy_revision.as_ref(), &e)
+                    .await;
                 return Err(e);
             }
         };
@@ -1488,11 +1525,17 @@ async fn load_policy(
         let policy = match SandboxPolicy::try_from(proto_policy.clone()) {
             Ok(policy) => policy,
             Err(e) => {
-                report_initial_policy_failure(endpoint, id, &proto_policy, &e).await;
+                report_initial_policy_failure(endpoint, id, loaded_policy_revision.as_ref(), &e)
+                    .await;
                 return Err(e);
             }
         };
-        return Ok((policy, opa_engine, Some(proto_policy)));
+        return Ok((
+            policy,
+            opa_engine,
+            Some(proto_policy),
+            loaded_policy_revision,
+        ));
     }
 
     // No policy source available
@@ -1611,13 +1654,28 @@ fn discover_policy_from_path(path: &std::path::Path) -> openshell_core::proto::S
     }
 }
 
+/// Identity returned with the exact policy snapshot used to construct OPA.
+#[derive(Clone, Debug, PartialEq, Eq)]
+struct LoadedPolicyRevision {
+    version: u32,
+    policy_hash: String,
+    config_revision: u64,
+    policy_source: openshell_core::proto::PolicySource,
+}
+
+impl LoadedPolicyRevision {
+    fn from_snapshot(snapshot: &openshell_core::grpc_client::SettingsPollResult) -> Self {
+        Self {
+            version: snapshot.version,
+            policy_hash: snapshot.policy_hash.clone(),
+            config_revision: snapshot.config_revision,
+            policy_source: snapshot.policy_source,
+        }
+    }
+}
+
 /// A sandbox-scoped policy revision that was constructed successfully at
 /// startup and must be acknowledged to the gateway exactly once.
-///
-/// The supervisor loads the initial policy, builds the OPA engine, and then
-/// reports the exact revision as loaded. Without this, an enriched initial
-/// revision stays `Pending` forever because the poll loop seeds itself with
-/// that revision's hash and never treats it as a change.
 #[derive(Clone, Debug, PartialEq, Eq)]
 struct InitialPolicyAck {
     version: u32,
@@ -1633,8 +1691,6 @@ struct PolicyStatusUpdate {
     initial_policy_hash: Option<String>,
 }
 
-const POLICY_STATUS_QUEUE_CAPACITY: usize = 32;
-
 impl PolicyStatusUpdate {
     fn initial_loaded(ack: &InitialPolicyAck) -> Self {
         Self {
@@ -1674,34 +1730,38 @@ enum InitialPollDisposition {
 /// authoritative sandbox-scoped revision that must be acknowledged.
 ///
 /// Returns `Some` only for sandbox-sourced revisions (version > 0) whose
-/// canonical gateway content structurally matches the policy the supervisor
-/// loaded into the OPA engine. Global policies, local-file development
-/// policies, version zero, and content that cannot be matched yield `None`,
-/// so those paths never emit a sandbox-revision acknowledgement.
+/// captured gateway identity matches the current version and hash. Global
+/// policies, local-file development policies, version zero, and changed
+/// identities yield `None`, so those paths never emit a sandbox-revision
+/// acknowledgement.
 fn initial_policy_ack_candidate(
-    loaded: Option<&openshell_core::proto::SandboxPolicy>,
+    loaded: Option<&LoadedPolicyRevision>,
     canonical: &openshell_core::grpc_client::SettingsPollResult,
 ) -> Option<InitialPolicyAck> {
     let loaded = loaded?;
-    if canonical.policy_source != openshell_core::proto::PolicySource::Sandbox {
+    if loaded.policy_source != openshell_core::proto::PolicySource::Sandbox
+        || canonical.policy_source != openshell_core::proto::PolicySource::Sandbox
+    {
         return None;
     }
-    if canonical.version == 0 {
+    if loaded.version == 0 || canonical.version == 0 {
         return None;
     }
-    let canonical_policy = canonical.policy.as_ref()?;
-    if !policies_structurally_match(loaded, canonical_policy) {
+    if loaded.version != canonical.version
+        || loaded.policy_hash != canonical.policy_hash
+        || canonical.config_revision < loaded.config_revision
+    {
         return None;
     }
     Some(InitialPolicyAck {
-        version: canonical.version,
-        policy_hash: canonical.policy_hash.clone(),
+        version: loaded.version,
+        policy_hash: loaded.policy_hash.clone(),
         config_revision: canonical.config_revision,
     })
 }
 
 fn initial_poll_disposition(
-    loaded: Option<&openshell_core::proto::SandboxPolicy>,
+    loaded: Option<&LoadedPolicyRevision>,
     canonical: &openshell_core::grpc_client::SettingsPollResult,
 ) -> InitialPollDisposition {
     initial_policy_ack_candidate(loaded, canonical).map_or(
@@ -1710,25 +1770,6 @@ fn initial_poll_disposition(
     )
 }
 
-/// Compare two sandbox policies for enforcement-equivalent content.
-///
-/// The policy `version` field and provider-managed rule names are ignored:
-/// the gateway strips provider rule names on store and re-composes them on
-/// read, so comparing them directly would spuriously reject a revision the
-/// supervisor actually loaded.
-fn policies_structurally_match(
-    a: &openshell_core::proto::SandboxPolicy,
-    b: &openshell_core::proto::SandboxPolicy,
-) -> bool {
-    let mut a = a.clone();
-    let mut b = b.clone();
-    a.version = 0;
-    b.version = 0;
-    strip_proto_provider_policy_entries(&mut a);
-    strip_proto_provider_policy_entries(&mut b);
-    a == b
-}
-
 /// Deliver policy status updates independently from policy reconciliation.
 ///
 /// The channel is FIFO, so a delayed older status can never arrive after a
@@ -1737,36 +1778,47 @@ fn policies_structurally_match(
 async fn run_policy_status_reporter(
     client: openshell_core::grpc_client::CachedOpenShellClient,
     sandbox_id: String,
-    mut updates: tokio::sync::mpsc::Receiver<PolicyStatusUpdate>,
+    mut updates: tokio::sync::mpsc::UnboundedReceiver<PolicyStatusUpdate>,
 ) {
-    while let Some(update) = updates.recv().await {
+    'updates: while let Some(update) = updates.recv().await {
         let operation = if update.initial_policy_hash.is_some() {
             "Initial policy acknowledgement"
         } else {
             "Policy status report"
         };
-        let version = update.version;
-        let loaded = update.loaded;
-        let result = grpc_retry(operation, || {
+        let mut attempt = 1_u32;
+        loop {
             let sandbox_id = sandbox_id.clone();
             let error = update.error.clone();
             let client = client.clone();
-            async move {
-                client
-                    .report_policy_status(&sandbox_id, version, loaded, &error)
-                    .await
+            match client
+                .report_policy_status(&sandbox_id, update.version, update.loaded, &error)
+                .await
+            {
+                Ok(()) => break,
+                Err(error) if is_retryable_error(&error) => {
+                    let backoff = Duration::from_secs(1_u64 << attempt.saturating_sub(1).min(5));
+                    warn!(
+                        %error,
+                        attempt,
+                        version = update.version,
+                        loaded = update.loaded,
+                        retry_in_secs = backoff.as_secs(),
+                        "{operation} failed transiently; retaining ordered update"
+                    );
+                    tokio::time::sleep(backoff).await;
+                    attempt = attempt.saturating_add(1);
+                }
+                Err(error) => {
+                    warn!(
+                        %error,
+                        version = update.version,
+                        loaded = update.loaded,
+                        "Discarding terminal policy status update"
+                    );
+                    continue 'updates;
+                }
             }
-        })
-        .await;
-
-        if let Err(error) = result {
-            warn!(
-                %error,
-                version = update.version,
-                loaded = update.loaded,
-                "Failed to deliver policy status update; continuing enforcement"
-            );
-            continue;
         }
 
         if let Some(policy_hash) = update.initial_policy_hash {
@@ -1787,16 +1839,13 @@ async fn run_policy_status_reporter(
     }
 }
 
-fn enqueue_policy_status(
-    sender: &tokio::sync::mpsc::Sender<PolicyStatusUpdate>,
-    update: PolicyStatusUpdate,
-) {
+fn enqueue_policy_status(sender: &UnboundedSender<PolicyStatusUpdate>, update: PolicyStatusUpdate) {
     let version = update.version;
-    if let Err(error) = sender.try_send(update) {
+    if let Err(error) = sender.send(update) {
         warn!(
             %error,
             version,
-            "Policy status reporter unavailable; dropping update"
+            "Policy status reporter unavailable during shutdown"
         );
     }
 }
@@ -1804,16 +1853,21 @@ fn enqueue_policy_status(
 /// Best-effort `FAILED` acknowledgement when initial policy construction or
 /// conversion fails.
 ///
-/// Re-fetches the authoritative sandbox revision so the exact version is
-/// reported, and preserves the original construction error as the reported
-/// message. A delivery failure here is swallowed so it can never mask the
-/// original construction error returned to the caller.
+/// Uses the revision identity captured with the policy that failed to build,
+/// and preserves the original construction error as the reported message. A
+/// delivery failure here is swallowed so it can never mask that error.
 async fn report_initial_policy_failure(
     endpoint: &str,
     sandbox_id: &str,
-    loaded: &openshell_core::proto::SandboxPolicy,
+    revision: Option<&LoadedPolicyRevision>,
     error: &miette::Report,
 ) {
+    let Some(revision) = revision.filter(|revision| {
+        revision.version > 0
+            && revision.policy_source == openshell_core::proto::PolicySource::Sandbox
+    }) else {
+        return;
+    };
     let client = match openshell_core::grpc_client::CachedOpenShellClient::connect(endpoint).await {
         Ok(client) => client,
         Err(e) => {
@@ -1821,29 +1875,19 @@ async fn report_initial_policy_failure(
             return;
         }
     };
-    let canonical = match client.poll_settings(sandbox_id).await {
-        Ok(result) => result,
-        Err(e) => {
-            warn!(error = %e, "Failed to fetch config to report initial policy failure");
-            return;
-        }
-    };
-    let Some(candidate) = initial_policy_ack_candidate(Some(loaded), &canonical) else {
-        return;
-    };
     let message = error.to_string();
     if let Err(e) = grpc_retry("Initial policy failure report", || {
         let client = client.clone();
         let message = message.clone();
         async move {
             client
-                .report_policy_status(sandbox_id, candidate.version, false, &message)
+                .report_policy_status(sandbox_id, revision.version, false, &message)
                 .await
         }
     })
     .await
     {
-        warn!(error = %e, version = candidate.version, "Failed to report initial policy failure");
+        warn!(error = %e, version = revision.version, "Failed to report initial policy failure");
     }
 }
 
@@ -1859,10 +1903,9 @@ struct PolicyPollLoopContext {
     endpoint: String,
     sandbox_id: String,
     opa_engine: Arc<OpaEngine>,
-    /// The policy the supervisor loaded into the OPA engine, used to identify
-    /// and acknowledge the initial sandbox-scoped revision. `None` for
-    /// local-file mode, where no gateway acknowledgement applies.
-    loaded_policy: Option<openshell_core::proto::SandboxPolicy>,
+    /// Gateway identity captured with the exact policy used to construct OPA.
+    /// `None` when the loaded policy could not be bound to a gateway revision.
+    loaded_policy_revision: Option<LoadedPolicyRevision>,
     entrypoint_pid: Arc<AtomicU32>,
     interval_secs: u64,
     ocsf_enabled: Arc<std::sync::atomic::AtomicBool>,
@@ -1876,7 +1919,7 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
     use std::sync::atomic::Ordering;
 
     let client = CachedOpenShellClient::connect(&ctx.endpoint).await?;
-    let (status_sender, status_receiver) = tokio::sync::mpsc::channel(POLICY_STATUS_QUEUE_CAPACITY);
+    let (status_sender, status_receiver) = tokio::sync::mpsc::unbounded_channel();
     tokio::spawn(run_policy_status_reporter(
         client.clone(),
         ctx.sandbox_id.clone(),
@@ -1900,23 +1943,25 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
     // policy revision the supervisor actually loaded. A mismatched result is
     // reconciled below instead of being recorded as already applied.
     match client.poll_settings(&ctx.sandbox_id).await {
-        Ok(result) => match initial_poll_disposition(ctx.loaded_policy.as_ref(), &result) {
-            InitialPollDisposition::Acknowledge(candidate) => {
-                apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings);
-                current_config_revision = candidate.config_revision;
-                current_policy_hash.clone_from(&candidate.policy_hash);
-                current_settings = result.settings;
-                enqueue_policy_status(
-                    &status_sender,
-                    PolicyStatusUpdate::initial_loaded(&candidate),
-                );
-                debug!(
-                    config_revision = current_config_revision,
-                    "Settings poll: initial policy matches loaded revision"
-                );
+        Ok(result) => {
+            match initial_poll_disposition(ctx.loaded_policy_revision.as_ref(), &result) {
+                InitialPollDisposition::Acknowledge(candidate) => {
+                    apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings);
+                    current_config_revision = candidate.config_revision;
+                    current_policy_hash.clone_from(&candidate.policy_hash);
+                    current_settings = result.settings;
+                    enqueue_policy_status(
+                        &status_sender,
+                        PolicyStatusUpdate::initial_loaded(&candidate),
+                    );
+                    debug!(
+                        config_revision = current_config_revision,
+                        "Settings poll: initial policy matches loaded revision"
+                    );
+                }
+                InitialPollDisposition::Reconcile => pending_result = Some(result),
             }
-            InitialPollDisposition::Reconcile => pending_result = Some(result),
-        },
+        }
         Err(e) => {
             warn!(error = %e, "Settings poll: failed to fetch initial version, will retry");
         }
@@ -2381,12 +2426,12 @@ filesystem_policy:
 
     #[test]
     fn initial_ack_candidate_matches_sandbox_revision() {
-        let loaded = proto_policy_fixture();
         let canonical = settings_poll_result(
-            Some(loaded.clone()),
+            Some(proto_policy_fixture()),
             2,
             openshell_core::proto::PolicySource::Sandbox,
         );
+        let loaded = LoadedPolicyRevision::from_snapshot(&canonical);
 
         let ack = initial_policy_ack_candidate(Some(&loaded), &canonical)
             .expect("sandbox-sourced matching revision should be acknowledged");
@@ -2398,24 +2443,24 @@ filesystem_policy:
 
     #[test]
     fn initial_ack_candidate_ignores_global_policy() {
-        let loaded = proto_policy_fixture();
         let canonical = settings_poll_result(
-            Some(loaded.clone()),
+            Some(proto_policy_fixture()),
             1,
             openshell_core::proto::PolicySource::Global,
         );
+        let loaded = LoadedPolicyRevision::from_snapshot(&canonical);
 
         assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
     }
 
     #[test]
     fn initial_ack_candidate_ignores_version_zero() {
-        let loaded = proto_policy_fixture();
         let canonical = settings_poll_result(
-            Some(loaded.clone()),
+            Some(proto_policy_fixture()),
             0,
             openshell_core::proto::PolicySource::Sandbox,
         );
+        let loaded = LoadedPolicyRevision::from_snapshot(&canonical);
 
         assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
     }
@@ -2434,23 +2479,15 @@ filesystem_policy:
     }
 
     #[test]
-    fn initial_ack_candidate_rejects_missing_canonical_policy() {
-        let loaded = proto_policy_fixture();
-        let canonical = settings_poll_result(None, 2, openshell_core::proto::PolicySource::Sandbox);
-
-        assert!(initial_policy_ack_candidate(Some(&loaded), &canonical).is_none());
-    }
-
-    #[test]
-    fn initial_ack_candidate_rejects_mismatched_content() {
-        let loaded = proto_policy_fixture();
-        let mut different = proto_policy_fixture();
-        different.network_policies.insert(
-            "user_rule".to_string(),
-            openshell_core::proto::NetworkPolicyRule::default(),
+    fn initial_ack_candidate_rejects_mismatched_identity() {
+        let loaded_snapshot = settings_poll_result(
+            Some(proto_policy_fixture()),
+            1,
+            openshell_core::proto::PolicySource::Sandbox,
         );
+        let loaded = LoadedPolicyRevision::from_snapshot(&loaded_snapshot);
         let canonical = settings_poll_result(
-            Some(different),
+            Some(proto_policy_fixture()),
             2,
             openshell_core::proto::PolicySource::Sandbox,
         );
@@ -2459,15 +2496,25 @@ filesystem_policy:
     }
 
     #[test]
-    fn initial_poll_reconciles_policy_that_was_not_loaded() {
-        let loaded = proto_policy_fixture();
+    fn initial_poll_reconciles_provider_composition_that_was_not_loaded() {
+        let loaded_snapshot = settings_poll_result(
+            Some(proto_policy_fixture()),
+            1,
+            openshell_core::proto::PolicySource::Sandbox,
+        );
+        let loaded = LoadedPolicyRevision::from_snapshot(&loaded_snapshot);
         let mut newer = proto_policy_fixture();
         newer.network_policies.insert(
-            "new_deny_rule".to_string(),
+            "_provider_work_github".to_string(),
             openshell_core::proto::NetworkPolicyRule::default(),
         );
         let canonical =
-            settings_poll_result(Some(newer), 2, openshell_core::proto::PolicySource::Sandbox);
+            settings_poll_result(Some(newer), 1, openshell_core::proto::PolicySource::Sandbox);
+        let canonical = openshell_core::grpc_client::SettingsPollResult {
+            policy_hash: "hash-provider-change".to_string(),
+            config_revision: loaded.config_revision + 1,
+            ..canonical
+        };
 
         assert_eq!(
             initial_poll_disposition(Some(&loaded), &canonical),
@@ -2476,54 +2523,17 @@ filesystem_policy:
     }
 
     #[test]
-    fn policy_status_queue_preserves_revision_order() {
-        let (sender, mut receiver) = tokio::sync::mpsc::channel(2);
-        enqueue_policy_status(&sender, PolicyStatusUpdate::loaded(1));
-        enqueue_policy_status(
-            &sender,
-            PolicyStatusUpdate::failed(2, "invalid policy".to_string()),
-        );
-
-        assert_eq!(receiver.try_recv().unwrap(), PolicyStatusUpdate::loaded(1));
-        assert_eq!(
-            receiver.try_recv().unwrap(),
-            PolicyStatusUpdate::failed(2, "invalid policy".to_string())
-        );
-    }
-
-    #[test]
-    fn structural_match_ignores_version_field() {
-        let mut a = proto_policy_fixture();
-        let mut b = proto_policy_fixture();
-        a.version = 1;
-        b.version = 2;
-
-        assert!(policies_structurally_match(&a, &b));
-    }
-
-    #[test]
-    fn structural_match_ignores_provider_rule_names() {
-        // The gateway strips provider rule names on store and re-composes them
-        // on read, so a policy differing only by provider rules still matches.
-        let base = proto_policy_fixture();
-        let mut with_provider = base.clone();
-        with_provider.network_policies.insert(
-            "_provider_work_github".to_string(),
-            openshell_core::proto::NetworkPolicyRule::default(),
-        );
-
-        assert!(policies_structurally_match(&base, &with_provider));
-    }
-
-    #[test]
-    fn structural_match_detects_user_rule_differences() {
-        let base = proto_policy_fixture();
-        let mut changed = base.clone();
-        changed.network_policies.insert(
-            "user_rule".to_string(),
-            openshell_core::proto::NetworkPolicyRule::default(),
-        );
+    fn policy_status_outbox_preserves_all_revision_order() {
+        let (sender, mut receiver) = tokio::sync::mpsc::unbounded_channel();
+        for version in 1..=128 {
+            enqueue_policy_status(&sender, PolicyStatusUpdate::loaded(version));
+        }
 
-        assert!(!policies_structurally_match(&base, &changed));
+        for version in 1..=128 {
+            assert_eq!(
+                receiver.try_recv().unwrap(),
+                PolicyStatusUpdate::loaded(version)
+            );
+        }
     }
 }

From f9448b7bc5e46b44faac2295ac5de69bf3136f51 Mon Sep 17 00:00:00 2001
From: John Myers <9696606+johntmyers@users.noreply.github.com>
Date: Wed, 8 Jul 2026 16:43:00 -0700
Subject: [PATCH 10/10] fix(sandbox): preserve local policy overrides

Signed-off-by: John Myers <9696606+johntmyers@users.noreply.github.com>
---
 architecture/sandbox.md              |   5 +-
 crates/openshell-sandbox/src/lib.rs  | 146 ++++++++++++++++++------
 e2e/rust/tests/live_policy_update.rs | 164 +++++++++++++++++++++++++++
 3 files changed, 281 insertions(+), 34 deletions(-)

diff --git a/architecture/sandbox.md b/architecture/sandbox.md
index 7d90043ed..30d1bf478 100644
--- a/architecture/sandbox.md
+++ b/architecture/sandbox.md
@@ -198,7 +198,10 @@ refreshes and cannot permanently lose the initial acknowledgement.
 
 Only sandbox-scoped revisions (`PolicySource::Sandbox`, version greater than
 zero) are acknowledged. Global policies and local-file development policies do
-not use the sandbox revision API and produce no acknowledgement.
+not use the sandbox revision API and produce no acknowledgement. When explicit
+local Rego and data files are configured, the supervisor continues polling the
+gateway for settings and provider refreshes but never replaces the local OPA
+engine with a gateway policy revision.
 
 ## Failure Behavior
 
diff --git a/crates/openshell-sandbox/src/lib.rs b/crates/openshell-sandbox/src/lib.rs
index 6f5951f9f..cf193c63c 100644
--- a/crates/openshell-sandbox/src/lib.rs
+++ b/crates/openshell-sandbox/src/lib.rs
@@ -128,7 +128,7 @@ pub async fn run_sandbox(
     // Load policy and initialize OPA engine
     let openshell_endpoint_for_proxy = openshell_endpoint.clone();
     let sandbox_name_for_agg = sandbox.clone();
-    let (mut policy, opa_engine, retained_proto, loaded_policy_revision) = load_policy(
+    let (mut policy, opa_engine, retained_proto, loaded_policy_origin) = load_policy(
         sandbox_id.clone(),
         sandbox,
         openshell_endpoint.clone(),
@@ -418,7 +418,7 @@ pub async fn run_sandbox(
             endpoint: poll_endpoint,
             sandbox_id: poll_id,
             opa_engine: poll_engine,
-            loaded_policy_revision,
+            loaded_policy_origin,
             entrypoint_pid: poll_pid,
             interval_secs: poll_interval_secs,
             ocsf_enabled: poll_ocsf_enabled,
@@ -1371,7 +1371,7 @@ async fn load_policy(
     SandboxPolicy,
     Option<Arc<OpaEngine>>,
     Option<openshell_core::proto::SandboxPolicy>,
-    Option<LoadedPolicyRevision>,
+    LoadedPolicyOrigin,
 )> {
     // File mode: load OPA engine from rego rules + YAML data (dev override)
     if let (Some(policy_file), Some(data_file)) = (&policy_rules, &policy_data) {
@@ -1401,7 +1401,12 @@ async fn load_policy(
             process: config.process,
         };
         enrich_sandbox_baseline_paths(&mut policy);
-        return Ok((policy, Some(Arc::new(engine)), None, None));
+        return Ok((
+            policy,
+            Some(Arc::new(engine)),
+            None,
+            LoadedPolicyOrigin::LocalOverride,
+        ));
     }
 
     // gRPC mode: fetch typed proto policy, construct OPA engine from baked rules + proto data
@@ -1534,7 +1539,9 @@ async fn load_policy(
             policy,
             opa_engine,
             Some(proto_policy),
-            loaded_policy_revision,
+            LoadedPolicyOrigin::Gateway {
+                revision: loaded_policy_revision,
+            },
         ));
     }
 
@@ -1663,6 +1670,28 @@ struct LoadedPolicyRevision {
     policy_source: openshell_core::proto::PolicySource,
 }
 
+/// Identifies where the policy currently loaded into OPA came from.
+///
+/// A missing gateway revision means the policy was loaded from the gateway but
+/// could not be bound to an authoritative snapshot (for example, enrichment
+/// sync failed). That state must reconcile on the first successful poll. A
+/// local-file override is different: gateway policy revisions are observed for
+/// settings/provider refreshes but must never replace the explicit local OPA
+/// policy.
+#[derive(Clone, Debug, PartialEq, Eq)]
+enum LoadedPolicyOrigin {
+    LocalOverride,
+    Gateway {
+        revision: Option<LoadedPolicyRevision>,
+    },
+}
+
+impl LoadedPolicyOrigin {
+    fn allows_gateway_policy_reload(&self) -> bool {
+        matches!(self, Self::Gateway { .. })
+    }
+}
+
 impl LoadedPolicyRevision {
     fn from_snapshot(snapshot: &openshell_core::grpc_client::SettingsPollResult) -> Self {
         Self {
@@ -1724,6 +1753,7 @@ impl PolicyStatusUpdate {
 enum InitialPollDisposition {
     Acknowledge(InitialPolicyAck),
     Reconcile,
+    TrackOnly,
 }
 
 /// Determine whether the initially loaded policy corresponds to an
@@ -1761,13 +1791,18 @@ fn initial_policy_ack_candidate(
 }
 
 fn initial_poll_disposition(
-    loaded: Option<&LoadedPolicyRevision>,
+    origin: &LoadedPolicyOrigin,
     canonical: &openshell_core::grpc_client::SettingsPollResult,
 ) -> InitialPollDisposition {
-    initial_policy_ack_candidate(loaded, canonical).map_or(
-        InitialPollDisposition::Reconcile,
-        InitialPollDisposition::Acknowledge,
-    )
+    match origin {
+        LoadedPolicyOrigin::LocalOverride => InitialPollDisposition::TrackOnly,
+        LoadedPolicyOrigin::Gateway { revision } => {
+            initial_policy_ack_candidate(revision.as_ref(), canonical).map_or(
+                InitialPollDisposition::Reconcile,
+                InitialPollDisposition::Acknowledge,
+            )
+        }
+    }
 }
 
 /// Deliver policy status updates independently from policy reconciliation.
@@ -1903,9 +1938,10 @@ struct PolicyPollLoopContext {
     endpoint: String,
     sandbox_id: String,
     opa_engine: Arc<OpaEngine>,
-    /// Gateway identity captured with the exact policy used to construct OPA.
-    /// `None` when the loaded policy could not be bound to a gateway revision.
-    loaded_policy_revision: Option<LoadedPolicyRevision>,
+    /// Source of the policy currently loaded into OPA. This distinguishes an
+    /// explicit local-file override from an unbound gateway revision so the
+    /// former is never replaced by policy polling.
+    loaded_policy_origin: LoadedPolicyOrigin,
     entrypoint_pid: Arc<AtomicU32>,
     interval_secs: u64,
     ocsf_enabled: Arc<std::sync::atomic::AtomicBool>,
@@ -1943,25 +1979,33 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
     // policy revision the supervisor actually loaded. A mismatched result is
     // reconciled below instead of being recorded as already applied.
     match client.poll_settings(&ctx.sandbox_id).await {
-        Ok(result) => {
-            match initial_poll_disposition(ctx.loaded_policy_revision.as_ref(), &result) {
-                InitialPollDisposition::Acknowledge(candidate) => {
-                    apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings);
-                    current_config_revision = candidate.config_revision;
-                    current_policy_hash.clone_from(&candidate.policy_hash);
-                    current_settings = result.settings;
-                    enqueue_policy_status(
-                        &status_sender,
-                        PolicyStatusUpdate::initial_loaded(&candidate),
-                    );
-                    debug!(
-                        config_revision = current_config_revision,
-                        "Settings poll: initial policy matches loaded revision"
-                    );
-                }
-                InitialPollDisposition::Reconcile => pending_result = Some(result),
+        Ok(result) => match initial_poll_disposition(&ctx.loaded_policy_origin, &result) {
+            InitialPollDisposition::Acknowledge(candidate) => {
+                apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings);
+                current_config_revision = candidate.config_revision;
+                current_policy_hash.clone_from(&candidate.policy_hash);
+                current_settings = result.settings;
+                enqueue_policy_status(
+                    &status_sender,
+                    PolicyStatusUpdate::initial_loaded(&candidate),
+                );
+                debug!(
+                    config_revision = current_config_revision,
+                    "Settings poll: initial policy matches loaded revision"
+                );
             }
-        }
+            InitialPollDisposition::Reconcile => pending_result = Some(result),
+            InitialPollDisposition::TrackOnly => {
+                apply_ocsf_json_setting(&ctx.ocsf_enabled, &result.settings);
+                current_config_revision = result.config_revision;
+                current_policy_hash = result.policy_hash.clone();
+                current_settings = result.settings;
+                debug!(
+                    config_revision = current_config_revision,
+                    "Settings poll: tracking gateway config while preserving local policy override"
+                );
+            }
+        },
         Err(e) => {
             warn!(error = %e, "Settings poll: failed to fetch initial version, will retry");
         }
@@ -2047,7 +2091,7 @@ async fn run_policy_poll_loop(ctx: PolicyPollLoopContext) -> Result<()> {
         }
 
         // Only reload OPA when the policy payload actually changed.
-        if policy_changed {
+        if policy_changed && ctx.loaded_policy_origin.allows_gateway_policy_reload() {
             let Some(policy) = result.policy.as_ref() else {
                 ocsf_emit!(ConfigStateChangeBuilder::new(ocsf_ctx())
                     .severity(SeverityId::Medium)
@@ -2517,9 +2561,45 @@ filesystem_policy:
         };
 
         assert_eq!(
-            initial_poll_disposition(Some(&loaded), &canonical),
+            initial_poll_disposition(
+                &LoadedPolicyOrigin::Gateway {
+                    revision: Some(loaded),
+                },
+                &canonical,
+            ),
+            InitialPollDisposition::Reconcile
+        );
+    }
+
+    #[test]
+    fn initial_poll_tracks_local_override_without_reconciliation() {
+        let canonical = settings_poll_result(
+            Some(proto_policy_fixture()),
+            2,
+            openshell_core::proto::PolicySource::Sandbox,
+        );
+
+        assert_eq!(
+            initial_poll_disposition(&LoadedPolicyOrigin::LocalOverride, &canonical),
+            InitialPollDisposition::TrackOnly
+        );
+        assert!(!LoadedPolicyOrigin::LocalOverride.allows_gateway_policy_reload());
+    }
+
+    #[test]
+    fn initial_poll_reconciles_unbound_gateway_policy() {
+        let canonical = settings_poll_result(
+            Some(proto_policy_fixture()),
+            2,
+            openshell_core::proto::PolicySource::Sandbox,
+        );
+        let origin = LoadedPolicyOrigin::Gateway { revision: None };
+
+        assert_eq!(
+            initial_poll_disposition(&origin, &canonical),
             InitialPollDisposition::Reconcile
         );
+        assert!(origin.allows_gateway_policy_reload());
     }
 
     #[test]
diff --git a/e2e/rust/tests/live_policy_update.rs b/e2e/rust/tests/live_policy_update.rs
index ec7e534bd..c3bad7e80 100644
--- a/e2e/rust/tests/live_policy_update.rs
+++ b/e2e/rust/tests/live_policy_update.rs
@@ -32,6 +32,30 @@ use openshell_e2e::harness::output::{extract_field, strip_ansi};
 use openshell_e2e::harness::sandbox::SandboxGuard;
 use tempfile::NamedTempFile;
 
+#[cfg(feature = "e2e-docker")]
+const LOCAL_OVERRIDE_REGO: &str = include_str!(concat!(
+    env!("CARGO_MANIFEST_DIR"),
+    "/../../crates/openshell-supervisor-network/data/sandbox-policy.rego"
+));
+
+#[cfg(feature = "e2e-docker")]
+const LOCAL_OVERRIDE_DOCKERFILE: &str = r#"FROM public.ecr.aws/docker/library/python:3.13-slim
+
+RUN apt-get update && apt-get install -y --no-install-recommends iproute2 \
+    && rm -rf /var/lib/apt/lists/*
+RUN groupadd -g 1000660000 sandbox && \
+    useradd -m -u 1000660000 -g sandbox sandbox
+
+COPY local-policy.rego /etc/openshell/local-policy.rego
+COPY local-policy.yaml /etc/openshell/local-policy.yaml
+
+ENV OPENSHELL_POLICY_RULES=/etc/openshell/local-policy.rego
+ENV OPENSHELL_POLICY_DATA=/etc/openshell/local-policy.yaml
+ENV OPENSHELL_POLICY_POLL_INTERVAL_SECS=1
+
+CMD ["sleep", "infinity"]
+"#;
+
 // ---------------------------------------------------------------------------
 // Policy YAML builders
 // ---------------------------------------------------------------------------
@@ -130,6 +154,44 @@ process:
     Ok(file)
 }
 
+#[cfg(feature = "e2e-docker")]
+fn write_local_override_image() -> Result<tempfile::TempDir, String> {
+    let dir = tempfile::tempdir().map_err(|e| format!("create image context: {e}"))?;
+    std::fs::write(dir.path().join("Dockerfile"), LOCAL_OVERRIDE_DOCKERFILE)
+        .map_err(|e| format!("write local override Dockerfile: {e}"))?;
+    std::fs::write(dir.path().join("local-policy.rego"), LOCAL_OVERRIDE_REGO)
+        .map_err(|e| format!("write local override Rego policy: {e}"))?;
+    std::fs::write(
+        dir.path().join("local-policy.yaml"),
+        r"version: 1
+
+filesystem_policy:
+  include_workdir: true
+  read_only:
+    - /usr
+    - /lib
+    - /proc
+    - /dev/urandom
+    - /etc
+  read_write:
+    - /sandbox
+    - /tmp
+    - /dev/null
+
+landlock:
+  compatibility: best_effort
+
+process:
+  run_as_user: sandbox
+  run_as_group: sandbox
+
+network_policies: {}
+",
+    )
+    .map_err(|e| format!("write local override policy data: {e}"))?;
+    Ok(dir)
+}
+
 // ---------------------------------------------------------------------------
 // CLI helpers
 // ---------------------------------------------------------------------------
@@ -493,3 +555,105 @@ async fn initial_sparse_policy_is_acknowledged_as_loaded() {
 
     guard.cleanup().await;
 }
+
+/// An explicit local Rego/data override remains authoritative even when the
+/// sandbox has a gateway policy and that policy changes while it is running.
+/// Gateway polling must continue for settings and providers without replacing
+/// the locally loaded OPA engine.
+#[cfg(feature = "e2e-docker")]
+#[tokio::test]
+async fn local_policy_override_survives_gateway_policy_polls() {
+    let image_context = write_local_override_image().expect("write local override image");
+    let dockerfile = image_context.path().join("Dockerfile");
+    let dockerfile = dockerfile.to_str().expect("Dockerfile path should be utf-8");
+
+    let gateway_policy_a_file = write_policy(&["example.com"]).expect("write gateway policy A");
+    let gateway_policy_a_path = gateway_policy_a_file
+        .path()
+        .to_str()
+        .expect("gateway policy A path should be utf-8")
+        .to_string();
+    let gateway_policy_b_file =
+        write_policy(&["example.com", "api.anthropic.com"]).expect("write gateway policy B");
+    let gateway_policy_b_path = gateway_policy_b_file
+        .path()
+        .to_str()
+        .expect("gateway policy B path should be utf-8")
+        .to_string();
+
+    let mut guard = SandboxGuard::create_keep_with_args(
+        &[
+            "--name",
+            "e2e-local-policy-override",
+            "--from",
+            dockerfile,
+            "--policy",
+            &gateway_policy_a_path,
+            "--no-tty",
+        ],
+        &["sh", "-c", "echo Ready && sleep infinity"],
+        "Ready",
+    )
+    .await
+    .expect("create sandbox with local policy override");
+
+    // Allow several one-second poll intervals. Before the fix, the first poll
+    // immediately reloaded gateway policy A over the local override.
+    tokio::time::sleep(std::time::Duration::from_secs(4)).await;
+    let initial_logs = run_cli(&[
+        "logs",
+        &guard.name,
+        "-n",
+        "500",
+        "--since",
+        "1m",
+        "--source",
+        "sandbox",
+    ])
+    .await;
+    assert!(initial_logs.success, "fetch initial sandbox logs:\n{}", initial_logs.output);
+    assert!(
+        initial_logs
+            .output
+            .contains("Loading OPA policy engine from local files"),
+        "sandbox should load the explicit local policy:\n{}",
+        initial_logs.output
+    );
+    assert!(
+        !initial_logs.output.contains("Policy reloaded successfully"),
+        "the first gateway poll must not replace the local policy:\n{}",
+        initial_logs.output
+    );
+
+    let update = run_cli(&[
+        "policy",
+        "set",
+        &guard.name,
+        "--policy",
+        &gateway_policy_b_path,
+    ])
+    .await;
+    assert!(update.success, "publish gateway policy B:\n{}", update.output);
+
+    // A later gateway revision must also remain observational in local mode.
+    tokio::time::sleep(std::time::Duration::from_secs(4)).await;
+    let updated_logs = run_cli(&[
+        "logs",
+        &guard.name,
+        "-n",
+        "500",
+        "--since",
+        "1m",
+        "--source",
+        "sandbox",
+    ])
+    .await;
+    assert!(updated_logs.success, "fetch updated sandbox logs:\n{}", updated_logs.output);
+    assert!(
+        !updated_logs.output.contains("Policy reloaded successfully"),
+        "gateway policy updates must not replace the local override:\n{}",
+        updated_logs.output
+    );
+
+    guard.cleanup().await;
+}