From 6e3ebf36538505931160f366a01277a0fc6cff65 Mon Sep 17 00:00:00 2001 From: cawIgor Date: Sat, 24 Oct 2026 15:19:11 +0000 Subject: [PATCH 1/2] refactor: enforce durable request transitions --- bot/telegram/bloc_bridge/request_store.py | 78 +++++++++++++++++++++++ tests/test_request_lifecycle.py | 48 ++++++++++++++ 2 files changed, 126 insertions(+) create mode 100644 tests/test_request_lifecycle.py diff --git a/bot/telegram/bloc_bridge/request_store.py b/bot/telegram/bloc_bridge/request_store.py index 600ecd7..266e9d6 100644 --- a/bot/telegram/bloc_bridge/request_store.py +++ b/bot/telegram/bloc_bridge/request_store.py @@ -34,6 +34,27 @@ } ) +ALLOWED_TRANSITIONS = { + "acknowledging": frozenset(("acknowledging", "queued", "failed", "acknowledgement_unknown")), + "queued": frozenset(("queued", "submitting", "failed", "interrupted", "rejected")), + "submitting": frozenset(("submitting", "running", "uncertain", "failed")), + "running": frozenset(("running", "cancellation_requested", "delivering", "uncertain")), + "cancellation_requested": frozenset( + ("cancellation_requested", "running", "cancelled", "timed_out", "uncertain") + ), + "recovering": frozenset(("recovering", "running", "uncertain", "failed")), + "uncertain": frozenset(("uncertain", "recovering", "running", "failed", "cancelled", "timed_out")), + "delivering": frozenset( + ("delivering", "completed", "failed", "cancelled", "timed_out", "delivery_unknown") + ), +} +for _terminal_state in TERMINAL_STATES: + ALLOWED_TRANSITIONS[_terminal_state] = frozenset((_terminal_state,)) + + +class InvalidTransition(RuntimeError): + """Raised when a request attempts a state transition outside its lifecycle.""" + def _connect(): directory = os.path.dirname(REQUESTS_PATH) @@ -90,6 +111,8 @@ def initialize(): def create(reference, chat_id, update_id, state="acknowledging", detail=None): """Create one request for a Telegram update; return its row and whether it was new.""" + if state not in ALLOWED_TRANSITIONS: + raise InvalidTransition(f"unknown initial state {state}") now = time.time() with _connection() as connection: # Telegram may redeliver an update until the next poll confirms its offset. @@ -157,6 +180,10 @@ def update(reference, **fields): if not updates: return get(reference) + requested_state = updates.get("state") + if requested_state is not None: + return transition(reference, requested_state, **{key: value for key, value in updates.items() if key != "state"}) + updates["updated_at"] = time.time() assignments = ", ".join(f"{key} = ?" for key in updates) values = list(updates.values()) + [reference] @@ -170,6 +197,40 @@ def update(reference, **fields): return dict(row) if row is not None else None +def transition(reference, new_state, **fields): + """Atomically validate and apply one request lifecycle transition.""" + allowed = { + "acknowledgement_id", + "session_id", + "prompt_message_id", + "detail", + "started_at", + "completed_at", + } + updates = {key: value for key, value in fields.items() if key in allowed} + updates["state"] = new_state + updates["updated_at"] = time.time() + assignments = ", ".join(f"{key} = ?" for key in updates) + with _connection() as connection: + row = connection.execute( + "SELECT state FROM requests WHERE reference = ?", (reference,) + ).fetchone() + if row is None: + return None + current_state = row["state"] + if new_state not in ALLOWED_TRANSITIONS.get(current_state, ()): + raise InvalidTransition(f"cannot transition from {current_state} to {new_state}") + values = list(updates.values()) + [reference, current_state] + connection.execute( + f"UPDATE requests SET {assignments} WHERE reference = ? AND state = ?", + values, + ) + row = connection.execute( + "SELECT * FROM requests WHERE reference = ?", (reference,) + ).fetchone() + return dict(row) if row is not None else None + + def update_if_state(reference, expected_states, **fields): """Atomically update a request only if it is still in an expected state.""" allowed = { @@ -184,12 +245,29 @@ def update_if_state(reference, expected_states, **fields): updates = {key: value for key, value in fields.items() if key in allowed} if not updates: return get(reference) + requested_state = updates.get("state") + updates.pop("state", None) updates["updated_at"] = time.time() assignments = ", ".join(f"{key} = ?" for key in updates) expected = tuple(expected_states) placeholders = ", ".join("?" for _ in expected) values = list(updates.values()) + [reference, *expected] with _connection() as connection: + if requested_state is not None: + current = connection.execute( + "SELECT state FROM requests WHERE reference = ? AND state IN ({} )".format(placeholders), + (reference, *expected), + ).fetchone() + if current is None: + row = connection.execute( + "SELECT * FROM requests WHERE reference = ?", (reference,) + ).fetchone() + return dict(row) if row is not None else None + if requested_state not in ALLOWED_TRANSITIONS.get(current["state"], ()): + raise InvalidTransition(f"cannot transition from {current['state']} to {requested_state}") + updates["state"] = requested_state + assignments = ", ".join(f"{key} = ?" for key in updates) + values = list(updates.values()) + [reference, *expected] connection.execute( f"UPDATE requests SET {assignments} WHERE reference = ? AND state IN ({placeholders})", values, diff --git a/tests/test_request_lifecycle.py b/tests/test_request_lifecycle.py new file mode 100644 index 0000000..4890a76 --- /dev/null +++ b/tests/test_request_lifecycle.py @@ -0,0 +1,48 @@ +import os +import sys +import tempfile +import unittest +from unittest.mock import patch + + +sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "bot", "telegram")) +with patch.dict(os.environ, {"TELEGRAM_BOT_TOKEN": "test-token", "TELEGRAM_ALLOWED_CHAT_IDS": "123"}): + from bloc_bridge import request_store + + +class RequestLifecycleTests(unittest.TestCase): + def setUp(self): + self.directory = tempfile.TemporaryDirectory() + self.addCleanup(self.directory.cleanup) + self.path = patch.object(request_store, "REQUESTS_PATH", os.path.join(self.directory.name, "requests.sqlite3")) + self.path.start() + self.addCleanup(self.path.stop) + request_store.initialize() + + def test_valid_transitions_are_applied(self): + request_store.create("A" * 16, 123, 1) + for state in ("queued", "submitting", "running", "delivering", "completed"): + request_store.transition("A" * 16, state) + self.assertEqual(request_store.get("A" * 16)["state"], "completed") + + def test_invalid_transition_is_rejected_without_mutating_state(self): + request_store.create("B" * 16, 123, 2) + with self.assertRaises(request_store.InvalidTransition): + request_store.transition("B" * 16, "completed") + self.assertEqual(request_store.get("B" * 16)["state"], "acknowledging") + + def test_terminal_state_cannot_be_reopened(self): + request_store.create("C" * 16, 123, 3) + request_store.transition("C" * 16, "queued") + request_store.transition("C" * 16, "submitting") + request_store.transition("C" * 16, "running") + request_store.transition("C" * 16, "delivering") + request_store.transition("C" * 16, "failed") + with self.assertRaises(request_store.InvalidTransition): + request_store.transition("C" * 16, "running") + + def test_legacy_update_with_state_uses_the_same_validation(self): + request_store.create("D" * 16, 123, 4) + request_store.update("D" * 16, state="queued") + with self.assertRaises(request_store.InvalidTransition): + request_store.update("D" * 16, state="completed") From ac65086c656261c8f5778c692126cd3e9724d7d2 Mon Sep 17 00:00:00 2001 From: cawIgor Date: Sat, 24 Oct 2026 15:23:13 +0000 Subject: [PATCH 2/2] fix: allow existing request lifecycle paths --- bot/telegram/bloc_bridge/request_store.py | 8 ++++---- tests/test_request_lifecycle.py | 13 +++++++++++++ 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/bot/telegram/bloc_bridge/request_store.py b/bot/telegram/bloc_bridge/request_store.py index 266e9d6..cef60ee 100644 --- a/bot/telegram/bloc_bridge/request_store.py +++ b/bot/telegram/bloc_bridge/request_store.py @@ -36,11 +36,11 @@ ALLOWED_TRANSITIONS = { "acknowledging": frozenset(("acknowledging", "queued", "failed", "acknowledgement_unknown")), - "queued": frozenset(("queued", "submitting", "failed", "interrupted", "rejected")), - "submitting": frozenset(("submitting", "running", "uncertain", "failed")), - "running": frozenset(("running", "cancellation_requested", "delivering", "uncertain")), + "queued": frozenset(("queued", "submitting", "delivering", "failed", "interrupted", "rejected")), + "submitting": frozenset(("submitting", "running", "delivering", "uncertain", "failed")), + "running": frozenset(("running", "cancellation_requested", "recovering", "delivering", "uncertain", "failed")), "cancellation_requested": frozenset( - ("cancellation_requested", "running", "cancelled", "timed_out", "uncertain") + ("cancellation_requested", "running", "delivering", "cancelled", "timed_out", "uncertain") ), "recovering": frozenset(("recovering", "running", "uncertain", "failed")), "uncertain": frozenset(("uncertain", "recovering", "running", "failed", "cancelled", "timed_out")), diff --git a/tests/test_request_lifecycle.py b/tests/test_request_lifecycle.py index 4890a76..67215d2 100644 --- a/tests/test_request_lifecycle.py +++ b/tests/test_request_lifecycle.py @@ -46,3 +46,16 @@ def test_legacy_update_with_state_uses_the_same_validation(self): request_store.update("D" * 16, state="queued") with self.assertRaises(request_store.InvalidTransition): request_store.update("D" * 16, state="completed") + + def test_existing_command_and_recovery_paths_are_valid(self): + request_store.create("E" * 16, 123, 5) + request_store.transition("E" * 16, "queued") + request_store.transition("E" * 16, "delivering") + request_store.transition("E" * 16, "completed") + + request_store.create("F" * 16, 123, 6) + request_store.transition("F" * 16, "queued") + request_store.transition("F" * 16, "submitting") + request_store.transition("F" * 16, "running") + request_store.transition("F" * 16, "recovering") + request_store.transition("F" * 16, "failed")