From d8b61b36583e69f2898add3cf75875f427eb0534 Mon Sep 17 00:00:00 2001 From: claude Date: Fri, 24 Jul 2026 06:26:06 +0000 Subject: [PATCH] fix(supervise): make the response poll idempotent and slow it to 250ms MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The control-plane supervise poll archived the proposal on the same call that returned the decision, so a dropped connection lost the operator's decision (the retry saw `unknown`). Poll no longer archives — a re-poll returns the same decision. Decided proposals still drop off the operator's pending list (a response row exists); their rows are reaped when the bottle is torn down or reconciled. Also raise the grace-window poll interval from 50ms to 250ms now that each poll is an HTTP round trip rather than a local file read. Co-Authored-By: Claude Opus 4.8 --- bot_bottle/orchestrator/control_plane.py | 10 ++++--- bot_bottle/orchestrator/service.py | 21 ++++++++----- bot_bottle/queue_store.py | 17 +++++++++++ bot_bottle/supervise.py | 7 +++++ bot_bottle/supervise_server.py | 6 +++- tests/unit/test_orchestrator_control_plane.py | 8 ++--- tests/unit/test_orchestrator_service.py | 30 +++++++++++++++++-- tests/unit/test_supervise_server.py | 17 ++++++----- 8 files changed, 90 insertions(+), 26 deletions(-) diff --git a/bot_bottle/orchestrator/control_plane.py b/bot_bottle/orchestrator/control_plane.py index add9420..5760923 100644 --- a/bot_bottle/orchestrator/control_plane.py +++ b/bot_bottle/orchestrator/control_plane.py @@ -35,10 +35,12 @@ vsock / unix-socket portability caveats): "proposal_id"} The `/supervise/propose` + `/supervise/poll` pair is the **agent** half of the -supervise flow: the data plane (supervise / egress / git-gate) queues a -proposal and polls for its response over RPC instead of opening `bot-bottle.db` -directly. Both attribute the caller by `(source_ip, identity_token)` exactly -like `/resolve`, so a bottle can only ever queue or read its own proposals. +supervise flow: the data plane (supervise / egress / git-gate) queues a proposal +and polls for its response over RPC instead of opening `bot-bottle.db` directly. +`poll` is idempotent — it never archives, so a dropped connection can't lose an +operator decision (the row is reaped when the bottle is torn down / reconciled). +Both attribute the caller by `(source_ip, identity_token)` exactly like +`/resolve`, so a bottle can only ever queue or read its own proposals. `POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or tear down) the bottle in the registry AND broker the backend-native launch diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py index 1f042f3..629ab16 100644 --- a/bot_bottle/orchestrator/service.py +++ b/bot_bottle/orchestrator/service.py @@ -40,7 +40,7 @@ from ..supervise import ( STATUS_REJECTED, TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK, - archive_proposal, + archive_all_proposals, list_all_pending_proposals, read_proposal, read_response, @@ -136,6 +136,8 @@ class Orchestrator: self._broker.submit(sign_request(req, self._secret)) self.registry.deregister(bottle_id) self._tokens.pop(bottle_id, None) + # Reap any supervise proposals the gone bottle never acknowledged. + archive_all_proposals(bottle_id) return True def reconcile( @@ -160,6 +162,8 @@ class Orchestrator: live_source_ips, grace_seconds=grace_seconds) for rec in reaped: self._tokens.pop(rec.bottle_id, None) + # Reap any supervise proposals the gone bottle never acknowledged. + archive_all_proposals(rec.bottle_id) return [rec.bottle_id for rec in reaped] def tokens_for(self, bottle_id: str) -> dict[str, str]: @@ -223,15 +227,19 @@ class Orchestrator: return proposal.id def supervise_poll_response(self, bottle_id: str, proposal_id: str) -> dict[str, object]: - """Non-blocking poll of one of `bottle_id`'s queued proposals. + """Non-blocking, **idempotent** poll of one of `bottle_id`'s proposals. Returns `{"status": ...}`: a terminal decision (`approved`/`modified`/ `rejected`, with `notes` + `final_file`) once the operator has responded, `pending` while it's still queued undecided, or `unknown` when no such - queued proposal exists for this bottle (wrong id, or already resolved and - read). A decided proposal is archived here — the same terminal step the - data plane used to run after reading the response — so `pending` - proposals stay visible to the operator until decided *and* polled. + queued proposal exists for this bottle (wrong id, or already archived). + + Poll does **not** archive — a decided proposal keeps returning the same + decision so a client whose connection drops after the decision but before + it consumes the response still gets it on retry (issue #469 review). A + decided proposal is already excluded from the operator's pending list (a + response row exists), so it doesn't linger there; the row itself is reaped + when the bottle is torn down / reconciled. Reads are scoped to `bottle_id` (the queue key), so a caller can never read another bottle's proposal even with a guessed id.""" @@ -243,7 +251,6 @@ class Orchestrator: except FileNotFoundError: return {"status": POLL_STATUS_UNKNOWN} return {"status": POLL_STATUS_PENDING} - archive_proposal(bottle_id, proposal_id) return { "status": response.status, "notes": response.notes, diff --git a/bot_bottle/queue_store.py b/bot_bottle/queue_store.py index b547d90..2dae5c9 100644 --- a/bot_bottle/queue_store.py +++ b/bot_bottle/queue_store.py @@ -192,6 +192,23 @@ class QueueStore(DbStore): (self.queue_key, proposal_id), ) + def archive_all(self) -> None: + """Archive every proposal + response for this queue key. Used to reap a + bottle's supervise records when it is torn down or reconciled away — + the retention path for a client that received a decision but died before + acknowledging it. Idempotent.""" + if not self.db_path.is_file(): + return + with self._connection() as conn: + conn.execute( + "UPDATE supervise_proposals SET archived = 1 WHERE queue_key = ?", + (self.queue_key,), + ) + conn.execute( + "UPDATE supervise_responses SET archived = 1 WHERE queue_key = ?", + (self.queue_key,), + ) + @staticmethod def _row_to_proposal(row: sqlite3.Row) -> Proposal: return Proposal( diff --git a/bot_bottle/supervise.py b/bot_bottle/supervise.py index 827bd30..184500f 100644 --- a/bot_bottle/supervise.py +++ b/bot_bottle/supervise.py @@ -171,6 +171,12 @@ def archive_proposal(bottle_slug: str, proposal_id: str) -> None: QueueStore(bottle_slug).archive_proposal(proposal_id) +def archive_all_proposals(bottle_slug: str) -> None: + """Archive every proposal + response for `bottle_slug` (bottle teardown / + reconcile cleanup). Idempotent.""" + QueueStore(bottle_slug).archive_all() + + # --- Audit log ------------------------------------------------------------- @@ -275,6 +281,7 @@ __all__ = [ "TOOL_EGRESS_TOKEN_ALLOW", "TOOL_LIST_EGRESS_ROUTES", "archive_proposal", + "archive_all_proposals", "audit_dir", "audit_log_path", "list_pending_proposals", diff --git a/bot_bottle/supervise_server.py b/bot_bottle/supervise_server.py index 7a90d21..aad06d1 100644 --- a/bot_bottle/supervise_server.py +++ b/bot_bottle/supervise_server.py @@ -82,7 +82,11 @@ ERR_INVALID_PARAMS = -32602 ERR_INTERNAL = -32603 DEFAULT_RESPONSE_TIMEOUT_SECONDS = 30.0 -MIN_RESPONSE_POLL_INTERVAL_SECONDS = 0.05 +# Grace-window poll cadence. 250 ms keeps the operator-visible latency low while +# staying gentle on the shared control plane — each poll is now an HTTP round +# trip, not a local file read, so the old 50 ms would be ~20 req/s per pending +# proposal across every waiting agent (issue #469 review). +MIN_RESPONSE_POLL_INTERVAL_SECONDS = 0.25 # The per-host orchestrator control plane the shared supervise server attributes # each proposal to, by source IP. Mandatory — there is no single-tenant diff --git a/tests/unit/test_orchestrator_control_plane.py b/tests/unit/test_orchestrator_control_plane.py index a47b376..b400bc5 100644 --- a/tests/unit/test_orchestrator_control_plane.py +++ b/tests/unit/test_orchestrator_control_plane.py @@ -527,7 +527,7 @@ class TestDispatchSuperviseAgentRpc(unittest.TestCase): })) self.assertEqual(400, status) - def test_poll_pending_then_decided_then_archived(self) -> None: + def test_poll_pending_then_decided_then_idempotent(self) -> None: rec = self._register() _, proposed = self._propose(rec) pid = proposed["proposal_id"] @@ -546,12 +546,12 @@ class TestDispatchSuperviseAgentRpc(unittest.TestCase): self.assertEqual("approved", decided["status"]) self.assertEqual("ok", decided["notes"]) - # The decided poll archived it: gone from pending, and a re-poll is - # 'unknown' rather than replaying the decision forever. + # Poll doesn't archive: it's gone from the operator's pending list, but a + # re-poll returns the same decision so a dropped connection can't lose it. _, listing = dispatch(self.orch, "GET", "/supervise/proposals", b"") self.assertEqual([], listing["proposals"]) _, again = self._poll(rec, pid) - self.assertEqual("unknown", again["status"]) + self.assertEqual(decided, again) def test_poll_cannot_read_another_bottles_proposal(self) -> None: rec_a = self._register("10.0.0.1", "a") diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index cd50bc6..06cf611 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/tests/unit/test_orchestrator_service.py @@ -336,7 +336,7 @@ class TestOrchestratorSupervise(unittest.TestCase): self.assertEqual( {"status": "pending"}, self.orch.supervise_poll_response(bottle_id, pid)) - def test_poll_returns_decision_and_archives(self) -> None: + def test_poll_is_idempotent_and_leaves_no_pending(self) -> None: bottle_id = self._register("demo", "routes: []\n") pid = self.orch.supervise_queue_proposal( bottle_id, tool=TOOL_EGRESS_ALLOW, @@ -346,10 +346,34 @@ class TestOrchestratorSupervise(unittest.TestCase): decided = self.orch.supervise_poll_response(bottle_id, pid) self.assertEqual("approved", decided["status"]) self.assertEqual("ok", decided["notes"]) - # Archived on read → gone from pending, and a re-poll is 'unknown'. + # Decided proposal drops off the operator's pending list (a response row + # exists), but poll does NOT archive — a re-poll returns the same + # decision so a dropped connection can't lose it (issue #469 review). self.assertEqual([], self.orch.supervise_pending()) + self.assertEqual(decided, self.orch.supervise_poll_response(bottle_id, pid)) + + def test_teardown_reaps_the_bottles_proposals(self) -> None: + rec = self.store.register("10.243.0.20", policy="routes: []\n") + pid = self.orch.supervise_queue_proposal( + rec.bottle_id, tool=TOOL_EGRESS_ALLOW, + proposed_file="routes:\n - host: google.com\n", justification="j") + self.orch.supervise_respond( + pid, bottle_slug=rec.bottle_id, decision="approve", notes="ok") + self.assertTrue(self.orch.teardown_bottle(rec.bottle_id)) + # The gone bottle's decided-but-unconsumed proposal is archived, so a + # late poll returns 'unknown' rather than lingering forever. self.assertEqual( - {"status": "unknown"}, self.orch.supervise_poll_response(bottle_id, pid)) + {"status": "unknown"}, self.orch.supervise_poll_response(rec.bottle_id, pid)) + + def test_reconcile_reaps_the_bottles_proposals(self) -> None: + rec = self.store.register("10.243.0.21", policy="routes: []\n") + pid = self.orch.supervise_queue_proposal( + rec.bottle_id, tool=TOOL_EGRESS_ALLOW, + proposed_file="routes:\n - host: google.com\n", justification="j") + # No live source IPs -> the bottle is reaped (grace 0 so it's immediate). + self.assertEqual([rec.bottle_id], self.orch.reconcile([], grace_seconds=0)) + self.assertEqual( + {"status": "unknown"}, self.orch.supervise_poll_response(rec.bottle_id, pid)) def test_poll_unknown_for_other_bottle(self) -> None: bottle_id = self._register("demo", "routes: []\n") diff --git a/tests/unit/test_supervise_server.py b/tests/unit/test_supervise_server.py index a3f956f..c666c55 100644 --- a/tests/unit/test_supervise_server.py +++ b/tests/unit/test_supervise_server.py @@ -108,7 +108,7 @@ class _FakeSuperviseResolver: except FileNotFoundError: return {"status": _sv.POLL_STATUS_UNKNOWN} return {"status": _sv.POLL_STATUS_PENDING} - _sv.archive_proposal(self.bottle_id, proposal_id) + # Idempotent: poll never archives (issue #469 review). return { "status": response.status, "notes": response.notes, "final_file": response.final_file, @@ -473,7 +473,7 @@ class TestHandleToolsCall(unittest.TestCase): self.assertEqual(ERR_INVALID_PARAMS, cm.exception.code) self.assertIn("unknown tool", cm.exception.message) - def test_archives_proposal_after_response(self): + def test_decided_proposal_drops_off_pending(self): responder = self._respond_when_proposal_appears(_sv.STATUS_APPROVED) try: _tools_call(self.resolver, { @@ -485,7 +485,8 @@ class TestHandleToolsCall(unittest.TestCase): }) finally: responder.join() - # No pending proposals left after the decided poll archives it. + # A decided proposal drops off the operator's pending list (a response + # row exists) — poll itself no longer archives (issue #469 review). self.assertEqual([], _sv.list_pending_proposals("dev")) def test_pending_response_times_out_without_archive(self): @@ -776,7 +777,7 @@ class TestNonBlockingSupervise(unittest.TestCase): # --- check-proposal poll --- - def test_check_returns_approved_and_archives(self): + def test_check_returns_approved_idempotently(self): p = self._seed_proposal() _sv.write_response("dev", _sv.Response(proposal_id=p.id, status=_sv.STATUS_APPROVED, notes="ok")) result = _check(self.resolver, {"arguments": {"proposal_id": p.id}}) @@ -784,8 +785,10 @@ class TestNonBlockingSupervise(unittest.TestCase): 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) + # Poll doesn't archive, so a re-check returns the same decision + # (issue #469 review) rather than 'unknown'. + again = _check(self.resolver, {"arguments": {"proposal_id": p.id}}) + self.assertEqual(result, again) def test_check_rejected_sets_isError(self): p = self._seed_proposal() @@ -853,7 +856,7 @@ class TestNonBlockingSupervise(unittest.TestCase): poll = _check(self.resolver, {"arguments": {"proposal_id": 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 + self.assertEqual([], _sv.list_pending_proposals("dev")) # resolved (response exists) if __name__ == "__main__":