test: replace /bin/sleep with portable Python sleep in test_gateway_init

/bin/sleep does not exist at FHS paths on NixOS (the KVM self-hosted
runner's OS). All 13 FileNotFoundError failures in TestSupervisor were
caused by _DaemonSpec tuples spawning /bin/sleep directly. Replace with
a module-level _SLEEP_30/_SLEEP_60 tuple using sys.executable so the
supervisor tests run on any platform. Also fix TestMainEndToEnd._run()
and remove the now-unnecessary setUpClass guard.

.gitea/workflows/test.yml

@@ -4,16 +4,15 @@
 # dependencies are required to execute it. Tests are split by directory:
 #
 #   tests/unit/         — pure unit tests; always run
-#   tests/integration/  — need a reachable Docker daemon; skip cleanly
-#                         (via tests/_docker.py:skip_unless_docker) when
-#                         Docker isn't available on the runner
+#   tests/integration/  — need a reachable backend; skip cleanly when
+#                         the backend isn't available on the runner
 #   tests/canaries/     — upstream regression canaries; run on a separate
 #                         schedule (see canaries.yml), not here
 #
-# This workflow assumes the Gitea Actions runner exposes the host Docker
-# socket to the job container so `docker` commands inside the job can
-# reach the daemon. If that's not yet configured on the runner the
-# integration tests will skip rather than fail.
+# Integration tests run once per backend in separate jobs. Each job sets
+# BOT_BOTTLE_BACKEND explicitly so the test suite uses the right backend.
+# Backends that aren't available on the runner fail the preflight step
+# rather than silently skipping inside the test output.
 
 name: test
 
@@ -23,9 +22,16 @@ on:
       - main
     paths:
       - '**.py'
+      - '.gitea/workflows/**.yml'
+      - 'scripts/**'
+      - 'README.md'
   pull_request:
     paths:
       - '**.py'
+      - '.gitea/workflows/**.yml'
+      - 'scripts/**'
+      - 'README.md'
+  workflow_dispatch:
 
 jobs:
   unit:
@@ -48,7 +54,7 @@ jobs:
       - name: Report unit coverage
         run: python3 -m coverage report -m
 
-  integration:
+  integration-docker:
     runs-on: ubuntu-latest
     steps:
       - name: Checkout
@@ -65,33 +71,96 @@ jobs:
             echo "docker not on PATH — integration tests will skip"
           fi
 
-      - name: Run integration tests
+      - name: Run integration tests (docker)
+        env:
+          BOT_BOTTLE_BACKEND: docker
         run: python3 -m unittest discover -t . -s tests/integration -v
 
-  # Combined unit+integration coverage report (informational). See
-  # docs/decisions/0004-coverage-policy.md.
+  # Integration tests against the Firecracker backend. Runs on a self-hosted
+  # KVM runner (label `kvm`) where /dev/kvm and the TAP/nft pool are available.
   #
-  # The hard diff-coverage gate (changed lines >= 90%) is DEFERRED: the
-  # Firecracker backend's VM/SSH orchestration is covered by the integration
-  # suite, which needs /dev/kvm + the provisioned TAP/nft pool — a
-  # container-based runner skips it and those lines read uncovered, so the
-  # gate can't pass here. Re-enabling it on a self-hosted KVM runner is
-  # tracked separately (see PRD 0069 / #348 and the ci-runner branch).
+  # Restricted to same-repo PRs, push to main, and workflow_dispatch — fork
+  # PRs don't execute untrusted code on the privileged runner.
+  #
+  # Runner prerequisites (provision once; see README "Firecracker on Linux"):
+  # `firecracker` on PATH, `/dev/kvm` accessible, Docker, cached kernel +
+  # static dropbear, and the pool as a persistent systemd unit.
+  integration-firecracker:
+    runs-on: [self-hosted, kvm]
+    if: >-
+      github.event_name == 'push' ||
+      github.event_name == 'workflow_dispatch' ||
+      (github.event_name == 'pull_request' &&
+       github.event.pull_request.head.repo.full_name == github.repository)
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Preflight — Firecracker host is ready
+        run: |
+          command -v firecracker >/dev/null || {
+            echo "firecracker not on PATH — provision the runner (README: Firecracker on Linux)"; exit 1; }
+          test -e /dev/kvm || { echo "/dev/kvm missing — KVM not available on this runner"; exit 1; }
+          # `backend status` exits non-zero unless the TAP pool is up + no
+          # range overlap; it prints the exact `backend setup` fix.
+          python3 cli.py backend status --backend=firecracker
+
+      # No dev-requirements install: the integration suite runs on stdlib
+      # `unittest` (pylint/pyright are lint.yml's concern, not this job's),
+      # and the self-hosted runner's Nix python env has no `pip` module
+      # (`python3 -m pip` → "No module named pip"). Nothing to install.
+      - name: Run integration tests (firecracker)
+        env:
+          BOT_BOTTLE_BACKEND: firecracker
+        run: python3 -m unittest discover -t . -s tests/integration -v
+
+  # Combined unit+integration coverage + the diff-coverage gate (the hard
+  # gate: new/changed lines >= 90%). See docs/decisions/0004-coverage-policy.md.
+  #
+  # This runs on a self-hosted KVM runner (label `kvm`), NOT ubuntu-latest,
+  # because the Firecracker backend's subprocess/VM orchestration
+  # (launch/boot/SSH/isolation-probe) is covered by the integration suite,
+  # and that suite needs `/dev/kvm` + the provisioned TAP/nft pool — which a
+  # container-based runner doesn't have. On such a runner the firecracker
+  # integration test skips and its ~230 orchestration lines read as
+  # uncovered, so the gate can't pass there.
+  #
+  # Restricted to the same events as integration-firecracker (same-repo PRs,
+  # push, workflow_dispatch) for the same security reason.
+  #
+  # See #414 for the planned follow-up: artifact-based coverage combination
+  # (run tests once in their respective jobs, combine .coverage files here).
   coverage:
-    runs-on: ubuntu-latest
+    timeout-minutes: 15
+    runs-on: [self-hosted, kvm]
+    if: >-
+      github.event_name == 'push' ||
+      github.event_name == 'workflow_dispatch' ||
+      (github.event_name == 'pull_request' &&
+       github.event.pull_request.head.repo.full_name == github.repository)
     steps:
       - name: Checkout
         uses: actions/checkout@v4
         with:
           fetch-depth: 0
 
-      # No actions/setup-python: the runner image already ships Python 3.12,
-      # and older act_runner engines mishandle setup-python's PATH (coverage
-      # lands in one interpreter, `python3` resolves to another). Install
-      # straight into the ephemeral job container's system Python —
-      # --break-system-packages is safe because the container is disposable.
-      - name: Install dev requirements
-        run: python3 -m pip install --break-system-packages -r requirements-dev.txt
+      - name: Preflight — Firecracker host is ready
+        run: |
+          command -v firecracker >/dev/null || {
+            echo "firecracker not on PATH — provision the runner (README: Firecracker on Linux)"; exit 1; }
+          test -e /dev/kvm || { echo "/dev/kvm missing — KVM not available on this runner"; exit 1; }
+          # `backend status` exits non-zero unless the TAP pool is up + no
+          # range overlap; it prints the exact `backend setup` fix.
+          python3 cli.py backend status --backend=firecracker
 
-      - name: Combined coverage report (unit + integration)
+      # No dev-requirements install: `coverage` is already provided by the
+      # self-hosted runner's Nix python env, and that env has no `pip`
+      # module to install into anyway. `scripts/coverage.sh` +
+      # `diff_coverage.py` need only `coverage` (not pylint/pyright).
+      - name: Combined coverage (unit + integration, incl. firecracker)
         run: PYTHON=python3 bash scripts/coverage.sh critical
+
+      - name: Diff-coverage gate (changed lines >= 90%)
+        run: |
+          git fetch --no-tags origin main:refs/remotes/origin/main
+          python3 scripts/diff_coverage.py --base origin/main --min 90

.gitea/workflows/tracker-policy-pr.yml

@@ -2,7 +2,7 @@ name: tracker-policy-pr
 
 on:
   pull_request:
-    types: [opened, edited, reopened, synchronized, labeled, unlabeled]
+    types: [opened, edited, reopened, synchronize, labeled, unlabeled]
 
 jobs:
   check-pr:

README.md

@@ -90,6 +90,8 @@ BOT_BOTTLE_BACKEND=firecracker ./cli.py start <agent>
 
 > **NixOS:** enable `virtualisation.docker`, ensure the KVM module is loaded (`boot.kernelModules = [ "kvm-intel" ];` or `kvm-amd`), and add your user to the `kvm` and `docker` groups. For the network pool, consume the flake module — `imports = [ inputs.bot-bottle.nixosModules.firecracker-netpool ]; services.bot-bottle-firecracker = { enable = true; owner = "you"; };` — then `nixos-rebuild switch` (imperative nft/TAP rules don't survive a rebuild; channel users can `imports = [ <bot-bottle>/nix/firecracker-netpool.nix ]`). `firecracker` isn't in nixpkgs by default as a user binary — install the release binary (pin the version) and put it on `PATH`.
 
+> **CI:** the coverage gate (`.gitea/workflows/test.yml` → `coverage` job) runs on a self-hosted runner labelled `kvm`, because the Firecracker backend's VM/SSH orchestration is exercised only by the integration suite, which needs `/dev/kvm` + the provisioned pool (a container runner would skip it and read as uncovered). Provision that runner exactly like a normal Firecracker host — `firecracker` on `PATH`, `/dev/kvm`, Docker, the cached guest kernel + static dropbear, and the pool installed as the persistent systemd unit — then register it with the `kvm` label. The unit/lint jobs still run on `ubuntu-latest`.
+
 ```sh
 ./cli.py start <agent>   # builds the image on first run, drops you into claude
 ```

bot_bottle/audit_store.py

@@ -42,7 +42,7 @@ class AuditStore(DbStore):
         super().__init__(db_path or host_db_path(), migrations)
 
     def write_audit_entry(self, entry: AuditEntry) -> Path:
-        with self._connect() as conn:
+        with self._connection() as conn:
             conn.execute(
                 """
                 INSERT INTO supervise_audit_entries (
@@ -66,7 +66,7 @@ class AuditStore(DbStore):
     def read_audit_entries(self, component: str, slug: str) -> list[AuditEntry]:
         if not self.db_path.is_file():
             return []
-        with self._connect() as conn:
+        with self._connection() as conn:
             rows = conn.execute(
                 """
                 SELECT * FROM supervise_audit_entries

bot_bottle/backend/docker/setup.py

@@ -27,10 +27,14 @@ def _docker_on_path() -> bool:
 def _daemon_reachable() -> bool:
     if not _docker_on_path():
         return False
-    return subprocess.run(
-        ["docker", "info"],
-        stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
-    ).returncode == 0
+    try:
+        return subprocess.run(
+            ["docker", "info"],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+            check=False, timeout=5,
+        ).returncode == 0
+    except subprocess.TimeoutExpired:
+        return False
 
 
 def _print_install_pointer() -> None:

bot_bottle/db_store.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import sqlite3
+from contextlib import contextmanager
 from pathlib import Path
 
 try:
@@ -28,12 +29,21 @@ class DbStore:
         conn.row_factory = sqlite3.Row
         return conn
 
+    @contextmanager
+    def _connection(self):
+        conn = self._connect()
+        try:
+            with conn:
+                yield conn
+        finally:
+            conn.close()
+
     def is_migrated(self) -> bool:
         """Return True if the DB is fully up-to-date, False if migration is needed."""
         if not self.db_path.exists():
             return False
         try:
-            with self._connect() as conn:
+            with self._connection() as conn:
                 row = conn.execute(
                     "SELECT version FROM schema_versions WHERE module = ?",
                     (self._migrations.schema_key,),
@@ -45,7 +55,7 @@ class DbStore:
 
     def migrate(self) -> None:
         """Apply any pending migrations and set permissions on the DB file."""
-        with self._connect() as conn:
+        with self._connection() as conn:
             self._migrations.apply(conn)
         self._chmod()
 

bot_bottle/orchestrator/registry.py

@@ -167,7 +167,7 @@ class RegistryStore(DbStore):
             metadata=metadata,
             policy=policy,
         )
-        with self._connect() as conn:
+        with self._connection() as conn:
             conn.execute(
                 "DELETE FROM orchestrator_bottles "
                 "WHERE source_ip = ? AND state = 'active' AND bottle_id != ?",
@@ -193,7 +193,7 @@ class RegistryStore(DbStore):
     def set_policy(self, bottle_id: str, policy: str) -> bool:
         """Update a bottle's policy in place (live reload). Returns True if
         the bottle exists."""
-        with self._connect() as conn:
+        with self._connection() as conn:
             cur = conn.execute(
                 "UPDATE orchestrator_bottles SET policy = ? WHERE bottle_id = ?",
                 (policy, bottle_id),
@@ -203,7 +203,7 @@ class RegistryStore(DbStore):
 
     def deregister(self, bottle_id: str) -> bool:
         """Remove a bottle. Returns True if a row was deleted."""
-        with self._connect() as conn:
+        with self._connection() as conn:
             cur = conn.execute(
                 "DELETE FROM orchestrator_bottles WHERE bottle_id = ?", (bottle_id,)
             )
@@ -211,7 +211,7 @@ class RegistryStore(DbStore):
 
     def get(self, bottle_id: str) -> BottleRecord | None:
         """Return the bottle by id, or None if absent."""
-        with self._connect() as conn:
+        with self._connection() as conn:
             row = conn.execute(
                 "SELECT * FROM orchestrator_bottles WHERE bottle_id = ?", (bottle_id,)
             ).fetchone()
@@ -219,7 +219,7 @@ class RegistryStore(DbStore):
 
     def all(self) -> list[BottleRecord]:
         """Every registered bottle, oldest first."""
-        with self._connect() as conn:
+        with self._connection() as conn:
             rows = conn.execute(
                 "SELECT * FROM orchestrator_bottles ORDER BY created_at"
             ).fetchall()
@@ -232,7 +232,7 @@ class RegistryStore(DbStore):
         source IP is unspoofable (Firecracker `/31` + nft) and the control
         plane is reachable only by the trusted gateway; pair with the
         identity token (`attribute`) elsewhere."""
-        with self._connect() as conn:
+        with self._connection() as conn:
             rows = conn.execute(
                 "SELECT * FROM orchestrator_bottles "
                 "WHERE source_ip = ? AND state = 'active'",

bot_bottle/queue_store.py

@@ -66,7 +66,7 @@ class QueueStore(DbStore):
         super().__init__(resolved, migrations)
 
     def write_proposal(self, proposal: Proposal) -> Path:
-        with self._connect() as conn:
+        with self._connection() as conn:
             conn.execute(
                 """
                 INSERT OR REPLACE INTO supervise_proposals (
@@ -89,7 +89,7 @@ class QueueStore(DbStore):
         return self.db_path
 
     def read_proposal(self, proposal_id: str) -> Proposal:
-        with self._connect() as conn:
+        with self._connection() as conn:
             row = conn.execute(
                 """
                 SELECT * FROM supervise_proposals
@@ -104,7 +104,7 @@ class QueueStore(DbStore):
     def list_pending_proposals(self) -> list[Proposal]:
         if not self.db_path.is_file():
             return []
-        with self._connect() as conn:
+        with self._connection() as conn:
             rows = conn.execute(
                 """
                 SELECT p.* FROM supervise_proposals p
@@ -125,7 +125,7 @@ class QueueStore(DbStore):
     def list_all_pending_proposals(self) -> list[Proposal]:
         if not self.db_path.is_file():
             return []
-        with self._connect() as conn:
+        with self._connection() as conn:
             rows = conn.execute(
                 """
                 SELECT p.* FROM supervise_proposals p
@@ -142,7 +142,7 @@ class QueueStore(DbStore):
         return [self._row_to_proposal(row) for row in rows]
 
     def write_response(self, response: Response) -> Path:
-        with self._connect() as conn:
+        with self._connection() as conn:
             conn.execute(
                 """
                 INSERT OR REPLACE INTO supervise_responses (
@@ -161,7 +161,7 @@ class QueueStore(DbStore):
         return self.db_path
 
     def read_response(self, proposal_id: str) -> Response:
-        with self._connect() as conn:
+        with self._connection() as conn:
             row = conn.execute(
                 """
                 SELECT * FROM supervise_responses
@@ -176,7 +176,7 @@ class QueueStore(DbStore):
     def archive_proposal(self, proposal_id: str) -> None:
         if not self.db_path.is_file():
             return
-        with self._connect() as conn:
+        with self._connection() as conn:
             conn.execute(
                 """
                 UPDATE supervise_proposals SET archived = 1

bot_bottle/supervise.py

@@ -47,6 +47,7 @@ from .supervise_types import (
     STATUS_MODIFIED,
     STATUS_REJECTED,
     TOOLS,
+    TOOL_CHECK_PROPOSAL,
     TOOL_EGRESS_ALLOW,
     TOOL_EGRESS_BLOCK,
     TOOL_EGRESS_TOKEN_ALLOW,
@@ -263,6 +264,7 @@ __all__ = [
     "TOOLS",
     "EGRESS_FORWARD_PROXY",
     "EGRESS_INTROSPECT_URL",
+    "TOOL_CHECK_PROPOSAL",
     "TOOL_EGRESS_ALLOW",
     "TOOL_EGRESS_BLOCK",
     "TOOL_GITLEAKS_ALLOW",

bot_bottle/supervise_server.py

@@ -2,14 +2,24 @@
 
 Per-bottle MCP server exposing tools the agent calls to propose egress
 config changes when stuck. The tools are `egress-allow`,
-`egress-block`, and `list-egress-routes`.
+`egress-block`, `list-egress-routes`, and `check-proposal`.
 
-Each queued tool call:
+Each queued proposal tool call:
 
   1. Validates the proposed file syntactically.
   2. Writes a Proposal to the host SQLite database.
-  3. Blocks polling for a matching Response row.
-  4. Returns the operator's `{status, notes}` to the agent.
+  3. Blocks polling for a matching Response row, up to a short grace
+     window (`SUPERVISE_RESPONSE_TIMEOUT_SECONDS`, default 30s).
+  4. On a decision within the window, returns the operator's
+     `{status, notes}`. On timeout, returns `status: pending` **with the
+     proposal id** and leaves the proposal queued — the flow is
+     non-blocking past the grace window (PRD prd-new / issue #412).
+
+`check-proposal` is the non-blocking companion: given a `proposal_id`
+returned by a `pending` response, it reports the current decision
+(`pending` | `approved` | `modified` | `rejected`) without re-proposing,
+so an approval made out-of-band (e.g. a web review console) can be resumed
+without holding an HTTP request open.
 
 One shared server fronts every bottle (PRD 0070) and attributes each
 proposal to the calling bottle by source IP, resolved from the orchestrator
@@ -22,7 +32,9 @@ Speaks MCP over HTTP+JSON-RPC. Methods handled:
   * `initialize`           — handshake; returns server info + caps.
   * `notifications/initialized` — ack-only.
   * `tools/list`           — returns the tool definitions.
-  * `tools/call`           — validates, queues, blocks, returns.
+  * `tools/call`           — validates, queues, waits out the grace
+                             window, returns (pending past it); or, for
+                             `check-proposal`, a non-blocking status poll.
 
 Everything else returns JSON-RPC error -32601 (method not found).
 
@@ -232,6 +244,31 @@ TOOL_DEFINITIONS: list[dict[str, object]] = [
         ),
         "inputSchema": _proposal_input_schema(),
     },
+    {
+        "name": _sv.TOOL_CHECK_PROPOSAL,
+        "description": (
+            "Poll a previously queued proposal for the operator's decision "
+            "WITHOUT blocking or re-proposing. Pass the `proposal_id` you "
+            "got back when an `egress-allow`/`egress-block` call returned "
+            "`status: pending`. Returns the current status: `pending` (no "
+            "decision yet — poll again later), `approved`, `modified`, "
+            "`rejected`, or `unknown` (no such queued proposal — wrong id, "
+            "or it was already resolved and read)."
+        ),
+        "inputSchema": {
+            "type": "object",
+            "properties": {
+                "proposal_id": {
+                    "type": "string",
+                    "description": (
+                        "The proposal id from a `pending` response."
+                    ),
+                },
+            },
+            "required": ["proposal_id"],
+            "additionalProperties": False,
+        },
+    },
 ]
 
 
@@ -353,7 +390,7 @@ def handle_tools_call(
             deadline=deadline,
         )
     except TimeoutError:
-        text = format_pending_response_text(config.response_timeout_seconds)
+        text = format_pending_response_text(proposal.id, config.response_timeout_seconds)
         return {
             "content": [{"type": "text", "text": text}],
             "isError": False,
@@ -370,6 +407,54 @@ def handle_tools_call(
     }
 
 
+def handle_check_proposal(
+    params: dict[str, object],
+    config: ServerConfig,
+) -> dict[str, object]:
+    """Non-blocking poll of a queued proposal's decision, by id.
+
+    Never creates a Proposal (so `check-proposal` isn't in `TOOLS`); it only
+    reads the queue. Resolution order mirrors the synchronous path's terminal
+    step — a decided proposal is archived here exactly as `handle_tools_call`
+    archives it after `wait_for_response`, so `pending` proposals stay visible
+    to the operator until they're both decided *and* polled."""
+    args_raw = params.get("arguments", {})
+    if not isinstance(args_raw, dict):
+        raise _RpcClientError(ERR_INVALID_PARAMS, "tools/call 'arguments' must be an object")
+    proposal_id = args_raw.get("proposal_id")
+    if not isinstance(proposal_id, str) or not proposal_id.strip():
+        raise _RpcClientError(
+            ERR_INVALID_PARAMS,
+            "check-proposal: 'proposal_id' is required and must be a non-empty string",
+        )
+    proposal_id = proposal_id.strip()
+
+    try:
+        response = _sv.read_response(config.bottle_slug, proposal_id)
+    except FileNotFoundError:
+        # No decision yet — distinguish "still queued" from "unknown id".
+        try:
+            _sv.read_proposal(config.bottle_slug, proposal_id)
+        except FileNotFoundError:
+            return {
+                "content": [{"type": "text", "text": format_unknown_proposal_text(proposal_id)}],
+                "isError": True,
+            }
+        return {
+            "content": [{"type": "text", "text": format_still_pending_text(proposal_id)}],
+            "isError": False,
+        }
+
+    try:
+        _sv.archive_proposal(config.bottle_slug, proposal_id)
+    except OSError as e:
+        raise _RpcInternalError(f"failed to archive proposal: {e}") from e
+    return {
+        "content": [{"type": "text", "text": format_response_text(response)}],
+        "isError": response.status == _sv.STATUS_REJECTED,
+    }
+
+
 def format_response_text(response: "_sv.Response") -> str:
     """Pretty-print a Response for the tool's text content. The agent
     reads the text and decides whether to retry / give up / surface."""
@@ -382,12 +467,35 @@ def format_response_text(response: "_sv.Response") -> str:
     return "\n".join(lines)
 
 
-def format_pending_response_text(timeout_seconds: float) -> str:
+def format_pending_response_text(proposal_id: str, timeout_seconds: float) -> str:
+    """Grace-window timeout: the proposal stays queued, and the agent is
+    told the id so it can `check-proposal` instead of re-proposing."""
     return "\n".join([
         "status: pending",
+        f"proposal_id: {proposal_id}",
         (
-            "notes: operator response timed out after "
-            f"{timeout_seconds:g}s; proposal remains queued"
+            f"notes: no operator decision within {timeout_seconds:g}s; the "
+            "proposal remains queued. Poll it (do not re-propose) by calling "
+            f"`check-proposal` with proposal_id={proposal_id!r}."
+        ),
+    ])
+
+
+def format_still_pending_text(proposal_id: str) -> str:
+    return "\n".join([
+        "status: pending",
+        f"proposal_id: {proposal_id}",
+        "notes: still queued; no operator decision yet. Call `check-proposal` again later.",
+    ])
+
+
+def format_unknown_proposal_text(proposal_id: str) -> str:
+    return "\n".join([
+        "status: unknown",
+        f"proposal_id: {proposal_id}",
+        (
+            "notes: no queued proposal with this id for this bottle — the id "
+            "may be wrong, or the proposal was already resolved and read."
         ),
     ])
 
@@ -482,6 +590,11 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
             # — silently dropping base routes like api.anthropic.com on approval.
             if req.params.get("name") == _sv.TOOL_LIST_EGRESS_ROUTES:
                 return self._resolved_routes_payload()
+            # `check-proposal` is a non-blocking read of the calling bottle's
+            # own queue — attributed by source IP like a proposal, but it
+            # never queues or blocks.
+            if req.params.get("name") == _sv.TOOL_CHECK_PROPOSAL:
+                return handle_check_proposal(req.params, self._attributed_config(config))
             # Attribute the proposal to the source-IP-resolved bottle, so the one
             # shared server queues each bottle's proposal under its own slug.
             return handle_tools_call(req.params, self._attributed_config(config))

bot_bottle/supervise_types.py

@@ -20,6 +20,10 @@ TOOL_EGRESS_ALLOW = "egress-allow"
 TOOL_GITLEAKS_ALLOW = "gitleaks-allow"
 TOOL_EGRESS_TOKEN_ALLOW = "egress-token-allow"
 TOOL_LIST_EGRESS_ROUTES = "list-egress-routes"
+# Read-only agent tool: poll a queued proposal for the operator's decision
+# without blocking or re-proposing. It never becomes a `Proposal.tool` (no
+# queue record is created for it), so it is intentionally NOT in `TOOLS`.
+TOOL_CHECK_PROPOSAL = "check-proposal"
 TOOLS: tuple[str, ...] = (
     TOOL_EGRESS_ALLOW,
     TOOL_EGRESS_BLOCK,
@@ -156,6 +160,7 @@ __all__ = [
     "TOOLS",
     "TOOL_EGRESS_ALLOW",
     "TOOL_EGRESS_BLOCK",
+    "TOOL_CHECK_PROPOSAL",
     "TOOL_EGRESS_TOKEN_ALLOW",
     "TOOL_GITLEAKS_ALLOW",
     "TOOL_LIST_EGRESS_ROUTES",

docs/prds/prd-new-nonblocking-supervise.md

@@ -0,0 +1,125 @@
+# PRD prd-new: Non-blocking supervise (async approval + proposal polling)
+
+- **Status:** Draft
+- **Author:** didericis
+- **Created:** 2026-07-18
+- **Issue:** #412
+
+## Summary
+
+The per-bottle supervise MCP server (`bot_bottle/supervise_server.py`)
+answers `tools/call` **synchronously**: it queues the agent's proposal and
+blocks the tool call polling for the operator's decision. On timeout it
+returns `status: pending` and leaves the proposal queued — but it hands the
+agent **no proposal id** and offers **no way to poll a specific pending
+proposal**, so the only way to learn the outcome is to re-propose (a
+duplicate).
+
+This PRD makes the MCP flow non-blocking and pollable, so an approval can
+happen out-of-band (a human taking minutes-to-hours in a review console)
+without holding an HTTP request open or wedging the agent:
+
+1. Include the `proposal_id` in the `pending` response.
+2. Add a `check-proposal` MCP tool: a non-blocking status lookup by
+   proposal id.
+3. Keep the short synchronous grace window for the common "operator is
+   right there" fast path.
+
+## Problem
+
+`handle_tools_call` → `_sv.wait_for_response(...)` blocks up to
+`SUPERVISE_RESPONSE_TIMEOUT_SECONDS` (default 30s). Two problems follow:
+
+- **Human latency ≠ tool-call latency.** A real review — rendered diff,
+  RBAC routing to an approver, someone tapping approve on their phone — is
+  minutes-to-hours. Holding the MCP request open that long is fragile
+  (proxy/keepalive timeouts, the mitmproxy egress hop, and the agent
+  harness's own tool-call timeout, which a long block can trip and stall
+  the whole turn).
+- **No resume path.** The pending fallback already exists, but without a
+  proposal id and a poll tool the agent can't reconnect to that specific
+  decision — it re-proposes, duplicating the queue entry.
+
+This is also the precondition for the planned web-console human-review
+flow (RBAC, audit retention, mobile) — see issue #412.
+
+**Safety note:** the MCP tools only *propose* policy changes; enforcement
+stays at the egress proxy and the git-gate. Returning early on `pending`
+therefore opens no hole — the agent still cannot egress or push anything
+unapproved.
+
+## Goals / success criteria
+
+- A `pending` MCP response carries the `proposal_id`.
+- An agent can call `check-proposal(proposal_id)` and get the current
+  state (`pending` | `approved` | `modified` | `rejected`) **without
+  blocking** and **without creating a new proposal**.
+- The synchronous fast path (operator approves within the grace window) is
+  unchanged: the first `tools/call` still returns the decision directly.
+- No change to enforcement, attribution (source-IP → bottle), or the
+  operator-side queue/response schema.
+
+## Non-goals
+
+- The git-gate `pre-receive` path (it is synchronous by nature and cannot
+  poll — its async variant is reject-fast + re-push; tracked as a
+  follow-up).
+- Backpressure / in-flight-proposal caps.
+- MCP server→client notifications (event-driven resume).
+- Any web-console UI (this PRD is the protocol groundwork it needs).
+
+## Design
+
+### `pending` response carries the id
+
+`handle_tools_call`'s timeout branch formats the pending text with the
+`proposal.id` and a pointer to `check-proposal`, so the agent knows what to
+poll.
+
+### `check-proposal` tool
+
+A new read-only MCP tool (`TOOL_CHECK_PROPOSAL = "check-proposal"`),
+attributed to the calling bottle by source IP exactly like the proposal
+tools. Input: `{ "proposal_id": string }`. Behavior:
+
+1. `read_response(slug, id)` →
+   - **found**: archive the proposal (same terminal step the synchronous
+     path takes) and return the decision via `format_response_text`;
+     `isError` iff rejected.
+2. **not found** → `read_proposal(slug, id)` →
+   - **found**: still queued → return `status: pending`.
+   - **not found**: unknown id, or already resolved-and-archived (e.g. a
+     second poll) → return `status: unknown`, `isError: true`.
+
+Both lookups already raise `FileNotFoundError` when absent
+(`queue_store.py`), so the handler needs no new store methods. `check-`
+`proposal` is the only path (besides the synchronous response) that
+archives, so a proposal that times out to `pending` stays visible to the
+operator until it is decided and then polled.
+
+### Grace window
+
+Left at the existing 30s default (`SUPERVISE_RESPONSE_TIMEOUT_SECONDS`),
+which doubles as the instant-approve fast path. Tuning it down is an
+operator setting, not a code change; noted for the console rollout.
+
+## Implementation chunks
+
+1. **(this PR)** `TOOL_CHECK_PROPOSAL` constant; `check-proposal` tool
+   definition + `handle_check_proposal`; dispatch wiring; `proposal_id` in
+   the pending text; unit tests. Files: `bot_bottle/supervise_types.py`,
+   `bot_bottle/supervise.py` (re-export), `bot_bottle/supervise_server.py`,
+   `tests/unit/test_supervise_server.py`.
+2. **(follow-up)** git-gate `pre-receive` reject-fast + re-push.
+3. **(follow-up)** per-bottle in-flight-proposal backpressure cap.
+4. **(follow-up)** MCP notifications for event-driven resume; web-console
+   review flow (RBAC, audit retention) on top.
+
+## Open questions
+
+- Should a resolved-but-unpolled proposal auto-archive after some TTL, or
+  only on poll? (Leaning: only on poll, so a decision is never lost to a
+  reaper before the agent sees it.)
+- Does the agent harness need an explicit "you have a pending proposal"
+  nudge, or is returning `pending` from the original call enough? (Deferred
+  to the notifications chunk.)

scripts/coverage.sh

@@ -26,8 +26,9 @@ rm -f .coverage
 echo "== unit ==" >&2
 "$PY" -m coverage run -m unittest discover -t . -s tests/unit
 
-echo "== integration (skips without Docker) ==" >&2
-"$PY" -m coverage run --append -m unittest discover -t . -s tests/integration
+echo "== integration (firecracker; skips docker tests) ==" >&2
+BOT_BOTTLE_BACKEND=firecracker SKIP_DOCKER_TESTS=1 \
+    "$PY" -m coverage run --append -m unittest discover -t . -s tests/integration
 
 echo "== combined report ==" >&2
 "$PY" -m coverage report -m

tests/_docker.py

@@ -2,23 +2,30 @@
 
 from __future__ import annotations
 
+import os
 import shutil
 import subprocess
 import unittest
 
 
 def docker_available() -> bool:
+    if os.environ.get("SKIP_DOCKER_TESTS"):
+        return False
     if shutil.which("docker") is None:
         return False
-    return (
-        subprocess.run(
-            ["docker", "info"],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=False,
-        ).returncode
-        == 0
-    )
+    try:
+        return (
+            subprocess.run(
+                ["docker", "info"],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                check=False,
+                timeout=5,
+            ).returncode
+            == 0
+        )
+    except subprocess.TimeoutExpired:
+        return False
 
 
 def skip_unless_docker(reason: str = "docker unreachable"):

tests/integration/test_sandbox_escape.py

@@ -69,11 +69,12 @@ _DUMMY_HOST_KEY = (
 
 @skip_unless_docker()
 @unittest.skipIf(
-    os.environ.get("GITEA_ACTIONS") == "true",
-    "skipped under act_runner: egress_tls_init uses a host bind mount "
-    "the runner container can't see, and the network topology hides "
-    "sibling-gateway visibility — same constraint as the other "
-    "bottle-bringup integration tests",
+    os.environ.get("GITEA_ACTIONS") == "true"
+    and os.environ.get("BOT_BOTTLE_BACKEND") != "firecracker",
+    "skipped under act_runner unless BOT_BOTTLE_BACKEND=firecracker: "
+    "egress_tls_init uses a host bind mount the runner container can't "
+    "see, and the network topology hides sibling-gateway visibility — "
+    "these constraints don't apply on the self-hosted KVM runner",
 )
 class TestSandboxEscape(unittest.TestCase):
     """End-to-end attacks against a real bottle. The bottle stays

tests/unit/test_cli_start_selector.py

@@ -57,6 +57,14 @@ class TestCmdStartSelector(unittest.TestCase):
         self._bottle_picker_mock = self._bottle_picker_patch.start()
         self._bottle_picker_mock.return_value = ["claude"]  # default: one bottle selected
 
+        # name_color_modal opens /dev/tty and blocks on keyboard input on
+        # self-hosted runners that have a real controlling terminal. Stub it
+        # out like the other tui pickers so tests don't wait for a keypress.
+        self._modal_patch = patch.object(
+            tui_mod, "name_color_modal", return_value=("researcher", ""),
+        )
+        self._modal_patch.start()
+
         self._env_patch = patch.dict(os.environ, {}, clear=False)
         self._env_patch.start()
         os.environ.pop("BOT_BOTTLE_BACKEND", None)
@@ -66,6 +74,7 @@ class TestCmdStartSelector(unittest.TestCase):
         self._launch_patch.stop()
         self._agent_picker_patch.stop()
         self._bottle_picker_patch.stop()
+        self._modal_patch.stop()
         self._env_patch.stop()
 
     # ------------------------------------------------------------------

tests/unit/test_gateway_init.py

@@ -2,9 +2,8 @@
 
 Tests both the helper functions in `bot_bottle.gateway_init`
 and the supervisor's end-to-end signal / exit-code behavior. The
-end-to-end tests use real subprocesses (`/bin/sleep`,
-`/bin/sh -c '...'`) — short-lived, no docker required — so they
-run under `tests/unit/` rather than `tests/integration/`."""
+end-to-end tests use real subprocesses — short-lived, no docker required
+— so they run under `tests/unit/` rather than `tests/integration/`."""
 
 from __future__ import annotations
 
@@ -15,7 +14,6 @@ import sys
 import time
 import unittest
 import warnings
-from pathlib import Path
 from unittest.mock import patch
 
 from bot_bottle.gateway_init import (
@@ -26,6 +24,11 @@ from bot_bottle.gateway_init import (
     _selected_daemons,
 )
 
+# /bin/sleep does not exist on FHS-free systems (e.g. NixOS). Use a
+# portable Python one-liner so supervisor tests run on any platform.
+_SLEEP_30 = (sys.executable, "-c", "import time; time.sleep(30)")
+_SLEEP_60 = (sys.executable, "-c", "import time; time.sleep(60)")
+
 
 class TestEnvForDaemon(unittest.TestCase):
     """Scope egress-only credential env vars to the egress daemon.
@@ -182,7 +185,7 @@ class TestSupervisor(unittest.TestCase):
         # up and the supervisor never set shutdown_at.
         specs = [
             _DaemonSpec("crasher", ("/bin/sh", "-c", "exit 1")),
-            _DaemonSpec("longrun", ("/bin/sleep", "30")),
+            _DaemonSpec("longrun", _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -214,7 +217,7 @@ class TestSupervisor(unittest.TestCase):
         # signal-killed longrun's negative returncode.
         specs = [
             _DaemonSpec("crasher", ("/bin/sh", "-c", "exit 1")),
-            _DaemonSpec("longrun", ("/bin/sleep", "30")),
+            _DaemonSpec("longrun", _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -259,7 +262,7 @@ class TestSupervisor(unittest.TestCase):
         )
         specs = [
             _DaemonSpec("egress", sighup_marker),
-            _DaemonSpec("other",  ("/bin/sleep", "30")),
+            _DaemonSpec("other",  _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -282,7 +285,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_forward_signal_unknown_daemon_no_op(self):
-        specs = [_DaemonSpec("a", ("/bin/sleep", "30"))]
+        specs = [_DaemonSpec("a", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         delivered = sup.forward_signal(signal.SIGHUP, "ghost")
@@ -294,8 +297,8 @@ class TestSupervisor(unittest.TestCase):
         # Restart one daemon; the other (supervise, the MCP server
         # in production) must remain untouched.
         specs = [
-            _DaemonSpec("git-gate", ("/bin/sleep", "30")),
-            _DaemonSpec("supervise", ("/bin/sleep", "30")),
+            _DaemonSpec("git-gate", _SLEEP_30),
+            _DaemonSpec("supervise", _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -319,8 +322,8 @@ class TestSupervisor(unittest.TestCase):
 
     def test_request_restart_is_drained_by_tick(self):
         specs = [
-            _DaemonSpec("git-gate", ("/bin/sleep", "30")),
-            _DaemonSpec("supervise", ("/bin/sleep", "30")),
+            _DaemonSpec("git-gate", _SLEEP_30),
+            _DaemonSpec("supervise", _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -343,7 +346,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_repeated_restart_requests_coalesce(self):
-        specs = [_DaemonSpec("git-gate", ("/bin/sleep", "30"))]
+        specs = [_DaemonSpec("git-gate", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         time.sleep(0.1)
@@ -366,7 +369,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_request_restart_unknown_daemon_no_op(self):
-        specs = [_DaemonSpec("a", ("/bin/sleep", "30"))]
+        specs = [_DaemonSpec("a", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         ok = sup.request_restart("ghost")
@@ -376,7 +379,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_restart_unknown_daemon_no_op(self):
-        specs = [_DaemonSpec("a", ("/bin/sleep", "30"))]
+        specs = [_DaemonSpec("a", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         ok = sup.restart_daemon("ghost")
@@ -385,7 +388,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_restart_during_shutdown_is_no_op(self):
-        specs = [_DaemonSpec("git-gate", ("/bin/sleep", "30"))]
+        specs = [_DaemonSpec("git-gate", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         sup.request_shutdown(reason="test")
@@ -395,7 +398,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_pending_restart_dropped_during_shutdown(self):
-        specs = [_DaemonSpec("git-gate", ("/bin/sleep", "30"))]
+        specs = [_DaemonSpec("git-gate", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         time.sleep(0.1)
@@ -413,8 +416,8 @@ class TestSupervisor(unittest.TestCase):
         # both should receive SIGTERM and exit. Signal-only
         # shutdown clamps to a zero supervisor exit code.
         specs = [
-            _DaemonSpec("a", ("/bin/sleep", "60")),
-            _DaemonSpec("b", ("/bin/sleep", "60")),
+            _DaemonSpec("a", _SLEEP_60),
+            _DaemonSpec("b", _SLEEP_60),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -449,7 +452,7 @@ class TestSupervisor(unittest.TestCase):
         self.assertEqual(0, rc)
 
     def test_idempotent_shutdown_requests(self):
-        specs = [_DaemonSpec("a", ("/bin/sleep", "60"))]
+        specs = [_DaemonSpec("a", _SLEEP_60)]
         sup = _Supervisor(specs)
         sup.start_all()
         time.sleep(0.1)
@@ -465,28 +468,22 @@ class TestSupervisor(unittest.TestCase):
 
 class TestMainEndToEnd(unittest.TestCase):
     """Run gateway_init.py as a real subprocess to cover the
-    signal-handler installation path. Skipped on platforms
-    without /bin/sleep + /bin/sh."""
-
-    @classmethod
-    def setUpClass(cls):
-        for p in ("/bin/sh", "/bin/sleep"):
-            if not Path(p).exists():
-                raise unittest.SkipTest(f"missing {p}")
+    signal-handler installation path."""
 
     def _run(self, daemons_csv: str, send_signal: int | None,
              wait_before_signal: float = 0.4,
              overall_timeout: float = 6.0) -> tuple[int, str]:
         """Spawn gateway_init.main() in a child process with the
-        DAEMONS list patched to harmless `sleep 30` commands.
+        DAEMONS list patched to harmless long-sleep commands.
         Returns (returncode, captured stdout)."""
 
         helper = (
             "import os, runpy, sys\n"
             "from bot_bottle import gateway_init as si\n"
+            "sleep_cmd = (sys.executable, '-c', 'import time; time.sleep(30)')\n"
             "si._DAEMONS = (\n"
-            "  si._DaemonSpec('alpha', ('/bin/sleep','30')),\n"
-            "  si._DaemonSpec('beta',  ('/bin/sleep','30')),\n"
+            "  si._DaemonSpec('alpha', sleep_cmd),\n"
+            "  si._DaemonSpec('beta',  sleep_cmd),\n"
             ")\n"
             "sys.exit(si.main([]))\n"
         )

tests/unit/test_supervise_server.py

@@ -32,7 +32,9 @@ from bot_bottle.supervise_server import (
     _RpcError,
     _RpcInternalError,
     _response_timeout_from_env,
+    format_pending_response_text,
     format_response_text,
+    handle_check_proposal,
     handle_initialize,
     handle_tools_call,
     handle_tools_list,
@@ -218,6 +220,7 @@ class TestHandleToolsList(unittest.TestCase):
                 _sv.TOOL_EGRESS_ALLOW,
                 _sv.TOOL_EGRESS_BLOCK,
                 _sv.TOOL_LIST_EGRESS_ROUTES,
+                _sv.TOOL_CHECK_PROPOSAL,
             ]),
             sorted(names),
         )
@@ -484,9 +487,10 @@ class TestFormatResponseText(unittest.TestCase):
 
 class TestFormatPendingResponseText(unittest.TestCase):
     def test_formats_timeout_message(self):
-        text = supervise_server.format_pending_response_text(12.5)
+        text = supervise_server.format_pending_response_text("prop-9", 12.5)
         self.assertIn("status: pending", text)
         self.assertIn("12.5s", text)
+        self.assertIn("proposal_id: prop-9", text)
 
 
 # --- End-to-end HTTP sanity ------------------------------------------------
@@ -685,5 +689,129 @@ class TestResolvedRoutesPayload(unittest.TestCase):
             _handler(None)._resolved_routes_payload()
 
 
+class TestNonBlockingSupervise(unittest.TestCase):
+    """PRD prd-new / issue #412: pending responses carry the proposal id, and
+    `check-proposal` polls a queued proposal without blocking or re-proposing."""
+
+    _ROUTES = "routes:\n  - host: example.com\n"
+
+    def setUp(self):
+        self._tmp = tempfile.TemporaryDirectory(prefix="supervise-nonblock-test.")
+        self._home_patch = use_bottle_root(Path(self._tmp.name) / ".bot-bottle")
+        self.config = ServerConfig(bottle_slug="dev")
+        _qs.QueueStore("dev").migrate()
+        _as.AuditStore().migrate()
+
+    def tearDown(self):
+        self._home_patch()
+        self._tmp.cleanup()
+
+    def _seed_proposal(self) -> "_sv.Proposal":
+        p = _sv.Proposal.new(
+            bottle_slug="dev",
+            tool=_sv.TOOL_EGRESS_ALLOW,
+            proposed_file=self._ROUTES,
+            justification="need example.com",
+            current_file_hash=_sv.sha256_hex(self._ROUTES),
+        )
+        _sv.write_proposal(p)
+        return p
+
+    def _check(self, proposal_id: str) -> dict[str, object]:
+        return handle_check_proposal({"arguments": {"proposal_id": proposal_id}}, self.config)
+
+    # --- pending response carries the id ---
+
+    def test_pending_text_includes_id_and_pointer(self):
+        text = format_pending_response_text("abc-123", 30.0)
+        self.assertIn("status: pending", text)
+        self.assertIn("proposal_id: abc-123", text)
+        self.assertIn("check-proposal", text)
+
+    def test_tools_call_timeout_returns_pending_with_id_and_stays_queued(self):
+        # No responder → the grace window expires → pending, not blocked forever.
+        result = handle_tools_call(
+            {
+                "name": _sv.TOOL_EGRESS_ALLOW,
+                "arguments": {"routes_yaml": self._ROUTES, "justification": "x"},
+            },
+            ServerConfig(bottle_slug="dev", response_timeout_seconds=0.05),
+        )
+        self.assertFalse(result["isError"])  # type: ignore[index]
+        text = result["content"][0]["text"]  # type: ignore[index]
+        self.assertIn("status: pending", text)
+        pending = _sv.list_pending_proposals("dev")
+        self.assertEqual(1, len(pending))  # still queued, not archived
+        self.assertIn(pending[0].id, text)  # agent got the id to poll
+
+    # --- check-proposal poll ---
+
+    def test_check_returns_approved_and_archives(self):
+        p = self._seed_proposal()
+        _sv.write_response("dev", _sv.Response(proposal_id=p.id, status=_sv.STATUS_APPROVED, notes="ok"))
+        result = self._check(p.id)
+        self.assertFalse(result["isError"])
+        text = result["content"][0]["text"]  # type: ignore[index]
+        self.assertIn("status: approved", text)
+        self.assertIn("notes: ok", text)
+        with self.assertRaises(FileNotFoundError):  # archived on read
+            _sv.read_proposal("dev", p.id)
+
+    def test_check_rejected_sets_isError(self):
+        p = self._seed_proposal()
+        _sv.write_response("dev", _sv.Response(proposal_id=p.id, status=_sv.STATUS_REJECTED, notes="no"))
+        result = self._check(p.id)
+        self.assertTrue(result["isError"])
+        self.assertIn("status: rejected", result["content"][0]["text"])  # type: ignore[index]
+
+    def test_check_pending_when_no_decision_yet(self):
+        p = self._seed_proposal()
+        result = self._check(p.id)
+        self.assertFalse(result["isError"])
+        text = result["content"][0]["text"]  # type: ignore[index]
+        self.assertIn("status: pending", text)
+        self.assertIn(p.id, text)
+        self.assertEqual(1, len(_sv.list_pending_proposals("dev")))  # not archived
+
+    def test_check_unknown_id_is_error(self):
+        result = self._check("no-such-proposal")
+        self.assertTrue(result["isError"])
+        self.assertIn("status: unknown", result["content"][0]["text"])  # type: ignore[index]
+
+    def test_check_missing_id_raises(self):
+        with self.assertRaises(_RpcClientError) as cm:
+            handle_check_proposal({"arguments": {}}, self.config)
+        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
+
+    def test_check_empty_id_raises(self):
+        with self.assertRaises(_RpcClientError) as cm:
+            handle_check_proposal({"arguments": {"proposal_id": "   "}}, self.config)
+        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
+
+    def test_check_arguments_must_be_object(self):
+        with self.assertRaises(_RpcClientError) as cm:
+            handle_check_proposal({"arguments": []}, self.config)
+        self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code)
+
+    def test_full_nonblocking_round_trip(self):
+        # 1. tools/call times out → pending with id
+        result = handle_tools_call(
+            {
+                "name": _sv.TOOL_EGRESS_ALLOW,
+                "arguments": {"routes_yaml": self._ROUTES, "justification": "x"},
+            },
+            ServerConfig(bottle_slug="dev", response_timeout_seconds=0.05),
+        )
+        pid = _sv.list_pending_proposals("dev")[0].id
+        self.assertIn(pid, result["content"][0]["text"])  # type: ignore[index]
+        # 2. operator decides out-of-band
+        _sv.write_response("dev", _sv.Response(proposal_id=pid, status=_sv.STATUS_APPROVED, notes="ok"))
+        # 3. agent resumes by polling — no re-proposing
+        poll = self._check(pid)
+        self.assertFalse(poll["isError"])
+        self.assertIn("status: approved", poll["content"][0]["text"])  # type: ignore[index]
+        self.assertEqual([], _sv.list_pending_proposals("dev"))  # resolved + archived
+
+
 if __name__ == "__main__":
     unittest.main()