refactor: enforce request lifecycle transitions

bot/telegram/bloc_bridge/request_store.py

@@ -34,6 +34,27 @@
     }
 )
 
+ALLOWED_TRANSITIONS = {
+    "acknowledging": frozenset(("acknowledging", "queued", "failed", "acknowledgement_unknown")),
+    "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", "delivering", "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,

tests/test_request_lifecycle.py

@@ -0,0 +1,61 @@
+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")
+
+    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")