fix: reconcile cloud remote approval scopes

src/codex_plugin_scanner/guard/daemon/server.py

@@ -2405,7 +2405,10 @@ def _handle_headless_remote_once(self, payload: dict[str, object]) -> None:
             return
         request_policy_action = self._optional_string(request_row.get("policy_action"))
         request_recommended_scope = self._optional_string(request_row.get("recommended_scope"))
-        if request_policy_action not in {"review", "require-reapproval"} or request_recommended_scope != "artifact":
+        if request_policy_action not in {"pause", "review", "require-reapproval"} or request_recommended_scope not in {
+            "artifact",
+            "one-time",
+        }:
             self._write_json({"error": "remote_once_not_permitted"}, status=409)
             return
         try:
@@ -2449,7 +2452,7 @@ def _handle_headless_remote_once(self, payload: dict[str, object]) -> None:
             result = self.server.store.resolve_request_with_signed_remote_result(  # type: ignore[attr-defined]
                 request_id,
                 resolution_action=resolution_action,
-                resolution_scope="artifact",
+                resolution_scope=request_recommended_scope or "artifact",
                 reason="Guard Cloud signed remote approval",
                 resolved_at=_now(),
             )
@@ -2472,7 +2475,7 @@ def _handle_headless_remote_once(self, payload: dict[str, object]) -> None:
                 "receipt_id": receipt_id,
                 "request_id": request_id,
                 "review_command": self._optional_string(resolved_request.get("review_command")),
-                "scope": "artifact",
+                "scope": request_recommended_scope or "artifact",
             },
             resolved_at,
         )

src/codex_plugin_scanner/guard/runtime/command_executors.py

@@ -320,6 +320,13 @@ def _execute_approval_operation(
         oauth=oauth,
         store=store,
     )
+    request_policy_action = _optional_string(request_row.get("policy_action"))
+    resolution_scope = _optional_string(request_row.get("recommended_scope"))
+    if request_policy_action not in {"pause", "review", "require-reapproval"} or resolution_scope not in {
+        "artifact",
+        "one-time",
+    }:
+        raise ValueError("remote_approval_not_permitted")
     receipt_id = _optional_string(envelope.get("receiptId"))
     if receipt_id is None:
         raise ValueError("invalid_remote_approval_receipt")
@@ -333,12 +340,13 @@ def _execute_approval_operation(
     if envelope_decision not in {"allow_once", "block"}:
         store.release_remote_once_receipt(receipt_id)
         raise ValueError("invalid_remote_approval_decision")
+    resolution_scope = resolution_scope or "artifact"
     resolution_action = "block" if envelope_decision == "block" else "allow"
     try:
         result = store.resolve_request_with_signed_remote_result(
             local_request_id,
             resolution_action=resolution_action,
-            resolution_scope="artifact",
+            resolution_scope=resolution_scope,
             reason=_optional_string(payload.get("reason")) or "Guard Cloud signed remote approval",
             resolved_at=generated_at,
         )

tests/test_guard_command_queue.py

@@ -1,12 +1,12 @@
 from __future__ import annotations
 
 import json
+import urllib.error
 from datetime import datetime, timedelta, timezone
 from pathlib import Path
 
-import urllib.error
-
 import pytest
+
 from codex_plugin_scanner.cli import main
 from codex_plugin_scanner.guard import store as guard_store_module
 from codex_plugin_scanner.guard.adapters.base import HarnessContext
@@ -2431,3 +2431,217 @@ def test_executor_rejects_loose_policy_memory_payload(tmp_path: Path) -> None:
     )
     assert "failureCode" in result
     assert "missing" in result["failureCode"]
