From ecd1bc2e336d4f9988dfd5a9d1ce11455d70c76e Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?=E9=9B=B7=E7=94=B5=E8=8A=BD=E8=A1=A3?=
 <zhangyanghaha0407@outlook.com>
Date: Fri, 17 Apr 2026 15:07:27 -0400
Subject: [PATCH] refactor(worker): single active-run map replaces parallel
 claim structures
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Worker scheduling invariants ("one agent at a time", "one exclusive
workspace lock at a time") were enforced across three parallel
structures: `activeByAgent: Map<string, string>`,
`activeByWorkspace: Map<string, string>`, and `queue: RunJob[]`. Every
claim, release, and tick had to update them in lockstep, with defensive
comments explaining the ordering to future readers.

Replace with one `active: Map<runId, ActiveRun>` where ActiveRun
carries agentId + workspaceKey. Busy checks iterate the map (≤ dozens
of entries); dispatch is a single set + splice. Release is a single
delete. No parallel structures to keep in sync.

- Add `isAgentBusy(agentId)` and `isWorkspaceBusy(lockKey)` as local
  helpers — both O(n) over the active map, which is fine at this scale.
- `tickQueue` is claim-before-dispatch: writes `active.set` synchronously
  before `startRun` so no async window exists between claim and
  dispatch.
- Drop the `lockKey` parameter threading through `startRun` — the map
  carries it.
---
 src/main/worker/worker-entry.ts | 61 ++++++++++++++++++++++-----------
 1 file changed, 41 insertions(+), 20 deletions(-)

diff --git a/src/main/worker/worker-entry.ts b/src/main/worker/worker-entry.ts
index d234856b..057a4bcf 100644
--- a/src/main/worker/worker-entry.ts
+++ b/src/main/worker/worker-entry.ts
@@ -8,8 +8,20 @@ if (!parentPort) {
   throw new Error("AgentCompany worker requires process.parentPort");
 }
 
-const activeByAgent = new Map<string, string>();
-const activeByWorkspace = new Map<string, string>();
+interface ActiveRun {
+  runId: string;
+  agentId: string;
+  workspaceKey: string | null;
+}
+
+/**
+ * Single source of truth for scheduling invariants: "one agent at a time"
+ * and "one exclusive-workspace lock at a time". Prior layout kept three
+ * parallel structures (`activeByAgent`, `activeByWorkspace`, queue) that
+ * had to be updated in lockstep. The Map-of-ActiveRun keyed by runId is
+ * the only authoritative record; agent/workspace busy checks iterate it.
+ */
+const active = new Map<string, ActiveRun>();
 const queue: RunJob[] = [];
 const cancelledRuns = new Set<string>();
 
@@ -24,11 +36,25 @@ const runner = new WorkerRunner({
   isCancelled: (runId) => cancelledRuns.has(runId),
 });
 
-function workspaceKey(job: RunJob) {
+function workspaceKey(job: RunJob): string | null {
   if (job.workspaceMode !== "exclusive") return null;
   return job.workspaceId ?? job.workspacePath ?? `task:${job.taskId}`;
 }
 
+function isAgentBusy(agentId: string): boolean {
+  for (const run of active.values()) {
+    if (run.agentId === agentId) return true;
+  }
+  return false;
+}
+
+function isWorkspaceBusy(lockKey: string): boolean {
+  for (const run of active.values()) {
+    if (run.workspaceKey === lockKey) return true;
+  }
+  return false;
+}
+
 function enqueue(job: RunJob) {
   queue.push(job);
   post({
@@ -37,39 +63,34 @@ function enqueue(job: RunJob) {
     status: "queued",
     message: "Run queued in orchestration worker.",
   });
-  maybeStartQueuedRuns();
+  tickQueue();
 }
 
 /**
- * Walks the queue in-place and dispatches every job whose agent + workspace
- * are not currently busy. Claims (activeByAgent / activeByWorkspace) are
- * written BEFORE the job is removed from the queue and BEFORE `startRun`
- * fires — this is the atomic point of commit for orchestrator ordering.
- * Hardens against a future refactor that adds an `await` before the claim
- * inside `startRun`, which would open a re-entrancy window.
+ * Claim-before-dispatch: for each queued job whose agent + workspace are
+ * free, register the claim in `active` synchronously *before* calling
+ * `startRun`. No async window between claim and dispatch.
  */
-function maybeStartQueuedRuns() {
+function tickQueue() {
   let i = 0;
   while (i < queue.length) {
     const job = queue[i];
-    if (activeByAgent.has(job.agentId)) { i += 1; continue; }
+    if (isAgentBusy(job.agentId)) { i += 1; continue; }
     const lockKey = workspaceKey(job);
-    if (lockKey && activeByWorkspace.has(lockKey)) { i += 1; continue; }
-    activeByAgent.set(job.agentId, job.runId);
-    if (lockKey) activeByWorkspace.set(lockKey, job.runId);
+    if (lockKey && isWorkspaceBusy(lockKey)) { i += 1; continue; }
+    active.set(job.runId, { runId: job.runId, agentId: job.agentId, workspaceKey: lockKey });
     queue.splice(i, 1);
-    void startRun(job, lockKey);
+    void startRun(job);
   }
 }
 
-async function startRun(job: RunJob, lockKey: string | null) {
+async function startRun(job: RunJob) {
   try {
     await runner.run(job);
   } finally {
     cancelledRuns.delete(job.runId);
-    activeByAgent.delete(job.agentId);
-    if (lockKey) activeByWorkspace.delete(lockKey);
-    maybeStartQueuedRuns();
+    active.delete(job.runId);
+    tickQueue();
   }
 }