From 74e706d427eb4c963729ce415beeee484944cbb4 Mon Sep 17 00:00:00 2001
From: Lucky Reddy <lkanthareddya@gmail.com>
Date: Sat, 26 Sep 2026 15:26:23 -0500
Subject: [PATCH] fix: let queued tasks start while a long task is running

TaskWorker.tick() awaited every run it claimed before releasing its lock,
so one slow task held up all queued work, scheduled watch checks and
approved tasks until it finished. Start the claimed runs and release the
tick right away, tracking started tasks in an inFlight set so a worker
still runs at most three at once and never starts the same task twice.
---
 apps/server/src/engine/worker.ts | 16 ++++++++++++----
 tests/engine.test.ts             | 31 +++++++++++++++++++++++++++++++
 2 files changed, 43 insertions(+), 4 deletions(-)

diff --git a/apps/server/src/engine/worker.ts b/apps/server/src/engine/worker.ts
index eeadea40c..b69fe6348 100644
--- a/apps/server/src/engine/worker.ts
+++ b/apps/server/src/engine/worker.ts
@@ -25,6 +25,8 @@ export class TaskWorker {
   private ticking = false;
   private stopping = false;
   private active = new Map<string, AbortController>();
+  // Tasks this worker has started and not yet settled, including ones still being claimed.
+  private inFlight = new Set<string>();
   lastTickAt?: string;
   constructor(
     private readonly db: Store,
@@ -56,7 +58,7 @@ export class TaskWorker {
     if (this.timer) clearInterval(this.timer);
     this.timer = undefined;
     for (const controller of this.active.values()) controller.abort();
-    while (this.active.size || this.ticking) await new Promise((r) => setTimeout(r, 10));
+    while (this.inFlight.size || this.ticking) await new Promise((r) => setTimeout(r, 10));
   }
   abort(taskId: string) {
     this.active.get(taskId)?.abort();
@@ -71,11 +73,12 @@ export class TaskWorker {
     if (this.ticking) return;
     this.ticking = true;
     this.lastTickAt = new Date(this.now()).toISOString();
+    const started: Promise<void>[] = [];
     try {
       const records = await this.db.scan<AgentTask>("tasks");
       const due = records.filter(
         ({ value: t }) =>
-          !this.active.has(t.id) &&
+          !this.inFlight.has(t.id) &&
           (t.status === "queued" ||
             (t.status === "scheduled" && Date.parse(t.nextRunAt ?? "") <= this.now()) ||
             (t.status === "running" && Date.parse(t.leaseUntil ?? "") <= this.now()) ||
@@ -83,6 +86,7 @@ export class TaskWorker {
       );
       const eligible = [];
       for (const record of due) {
+        if (this.inFlight.size + eligible.length >= 3) break;
         if (record.value.status === "waiting_approval") {
           const action = record.value.actionId
             ? await this.db.get<{ status: string; expiresAt?: string }>(
@@ -106,12 +110,16 @@ export class TaskWorker {
           else if (action && ["awaiting_review", "executing"].includes(action.status)) continue;
         }
         eligible.push(record);
-        if (eligible.length === 3) break;
       }
-      await Promise.all(eligible.map(({ owner, value }) => this.run(owner, value)));
+      // Runs outlive the tick, so a long task does not stop later ticks from starting queued work.
+      for (const { owner, value } of eligible) {
+        this.inFlight.add(value.id);
+        started.push(this.run(owner, value).finally(() => this.inFlight.delete(value.id)));
+      }
     } finally {
       this.ticking = false;
     }
+    await Promise.all(started);
   }
   private async run(owner: string, previous: AgentTask) {
     if (this.stopping) return;
diff --git a/tests/engine.test.ts b/tests/engine.test.ts
index 947b1007c..7b23ab06d 100644
--- a/tests/engine.test.ts
+++ b/tests/engine.test.ts
@@ -168,6 +168,37 @@ test("pending reviews do not starve queued work", async () => {
     await db.close();
   }
 });
+test("a long-running task does not starve queued work", async () => {
+  const db = await createStore();
+  let finish: () => void = () => {};
+  const finished = new Promise<void>((resolve) => {
+    finish = resolve;
+  });
+  let started: () => void = () => {};
+  const running = new Promise<void>((resolve) => {
+    started = resolve;
+  });
+  const worker = new TaskWorker(db, async (_owner, current) => {
+    if (current.id === "long") {
+      started();
+      await finished;
+    }
+    return { status: "succeeded" };
+  });
+  await db.put("owner", "tasks", task("long"));
+  const first = worker.tick();
+  try {
+    await running;
+    await db.put("owner", "tasks", task("ready"));
+    await worker.tick();
+    assert.equal((await db.get<AgentTask>("owner", "tasks", "ready"))?.status, "succeeded");
+    assert.equal((await db.get<AgentTask>("owner", "tasks", "long"))?.status, "running");
+  } finally {
+    finish();
+    await first;
+    await db.close();
+  }
+});
 test("run history keeps the time the run started", async () => {
   const db = await createStore();
   try {