fix(conductor): make batch cancellation authoritative (#726) (#798)

cancel_batch/stop_batch wrote status=CANCELLED, but a still-running worker's
whole-row 'INSERT OR REPLACE ... batch.status' (RUNNING) overwrote it; the loop
re-read RUNNING and kept launching tasks, spending tokens.

_save_batch now refuses to resurrect a persisted CANCELLED batch to a live
status (preserve_terminal_cancel, default on): it leaves the terminal record
untouched and flips the in-memory batch.status to CANCELLED so the execution
loop's existing get_batch check breaks. resume_batch passes
preserve_terminal_cancel=False for its intentional CANCELLED->RUNNING re-run.

Closes #726

Author: frankbria

codeframe/core/conductor.py

@@ -992,10 +992,11 @@ def resume_batch(
     if on_event:
         on_event("batch_resumed", {"batch_id": batch_id, "task_count": len(tasks_to_run)})
 
-    # Update status to running
+    # Update status to running. This is the one intentional CANCELLED->RUNNING
+    # transition, so bypass the terminal-cancel guard (#726).
     batch.status = BatchStatus.RUNNING
     batch.completed_at = None  # Clear completed_at since we're running again
-    _save_batch(workspace, batch)
+    _save_batch(workspace, batch, preserve_terminal_cancel=False)
 
     # Execute the tasks
     _execute_serial_resume(workspace, batch, tasks_to_run, on_event)
@@ -2185,11 +2186,22 @@ def _execute_task_subprocess(
         return RunStatus.FAILED.value
 
 
-def _save_batch(workspace: Workspace, batch: BatchRun) -> None:
+def _save_batch(
+    workspace: Workspace,
+    batch: BatchRun,
+    preserve_terminal_cancel: bool = True,
+) -> None:
     """Save or update a batch record in the database.
 
     Uses a thread lock to prevent SQLite "database is locked" errors
     when multiple workers try to update batch status concurrently.
+
+    When ``preserve_terminal_cancel`` is True (the default), a whole-row write
+    that would resurrect a persisted CANCELLED batch back to a live status is
+    refused (#726): a task-completion save from a still-running worker must not
+    undo a concurrent cancel. The in-memory ``batch.status`` is also flipped to
+    CANCELLED so the execution loop stops launching tasks. ``resume_batch``
+    passes ``preserve_terminal_cancel=False`` to intentionally re-run.
     """
     task_ids_json = json.dumps(batch.task_ids)
     results_json = json.dumps(batch.results) if batch.results else None
@@ -2208,6 +2220,20 @@ def _save_batch(workspace: Workspace, batch: BatchRun) -> None:
             except Exception:
                 pass  # Column already exists
 
+            # Cancellation is authoritative (#726): don't let a whole-row write
+            # from a still-running worker resurrect a batch a concurrent
+            # cancel_batch/stop_batch already marked CANCELLED.
+            if preserve_terminal_cancel and batch.status != BatchStatus.CANCELLED:
+                cursor.execute(
+                    "SELECT status FROM batch_runs WHERE id = ?", (batch.id,)
+                )
+                existing = cursor.fetchone()
+                if existing and existing[0] == BatchStatus.CANCELLED.value:
+                    # Reflect the cancel into the in-memory batch so the loop
+                    # stops, and leave the persisted terminal record untouched.
+                    batch.status = BatchStatus.CANCELLED
+                    return
+
             cursor.execute(
                 """
                 INSERT OR REPLACE INTO batch_runs

tests/core/test_conductor.py

@@ -305,6 +305,57 @@ def test_cancel_pending_batch(self, workspace_with_tasks):
         assert cancelled.completed_at is not None
 
 
+class TestCancellationIsAuthoritative:
+    """#726 / P0.15: a concurrent cancel must not be resurrected by a
+    still-running worker's whole-row save."""
+
+    def _running_batch(self, workspace, task_ids):
+        from datetime import datetime, timezone
+
+        batch = BatchRun(
+            id="test-cancel-race",
+            workspace_id=workspace.id,
+            task_ids=task_ids,
+            status=BatchStatus.RUNNING,
+            strategy="serial",
+            max_parallel=4,
+            on_failure=OnFailure.CONTINUE,
+            started_at=datetime.now(timezone.utc),
+            completed_at=None,
+            results={},
+        )
+        _save_batch(workspace, batch)
+        return batch
+
+    def test_worker_save_does_not_resurrect_cancelled(self, workspace_with_tasks):
+        workspace, task_list = workspace_with_tasks
+        worker_view = self._running_batch(workspace, [task_list[0].id])
+
+        # A concurrent cancel marks the persisted batch CANCELLED.
+        cancel_batch(workspace, worker_view.id)
+
+        # The still-running worker finishes a task and saves the whole row with
+        # its stale RUNNING in-memory status.
+        worker_view.status = BatchStatus.RUNNING
+        worker_view.results = {task_list[0].id: "COMPLETED"}
+        _save_batch(workspace, worker_view)
+
+        # Persisted status stays CANCELLED (not resurrected), and the in-memory
+        # view is flipped so the execution loop will stop.
+        assert get_batch(workspace, worker_view.id).status == BatchStatus.CANCELLED
+        assert worker_view.status == BatchStatus.CANCELLED
+
+    def test_resume_can_still_transition_cancelled_to_running(self, workspace_with_tasks):
+        workspace, task_list = workspace_with_tasks
+        batch = self._running_batch(workspace, [task_list[0].id])
+        cancel_batch(workspace, batch.id)
+
+        # resume_batch bypasses the guard (preserve_terminal_cancel=False).
+        batch.status = BatchStatus.RUNNING
+        _save_batch(workspace, batch, preserve_terminal_cancel=False)
+        assert get_batch(workspace, batch.id).status == BatchStatus.RUNNING
+
+
 class TestStopBatch:
     """Tests for stop_batch function."""