+
+
+def test_executor_rejects_remote_approval_for_unsupported_scope(tmp_path: Path) -> None:
+    class UnsupportedScopeStore(FakeStore):
+        def __init__(self, guard_home: Path) -> None:
+            super().__init__(guard_home)
+            self.claimed_receipts: list[str] = []
+            self.resolved: list[dict[str, object]] = []
+            self.request_row = _approval_request_row(
+                "request-broad-scope",
+                policy_action="require-reapproval",
+                recommended_scope="workspace",
+            )
+
+        def get_approval_request(self, request_id: str) -> dict[str, object] | None:
+            return self.request_row if request_id == "request-broad-scope" else None
+
+        def claim_remote_once_receipt(
+            self,
+            receipt_id: str,
+            *,
+            request_id: str,
+            claimed_at: str,
+        ) -> bool:
+            del request_id, claimed_at
+            self.claimed_receipts.append(receipt_id)
+            return True
+
+        def resolve_request_with_signed_remote_result(
+            self,
+            request_id: str,
+            *,
+            resolution_action: str,
+            resolution_scope: str,
+            reason: str | None,
+            resolved_at: str,
+        ) -> dict[str, object]:
+            self.resolved.append(
+                {
+                    "request_id": request_id,
+                    "resolution_action": resolution_action,
+                    "resolution_scope": resolution_scope,
+                    "reason": reason,
+                    "resolved_at": resolved_at,
+                }
+            )
+            return {"resolved": True, "resolved_request": {"request_id": request_id}}
+
+    store = UnsupportedScopeStore(tmp_path / "guard-home")
+    remote_approval = _signed_remote_approval(store, store.request_row)
+
+    result = command_executors.execute_guard_command_job(
+        {
+            "operation": "guard.approval.resolve",
+            "payload": {
+                "localRequestId": "request-broad-scope",
+                "action": "allow_once",
+                "remoteApproval": remote_approval,
+            },
+        },
+        context=_context(tmp_path),
+        store=store,  # type: ignore[arg-type]
+        now=lambda: "2026-06-13T00:00:00+00:00",
+    )
+
+    assert result["failureCode"] == "remote_approval_not_permitted"
+    assert store.claimed_receipts == []
+    assert store.resolved == []
+
+
+def test_executor_resolves_one_time_scope_with_allow(tmp_path: Path) -> None:
+    """Prove `recommended_scope='one-time'` flows into `resolution_scope` on allow."""
+
+    class OneTimeAllowStore(FakeStore):
+        def __init__(self, guard_home: Path) -> None:
+            super().__init__(guard_home)
+            self.request_row = _approval_request_row(
+                "request-ot-allow",
+                policy_action="require-reapproval",
+                recommended_scope="one-time",
+            )
+            self.resolved: list[dict[str, object]] = []
+            self.claimed_receipts: list[dict[str, str]] = []
+
+        def get_approval_request(self, request_id: str) -> dict[str, object] | None:
+            return self.request_row if request_id == "request-ot-allow" else None
+
+        def claim_remote_once_receipt(
+            self,
+            receipt_id: str,
+            *,
+            request_id: str,
+            claimed_at: str,
+        ) -> bool:
+            self.claimed_receipts.append({"receipt_id": receipt_id})
+            return True
+
+        def resolve_request_with_signed_remote_result(
+            self,
+            request_id: str,
+            *,
+            resolution_action: str,
+            resolution_scope: str,
+            reason: str | None,
+            resolved_at: str,
+        ) -> dict[str, object]:
+            self.resolved.append(
+                {
+                    "request_id": request_id,
+                    "resolution_action": resolution_action,
+                    "resolution_scope": resolution_scope,
+                }
+            )
+            return {"resolved": True, "resolved_request": {"request_id": request_id}}
+
+    store = OneTimeAllowStore(tmp_path / "guard-home")
+    remote_approval = _signed_remote_approval(store, store.request_row)
+    result = command_executors.execute_guard_command_job(
+        {
+            "operation": "guard.approval.resolve",
+            "payload": {
+                "action": "allow_once",
+                "localRequestId": "request-ot-allow",
+                "remoteApproval": remote_approval,
+            },
+        },
+        context=_context(tmp_path),
+        store=store,  # type: ignore[arg-type]
+        now=lambda: "2026-06-13T00:00:00+00:00",
+    )
+
+    assert result["generatedAt"] == "2026-06-13T00:00:00+00:00"
+    assert result["data"]["status"] == "completed"
+    assert store.resolved == [
+        {
+            "request_id": "request-ot-allow",
+            "resolution_action": "allow",
+            "resolution_scope": "one-time",
+        }
+    ]
+    assert len(store.claimed_receipts) == 1
+
+
+def test_executor_resolves_one_time_scope_with_block(tmp_path: Path) -> None:
+    """Prove `recommended_scope='one-time'` flows into `resolution_scope` on block."""
+
+    class OneTimeBlockStore(FakeStore):
+        def __init__(self, guard_home: Path) -> None:
+            super().__init__(guard_home)
+            self.request_row = _approval_request_row(
+                "request-ot-block",
+                policy_action="require-reapproval",
+                recommended_scope="one-time",
+            )
+            self.resolved: list[dict[str, object]] = []
+            self.claimed_receipts: list[dict[str, str]] = []
+
+        def get_approval_request(self, request_id: str) -> dict[str, object] | None:
+            return self.request_row if request_id == "request-ot-block" else None
+
+        def claim_remote_once_receipt(
+            self,
+            receipt_id: str,
+            *,
+            request_id: str,
+            claimed_at: str,
+        ) -> bool:
+            self.claimed_receipts.append({"receipt_id": receipt_id})
+            return True
+
+        def resolve_request_with_signed_remote_result(
+            self,
+            request_id: str,
+            *,
+            resolution_action: str,
+            resolution_scope: str,
+            reason: str | None,
+            resolved_at: str,
+        ) -> dict[str, object]:
+            self.resolved.append(
+                {
+                    "request_id": request_id,
+                    "resolution_action": resolution_action,
+                    "resolution_scope": resolution_scope,
+                }
+            )
+            return {"resolved": True, "resolved_request": {"request_id": request_id}}
+
+    store = OneTimeBlockStore(tmp_path / "guard-home")
+    remote_approval = _signed_remote_approval(store, store.request_row, decision="block")
+    result = command_executors.execute_guard_command_job(
+        {
+            "operation": "guard.approval.resolve",
+            "payload": {
+                "action": "allow_once",
+                "localRequestId": "request-ot-block",
+                "remoteApproval": remote_approval,
+            },
+        },
+        context=_context(tmp_path),
+        store=store,  # type: ignore[arg-type]
+        now=lambda: "2026-06-13T00:00:00+00:00",
+    )
+
+    assert result["generatedAt"] == "2026-06-13T00:00:00+00:00"
+    assert result["data"]["status"] == "completed"
+    assert store.resolved == [
+        {
+            "request_id": "request-ot-block",
+            "resolution_action": "block",
+            "resolution_scope": "one-time",
+        }
+    ]
+    assert len(store.claimed_receipts) == 1

