fix(scheduler): preserve FIFO behind active same-thread runs (#8231)

* fix(scheduler): preserve FIFO behind active same-thread runs

* test(scheduler): cover launching FIFO blockers

---------

Co-authored-by: mikamikasuki <211269698+mikamikasuki@users.noreply.github.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>

backend/packages/harness/deerflow/persistence/scheduled_task_runs/sql.py

@@ -480,10 +480,10 @@ class ScheduledTaskRunRepository:
     ):
         """SELECT for the next bounded drain batch, fair to task owners.
 
-        Only same-thread FIFO heads are eligible (an older active row on the
-        same thread hides every newer one). Rows at or before ``created_after``
-        are already due for timeout and are removed before FIFO checks, owner
-        ranking, and the batch limit; they must not occupy another row's slot.
+        Only same-thread FIFO heads are eligible. Older running or launching
+        rows on the same thread block newer rows regardless of age. Queued rows
+        at or before ``created_after`` are due for timeout and are excluded before
+        FIFO checks, owner ranking, and the batch limit.
         Eligible rows are ranked within their owner by
         ``attempt_count, created_at, id`` and the batch is ordered by that rank
         first, so every owner's head comes before any owner's second row: one
@@ -511,7 +511,10 @@ class ScheduledTaskRunRepository:
             ),
         ]
         if created_after is not None:
-            older_conditions.append(older.created_at > created_after)
+            # The timeout cutoff only removes expired queue entries. A running
+            # or launching occurrence remains a same-thread FIFO blocker no
+            # matter how long it has been active.
+            older_conditions.append(or_(older.status != "queued", older.created_at > created_after))
         older_same_thread = exists(select(older.id).where(*older_conditions))
         executing_run = aliased(ScheduledTaskRunRow)
         executing_task = aliased(ScheduledTaskRow)

backend/tests/test_scheduled_task_queue.py

@@ -1366,6 +1366,52 @@ async def test_fair_batch_keeps_same_thread_fifo(sqlite_repos, cap):
     assert batch == ["a-older"]
 
 
+@pytest.mark.parametrize("status", ["running", "launching"])
+async def test_queue_age_cutoff_keeps_active_same_thread_fifo_blocker(sqlite_repos, status):
+    sf, task_repo, run_repo = sqlite_repos
+    now = datetime.now(UTC)
+    await _owned_occurrence(
+        task_repo,
+        run_repo,
+        "a-old-active",
+        user_id=OWNER_A,
+        status=status,
+        thread_id="reuse-thread",
+    )
+    await _owned_occurrence(
+        task_repo,
+        run_repo,
+        "a-blocked-queued",
+        user_id=OWNER_A,
+        thread_id="reuse-thread",
+    )
+    await _owned_occurrence(
+        task_repo,
+        run_repo,
+        "a-healthy-queued",
+        user_id=OWNER_A,
+        thread_id="other-thread",
+    )
+
+    # The timeout applies to waiting rows, not to an older active execution.
+    async with sf() as session:
+        blocker = await session.get(ScheduledTaskRunRow, "a-old-active")
+        assert blocker is not None
+        blocker.created_at = now - timedelta(hours=1)
+        await session.commit()
+
+    cutoff = now - timedelta(minutes=5)
+    batch = await run_repo.list_queued_runs(
+        limit=16,
+        per_user_max_concurrent_runs=2,
+        created_after=cutoff,
+    )
+
+    # The active run still blocks its thread. Its newer queued row must not
+    # consume the owner's one remaining slot ahead of unrelated healthy work.
+    assert [row["id"] for row in batch] == ["a-healthy-queued"]
+
+
 async def test_orphan_queued_row_is_still_drained(sqlite_repos):
     _sf, task_repo, run_repo = sqlite_repos
     await _owned_occurrence(task_repo, run_repo, "a-running", user_id=OWNER_A, status="running")