feat: multi-agent orchestration with locking and streaming output

.gitignore

@@ -3,4 +3,5 @@ node_modules/
 .claude/worktrees/
 crewbit
 docs/.vitepress/dist
-docs/.vitepress/cache
+docs/.vitepress/cache
+.crewbit

src/commands/daemon.ts

@@ -1,5 +1,8 @@
+import { join } from "node:path";
+import { LOCK_DIR, acquireLock, getLockedKeys, releaseLock } from "../lock.js";
 import { createRunner } from "../runner/index.js";
-import { createProvider, loadConfig, resolveNextAction } from "../workflow.js";
+import type { QueueAction, WorkflowConfig } from "../types.js";
+import { createProvider, loadConfig, resolveNextActions } from "../workflow.js";
 
 const REPO_ROOT = process.cwd();
 
@@ -15,50 +18,151 @@ async function sleep(seconds: number): Promise<void> {
   return new Promise((resolvePromise) => setTimeout(resolvePromise, seconds * 1000));
 }
 
+type ActiveSession = {
+  issueKey: string;
+  promise: Promise<boolean>;
+  done: boolean;
+  result: boolean | null;
+};
+
+async function runSession(
+  action: QueueAction & { type: "run" },
+  config: WorkflowConfig,
+  dryRun: boolean,
+  lockDir: string,
+): Promise<boolean> {
+  const prefixedLog = (message: string) => log(`[${action.issueKey}] ${message}`);
+  const runner = createRunner(config, REPO_ROOT, prefixedLog);
+  try {
+    return await runner.run(action, config, dryRun);
+  } finally {
+    releaseLock(lockDir, action.issueKey);
+  }
+}
+
+function reapFinishedSessions(sessions: Map<string, ActiveSession>): boolean {
+  let hadFailure = false;
+  for (const [issueKey, session] of sessions.entries()) {
+    if (!session.done) continue;
+    if (session.result) {
+      log(`[${issueKey}] Session succeeded`);
+    } else {
+      log(`[${issueKey}] Session failed`);
+      hadFailure = true;
+    }
+    sessions.delete(issueKey);
+  }
+  return hadFailure;
+}
+
+async function startNewSessions(
+  config: WorkflowConfig,
+  sessions: Map<string, ActiveSession>,
+  lockDir: string,
+  dryRun: boolean,
+  maxConcurrent: number,
+): Promise<number> {
+  const slotsAvailable = Math.max(0, maxConcurrent - sessions.size);
+  if (slotsAvailable === 0) return 0;
+
+  const lockedKeys = new Set(getLockedKeys(lockDir));
+  for (const issueKey of sessions.keys()) {
+    lockedKeys.add(issueKey);
+  }
+
+  const provider = createProvider(config);
+  const actions = await resolveNextActions(config, provider, slotsAvailable, lockedKeys);
+  let startedCount = 0;
+
+  for (const action of actions) {
+    if (!acquireLock(lockDir, action.issueKey)) continue;
+
+    const tracked: ActiveSession = {
+      issueKey: action.issueKey,
+      promise: Promise.resolve(false),
+      done: false,
+      result: null,
+    };
+
+    tracked.promise = runSession(action, config, dryRun, lockDir)
+      .then((ok) => {
+        tracked.done = true;
+        tracked.result = ok;
+        return ok;
+      })
+      .catch((error) => {
+        tracked.done = true;
+        tracked.result = false;
+        log(
+          `[${action.issueKey}] Session crashed: ${error instanceof Error ? error.message : String(error)}`,
+        );
+        return false;
+      });
+
+    sessions.set(action.issueKey, tracked);
+    startedCount += 1;
+  }
+
+  return startedCount;
+}
+
 export async function runDaemonCommand(args: {
   configPath: string;
   dryRun: boolean;
 }): Promise<void> {
   const { configPath, dryRun } = args;
-  let exp = 1;
+  let emptyBackoffExp = 1;
+  let failureBackoffExp = 1;
+  const lockDir = join(REPO_ROOT, LOCK_DIR);
+  const sessions = new Map<string, ActiveSession>();
 
   log(`crewbit starting${dryRun ? " (dry-run)" : ""}`);
   log(`Config: ${configPath}`);
 
-  process.on("SIGINT", () => {
-    log("Stopped.");
-    process.exit(0);
-  });
-  process.on("SIGTERM", () => {
-    log("Stopped.");
-    process.exit(0);
-  });
+  const shutdown = () => {
+    log("Stopping... waiting for in-flight sessions to finish.");
+    Promise.allSettled([...sessions.values()].map((s) => s.promise)).then(() => {
+      log("Stopped.");
+      process.exit(0);
+    });
+  };
+  process.on("SIGINT", shutdown);
+  process.on("SIGTERM", shutdown);
 
   while (true) {
     try {
-      exp = Math.min(exp, 10);
-
       const config = loadConfig(configPath);
+      const maxConcurrent = config.daemon?.maxConcurrent ?? 1;
       const waitSeconds = Number(process.env.WAIT_SECONDS ?? config.daemon?.waitSeconds ?? 60);
+      const pollSeconds = 2;
+
+      const hadFailure = reapFinishedSessions(sessions);
+      const startedCount = await startNewSessions(config, sessions, lockDir, dryRun, maxConcurrent);
 
-      const provider = createProvider(config);
-      const action = await resolveNextAction(config, provider);
+      if (startedCount > 0) {
+        emptyBackoffExp = 1;
+        failureBackoffExp = 1;
+        await sleep(pollSeconds);
+        continue;
+      }
+
+      if (sessions.size > 0) {
+        await sleep(pollSeconds);
+        continue;
+      }
 
-      if (action.type === "idle") {
-        log(`Queue empty. Next check in ${waitSeconds * exp}s. (Ctrl+C to stop)`);
-        await sleep(waitSeconds * exp);
-        exp *= 2;
+      if (hadFailure) {
+        const backoff = waitSeconds * failureBackoffExp;
+        log(`Session failed. Backing off ${backoff}s before retry.`);
+        await sleep(backoff);
+        failureBackoffExp = Math.min(failureBackoffExp * 2, 32);
+        emptyBackoffExp = 1;
       } else {
-        const runner = createRunner(config, REPO_ROOT, log);
-        const ok = await runner.run(action, config, dryRun);
-        if (ok) {
-          exp = 1;
-        } else {
-          const backoff = waitSeconds * exp;
-          log(`Session failed. Backing off ${backoff}s before retry.`);
-          await sleep(backoff);
-          exp = Math.min(exp * 2, 32);
-        }
+        emptyBackoffExp = Math.min(emptyBackoffExp, 10);
+        log(`Queue empty. Next check in ${waitSeconds * emptyBackoffExp}s. (Ctrl+C to stop)`);
+        await sleep(waitSeconds * emptyBackoffExp);
+        emptyBackoffExp *= 2;
+        failureBackoffExp = 1;
       }
     } catch (error) {
       log(`Error: ${error instanceof Error ? error.message : String(error)}`);

src/lock.test.ts

@@ -0,0 +1,91 @@
+import assert from "node:assert/strict";
+import { existsSync, readFileSync, rmSync } from "node:fs";
+import { join } from "node:path";
+import { beforeEach, describe, it } from "node:test";
+import { acquireLock, getLockedKeys, isLocked, releaseLock } from "./lock.js";
+
+const TEST_LOCKS_DIR = join(process.cwd(), ".crewbit/test.locks");
+
+function cleanTestDir() {
+  if (existsSync(TEST_LOCKS_DIR)) {
+    rmSync(TEST_LOCKS_DIR, { recursive: true, force: true });
+  }
+}
+
+beforeEach(() => {
+  cleanTestDir();
+});
+
+describe("acquireLock", () => {
+  it("creates a lock file and returns true", () => {
+    const result = acquireLock(TEST_LOCKS_DIR, "JIR-1");
+    assert.equal(result, true);
+    assert.equal(isLocked(TEST_LOCKS_DIR, "JIR-1"), true);
+  });
+
+  it("returns false if lock already exists", () => {
+    acquireLock(TEST_LOCKS_DIR, "JIR-1");
+    const result = acquireLock(TEST_LOCKS_DIR, "JIR-1");
+    assert.equal(result, false);
+  });
+
+  it("creates the lock directory if it does not exist", () => {
+    assert.equal(existsSync(TEST_LOCKS_DIR), false);
+    acquireLock(TEST_LOCKS_DIR, "JIR-1");
+    assert.equal(existsSync(TEST_LOCKS_DIR), true);
+  });
+
+  it("writes metadata (pid, timestamp) to the lock file", () => {
+    acquireLock(TEST_LOCKS_DIR, "JIR-1");
+    const lockPath = join(TEST_LOCKS_DIR, "JIR-1.lock");
+    assert.equal(existsSync(lockPath), true);
+    const content = JSON.parse(readFileSync(lockPath, "utf8"));
+    assert.equal(typeof content.pid, "number");
+    assert.equal(typeof content.timestamp, "string");
+  });
+});
+
+describe("releaseLock", () => {
+  it("removes the lock file", () => {
+    acquireLock(TEST_LOCKS_DIR, "JIR-1");
+    releaseLock(TEST_LOCKS_DIR, "JIR-1");
+    assert.equal(isLocked(TEST_LOCKS_DIR, "JIR-1"), false);
+  });
+
+  it("does not throw if lock does not exist", () => {
+    assert.doesNotThrow(() => releaseLock(TEST_LOCKS_DIR, "NONEXISTENT"));
+  });
+});
+
+describe("isLocked", () => {
+  it("returns true when lock file exists", () => {
+    acquireLock(TEST_LOCKS_DIR, "JIR-1");
+    assert.equal(isLocked(TEST_LOCKS_DIR, "JIR-1"), true);
+  });
+
+  it("returns false when lock file does not exist", () => {
+    assert.equal(isLocked(TEST_LOCKS_DIR, "JIR-999"), false);
+  });
+
+  it("returns false when lock directory does not exist", () => {
+    assert.equal(isLocked("/nonexistent/path/locks", "JIR-1"), false);
+  });
+});
+
+describe("getLockedKeys", () => {
+  it("returns empty array when no locks exist", () => {
+    const keys = getLockedKeys(TEST_LOCKS_DIR);
+    assert.deepEqual(keys, []);
+  });
+
+  it("returns all locked issue keys", () => {
+    acquireLock(TEST_LOCKS_DIR, "JIR-1");
+    acquireLock(TEST_LOCKS_DIR, "JIR-2");
+    acquireLock(TEST_LOCKS_DIR, "PROJ-42");
+    const keys = getLockedKeys(TEST_LOCKS_DIR);
+    assert.equal(keys.length, 3);
+    assert.ok(keys.includes("JIR-1"));
+    assert.ok(keys.includes("JIR-2"));
+    assert.ok(keys.includes("PROJ-42"));
+  });
+});

src/lock.ts

@@ -0,0 +1,32 @@
+import { existsSync, mkdirSync, readdirSync, rmSync, writeFileSync } from "node:fs";
+import { join } from "node:path";
+
+export const LOCK_DIR = ".crewbit.locks";
+
+export function acquireLock(lockDir: string, issueKey: string): boolean {
+  const lockPath = join(lockDir, `${issueKey}.lock`);
+  if (existsSync(lockPath)) return false;
+  if (!existsSync(lockDir)) mkdirSync(lockDir, { recursive: true });
+  const metadata = JSON.stringify({
+    pid: process.pid,
+    timestamp: new Date().toISOString(),
+  });
+  writeFileSync(lockPath, metadata);
+  return true;
+}
+
+export function releaseLock(lockDir: string, issueKey: string): void {
+  const lockPath = join(lockDir, `${issueKey}.lock`);
+  if (existsSync(lockPath)) rmSync(lockPath, { force: true });
+}
+
+export function isLocked(lockDir: string, issueKey: string): boolean {
+  const lockPath = join(lockDir, `${issueKey}.lock`);
+  return existsSync(lockPath);
+}
+
+export function getLockedKeys(lockDir: string): string[] {
+  if (!existsSync(lockDir)) return [];
+  const files = readdirSync(lockDir);
+  return files.filter((f) => f.endsWith(".lock")).map((f) => f.replace(/\.lock$/, ""));
+}

src/types.ts

@@ -36,6 +36,7 @@ export interface WorkflowConfig {
     maxSessionSeconds: number;
     /** @deprecated use git.worktreePrefix instead */
     worktreePrefix?: string;
+    maxConcurrent?: number;
   };
   opencode?: OpenCodeConfig;
   git?: {