tests/test_guard_connect_flow.py

@@ -217,7 +217,7 @@ def fake_connect_preflight(preflight_store: GuardStore) -> dict[str, object]:
             grant_id="grant-123",
             machine_id="machine-123",
             supply_chain_entitlement={
-                "supply_chain_entitlement_expires_at": "2026-07-05T01:39:51+00:00",
+                "supply_chain_entitlement_expires_at": "2027-07-05T01:39:51+00:00",
                 "supply_chain_firewall": True,
                 "supply_chain_plan_id": "team",
             },
@@ -493,7 +493,7 @@ def close(self) -> None:
             grant_id="grant-1",
             machine_id="machine-1",
             supply_chain_entitlement={
-                "supply_chain_entitlement_expires_at": "2026-07-05T01:39:51+00:00",
+                "supply_chain_entitlement_expires_at": "2027-07-05T01:39:51+00:00",
                 "supply_chain_firewall": True,
                 "supply_chain_plan_id": "pro",
             },
@@ -560,7 +560,7 @@ def close(self) -> None:
             grant_id="grant-1",
             machine_id="machine-1",
             supply_chain_entitlement={
-                "supply_chain_entitlement_expires_at": "2026-07-05T01:39:51+00:00",
+                "supply_chain_entitlement_expires_at": "2027-07-05T01:39:51+00:00",
                 "supply_chain_firewall": True,
                 "supply_chain_plan_id": "pro",
             },
@@ -697,7 +697,7 @@ def test_sync_local_guard_cloud_proof_repairs_degraded_oauth_from_encrypted_fall
         dpop_public_jwk_thumbprint="thumbprint-old",
         grant_id="grant-old",
         machine_id="machine-old",
-        supply_chain_entitlement_expires_at="2026-07-05T01:39:51+00:00",
+        supply_chain_entitlement_expires_at="2027-07-05T01:39:51+00:00",
         supply_chain_firewall=True,
         supply_chain_plan_id="team",
         workspace_id="workspace-1",
@@ -721,7 +721,7 @@ def test_sync_local_guard_cloud_proof_repairs_degraded_oauth_from_encrypted_fall
             "access_token": "access-token-1",
             "refresh_token": "refresh-token-new",
             "package_firewall_entitlement": {
-                "supply_chain_entitlement_expires_at": "2026-07-05T01:39:51+00:00",
+                "supply_chain_entitlement_expires_at": "2027-07-05T01:39:51+00:00",
                 "supply_chain_firewall": True,
                 "supply_chain_plan_id": "team",
             },

tests/test_guard_headless_daemon_api.py

@@ -496,7 +496,7 @@ def test_supply_chain_package_firewall_connect_repairs_local_auth_and_unlocks_pa
         dpop_public_jwk_thumbprint="thumbprint-old",
         grant_id="grant-old",
         machine_id="machine-old",
-        supply_chain_entitlement_expires_at="2026-07-05T01:39:51+00:00",
+        supply_chain_entitlement_expires_at="2027-07-05T01:39:51+00:00",
         supply_chain_firewall=True,
         supply_chain_plan_id="team",
         workspace_id="workspace-1",
@@ -541,7 +541,7 @@ def close(self) -> None:
             grant_id="grant-new",
             machine_id="machine-new",
             supply_chain_entitlement={
-                "supply_chain_entitlement_expires_at": "2026-07-05T01:39:51+00:00",
+                "supply_chain_entitlement_expires_at": "2027-07-05T01:39:51+00:00",
                 "supply_chain_firewall": True,
                 "supply_chain_plan_id": "team",
             },
@@ -738,7 +738,7 @@ def close(self) -> None:
                 grant_id="grant-new",
                 machine_id="machine-new",
                 supply_chain_entitlement={
-                    "supply_chain_entitlement_expires_at": "2026-07-05T01:39:51+00:00",
+                    "supply_chain_entitlement_expires_at": "2027-07-05T01:39:51+00:00",
                     "supply_chain_firewall": True,
                     "supply_chain_plan_id": "team",
                 },
@@ -1374,7 +1374,7 @@ def test_supply_chain_package_firewall_status_accepts_paid_oauth_entitlement(tmp
         dpop_public_jwk_thumbprint="thumbprint-1",
         grant_id="grant-1",
         machine_id="machine-1",
-        supply_chain_entitlement_expires_at="2026-07-05T01:39:51+00:00",
+        supply_chain_entitlement_expires_at="2027-07-05T01:39:51+00:00",
         supply_chain_firewall=True,
         supply_chain_plan_id="pro",
         workspace_id="workspace-1",
@@ -2688,9 +2688,7 @@ def test_headless_remote_once_rejects_stale_requests_and_replays(tmp_path: Path)
             receipt_id="cloud-receipt-stale",
         )
         stale_remote_approval["actionEnvelopeHash"] = "f" * 64
-        stale_remote_approval["payloadHash"] = payload_hash_for_remote_approval_envelope(
-            stale_remote_approval
-        )
+        stale_remote_approval["payloadHash"] = payload_hash_for_remote_approval_envelope(stale_remote_approval)
         stale_remote_approval["signature"] = sign_review_payload(stale_remote_approval)
         stale_status, stale_payload = _read_json_response(
             _request(

tests/test_guard_package_shims_cli.py

@@ -26,7 +26,7 @@ def _seed_paid_oauth_entitlement(home_dir: Path) -> None:
         dpop_public_jwk_thumbprint="thumbprint-1",
         grant_id="grant-1",
         machine_id="machine-1",
-        supply_chain_entitlement_expires_at="2026-07-05T01:39:51+00:00",
+        supply_chain_entitlement_expires_at="2027-07-05T01:39:51+00:00",
         supply_chain_firewall=True,
         supply_chain_plan_id="pro",
         workspace_id="workspace-1",