From c3afb0f2bfa679022b567d8572d1fe4d9c53c98f Mon Sep 17 00:00:00 2001 From: Duke Date: Thu, 9 Apr 2026 21:55:29 -0300 Subject: [PATCH 1/3] feat: multi-agent orchestration with locking and streaming output Add support for running multiple concurrent agents with file-based locking to prevent duplicate issue processing. Each agent's output is prefixed with [issueKey] for real-time visibility into what each agent is doing. - File-based lock mechanism in .crewbit.locks/ with acquire/release/isLocked - resolveNextActions() returns up to N issues respecting locks - daemon.maxConcurrent config option (default 1 for backward compat) - Session pool with graceful shutdown on SIGINT/SIGTERM - Streaming output multiplexed with [issueKey] prefix per line Co-Authored-By: Claude Sonnet 4.6 --- .gitignore | 3 +- src/commands/daemon.ts | 149 ++++++++++++++++++++++++++++++++++------- src/lock.test.ts | 91 +++++++++++++++++++++++++ src/lock.ts | 32 +++++++++ src/types.ts | 1 + src/workflow.test.ts | 73 +++++++++++++++++++- src/workflow.ts | 35 ++++++++-- 7 files changed, 352 insertions(+), 32 deletions(-) create mode 100644 src/lock.test.ts create mode 100644 src/lock.ts diff --git a/.gitignore b/.gitignore index b2437df..ebf6a2c 100644 --- a/.gitignore +++ b/.gitignore @@ -3,4 +3,5 @@ node_modules/ .claude/worktrees/ crewbit docs/.vitepress/dist -docs/.vitepress/cache \ No newline at end of file +docs/.vitepress/cache +.crewbit \ No newline at end of file diff --git a/src/commands/daemon.ts b/src/commands/daemon.ts index 59f1bbe..53d2627 100644 --- a/src/commands/daemon.ts +++ b/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,148 @@ async function sleep(seconds: number): Promise { return new Promise((resolvePromise) => setTimeout(resolvePromise, seconds * 1000)); } +type ActiveSession = { + issueKey: string; + promise: Promise; + done: boolean; + result: boolean | null; +}; + +async function runSession( + action: QueueAction & { type: "run" }, + config: WorkflowConfig, + dryRun: boolean, + lockDir: string, +): Promise { + const logPrefix = `[${action.issueKey}]`; + const prefixedLog = (message: string) => { + log(`${logPrefix} ${message}`); + }; + const runner = createRunner(config, REPO_ROOT, prefixedLog); + try { + return await runner.run(action, config, dryRun); + } finally { + releaseLock(lockDir, action.issueKey); + } +} + export async function runDaemonCommand(args: { configPath: string; dryRun: boolean; }): Promise { const { configPath, dryRun } = args; - let exp = 1; + let emptyBackoffExp = 1; + let failureBackoffExp = 1; + const lockDir = join(REPO_ROOT, LOCK_DIR); log(`crewbit starting${dryRun ? " (dry-run)" : ""}`); log(`Config: ${configPath}`); + const sessions = new Map(); + process.on("SIGINT", () => { - log("Stopped."); - process.exit(0); + log("Stopping... waiting for in-flight sessions to finish."); + Promise.allSettled([...sessions.values()].map((session) => session.promise)).then(() => { + log("Stopped."); + process.exit(0); + }); }); process.on("SIGTERM", () => { - log("Stopped."); - process.exit(0); + log("Stopping... waiting for in-flight sessions to finish."); + Promise.allSettled([...sessions.values()].map((session) => session.promise)).then(() => { + log("Stopped."); + process.exit(0); + }); }); 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 provider = createProvider(config); - const action = await resolveNextAction(config, provider); - - if (action.type === "idle") { - log(`Queue empty. Next check in ${waitSeconds * exp}s. (Ctrl+C to stop)`); - await sleep(waitSeconds * exp); - exp *= 2; - } else { - const runner = createRunner(config, REPO_ROOT, log); - const ok = await runner.run(action, config, dryRun); - if (ok) { - exp = 1; + let hadFailure = false; + for (const [issueKey, session] of sessions.entries()) { + if (!session.done) continue; + if (session.result) { + log(`[${issueKey}] Session succeeded`); } else { - const backoff = waitSeconds * exp; - log(`Session failed. Backing off ${backoff}s before retry.`); - await sleep(backoff); - exp = Math.min(exp * 2, 32); + log(`[${issueKey}] Session failed`); + hadFailure = true; } + sessions.delete(issueKey); + } + + const lockedKeys = new Set(getLockedKeys(lockDir)); + for (const issueKey of sessions.keys()) { + lockedKeys.add(issueKey); + } + + const slotsAvailable = Math.max(0, maxConcurrent - sessions.size); + let startedCount = 0; + + if (slotsAvailable > 0) { + const provider = createProvider(config); + const actions = await resolveNextActions(config, provider, slotsAvailable, lockedKeys); + + 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, + }; + + const sessionPromise = 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; + }); + + tracked.promise = sessionPromise; + sessions.set(action.issueKey, tracked); + startedCount += 1; + } + } + + if (startedCount > 0) { + emptyBackoffExp = 1; + failureBackoffExp = 1; + await sleep(pollSeconds); + continue; + } + + if (sessions.size > 0) { + await sleep(pollSeconds); + continue; + } + + 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 { + 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)}`); diff --git a/src/lock.test.ts b/src/lock.test.ts new file mode 100644 index 0000000..1ec79bd --- /dev/null +++ b/src/lock.test.ts @@ -0,0 +1,91 @@ +import assert from "node:assert/strict"; +import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; +import { beforeEach, describe, it } from "node:test"; +import { LOCK_DIR, 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")); + }); +}); diff --git a/src/lock.ts b/src/lock.ts new file mode 100644 index 0000000..b40c895 --- /dev/null +++ b/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$/, "")); +} diff --git a/src/types.ts b/src/types.ts index be57bde..7c3e4c5 100644 --- a/src/types.ts +++ b/src/types.ts @@ -36,6 +36,7 @@ export interface WorkflowConfig { maxSessionSeconds: number; /** @deprecated use git.worktreePrefix instead */ worktreePrefix?: string; + maxConcurrent?: number; }; opencode?: OpenCodeConfig; git?: { diff --git a/src/workflow.test.ts b/src/workflow.test.ts index dbda151..2685c75 100644 --- a/src/workflow.test.ts +++ b/src/workflow.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import { describe, it } from "node:test"; import type { WorkflowConfig } from "./types.js"; -import { resolveNextAction } from "./workflow.js"; +import { resolveNextAction, resolveNextActions } from "./workflow.js"; const baseConfig: WorkflowConfig = { provider: "jira", @@ -85,3 +85,74 @@ describe("resolveNextAction", () => { } }); }); + +describe("resolveNextActions", () => { + it("returns empty array when queue is empty", async () => { + const config: WorkflowConfig = { + ...baseConfig, + transitions: { + Start: { from: "To Do", command: "/develop" }, + }, + }; + const actions = await resolveNextActions(config, makeProvider({}), 3, new Set()); + assert.deepEqual(actions, []); + }); + + it("returns up to N actions across transitions", async () => { + const config: WorkflowConfig = { + ...baseConfig, + transitions: { + Done: { from: "Accepted", command: "/merge" }, + Start: { from: "To Do", command: "/develop" }, + }, + }; + const provider = makeProvider({ Accepted: ["JIR-5", "JIR-10"], "To Do": ["JIR-6", "JIR-7"] }); + const actions = await resolveNextActions(config, provider, 3, new Set()); + assert.equal(actions.length, 3); + assert.equal(actions[0].issueKey, "JIR-5"); + assert.equal(actions[1].issueKey, "JIR-10"); + assert.equal(actions[2].issueKey, "JIR-6"); + }); + + it("skips issues that are already locked", async () => { + const config: WorkflowConfig = { + ...baseConfig, + transitions: { + Start: { from: "To Do", command: "/develop" }, + }, + }; + const provider = makeProvider({ "To Do": ["JIR-1", "JIR-2", "JIR-3"] }); + const locked = new Set(["JIR-1", "JIR-3"]); + const actions = await resolveNextActions(config, provider, 3, locked); + assert.equal(actions.length, 1); + assert.equal(actions[0].issueKey, "JIR-2"); + }); + + it("returns fewer actions when locked issues reduce available pool", async () => { + const config: WorkflowConfig = { + ...baseConfig, + transitions: { + Start: { from: "To Do", command: "/develop" }, + }, + }; + const provider = makeProvider({ "To Do": ["JIR-1", "JIR-2"] }); + const locked = new Set(["JIR-1", "JIR-2"]); + const actions = await resolveNextActions(config, provider, 5, locked); + assert.deepEqual(actions, []); + }); + + it("uses correct prompt for each action", async () => { + const config: WorkflowConfig = { + ...baseConfig, + transitions: { + Done: { from: "Accepted", command: "/merge", prompt: "Merge {issueKey}" }, + Start: { from: "To Do", command: "/develop" }, + }, + }; + const provider = makeProvider({ Accepted: ["JIR-5"], "To Do": ["JIR-6"] }); + const actions = await resolveNextActions(config, provider, 2, new Set()); + assert.equal(actions.length, 2); + assert.equal(actions[0].prompt, "Merge JIR-5"); + assert.equal(actions[1].prompt, "/develop JIR-6"); + }); +}); diff --git a/src/workflow.ts b/src/workflow.ts index affd9be..dc7aee3 100644 --- a/src/workflow.ts +++ b/src/workflow.ts @@ -5,6 +5,15 @@ import { GitHubProjectsProvider } from "./providers/github-projects.js"; import { JiraProvider } from "./providers/jira.js"; import type { IssueProvider, QueueAction, WorkflowConfig } from "./types.js"; +function buildAction( + issueKey: string, + transition: { command: string; prompt?: string }, +): QueueAction & { type: "run" } { + const template = transition.prompt ?? "{command} {issueKey}"; + const prompt = template.replace("{command}", transition.command).replace("{issueKey}", issueKey); + return { type: "run", issueKey, command: transition.command, prompt }; +} + export function loadConfig(workflowPath: string): WorkflowConfig { const raw = readFileSync(resolve(workflowPath), "utf8"); return yaml.load(raw) as WorkflowConfig; @@ -40,13 +49,27 @@ export async function resolveNextAction( for (const transition of Object.values(config.transitions)) { const issues = await provider.getIssuesByStatus(transition.from); if (issues.length > 0) { - const issueKey = issues[0].key; - const template = transition.prompt ?? "{command} {issueKey}"; - const prompt = template - .replace("{command}", transition.command) - .replace("{issueKey}", issueKey); - return { type: "run", issueKey, command: transition.command, prompt }; + return buildAction(issues[0].key, transition); } } return { type: "idle" }; } + +export async function resolveNextActions( + config: WorkflowConfig, + provider: IssueProvider, + maxCount: number, + lockedKeys: Set, +): Promise<(QueueAction & { type: "run" })[]> { + const actions: (QueueAction & { type: "run" })[] = []; + for (const transition of Object.values(config.transitions)) { + if (actions.length >= maxCount) break; + const issues = await provider.getIssuesByStatus(transition.from); + for (const issue of issues) { + if (actions.length >= maxCount) break; + if (lockedKeys.has(issue.key)) continue; + actions.push(buildAction(issue.key, transition)); + } + } + return actions; +} From 63fa950afb64e0906c69fa4f8d7f1176c2f985d5 Mon Sep 17 00:00:00 2001 From: Duke Date: Thu, 9 Apr 2026 21:59:42 -0300 Subject: [PATCH 2/3] chore: improve lock --- src/lock.test.ts | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/src/lock.test.ts b/src/lock.test.ts index 1ec79bd..b18e462 100644 --- a/src/lock.test.ts +++ b/src/lock.test.ts @@ -1,10 +1,10 @@ import assert from "node:assert/strict"; -import { existsSync, mkdirSync, readFileSync, rmSync, writeFileSync } from "node:fs"; +import { existsSync, readFileSync, rmSync } from "node:fs"; import { join } from "node:path"; import { beforeEach, describe, it } from "node:test"; -import { LOCK_DIR, acquireLock, getLockedKeys, isLocked, releaseLock } from "./lock.js"; +import { acquireLock, getLockedKeys, isLocked, releaseLock } from "./lock.js"; -const TEST_LOCKS_DIR = join(process.cwd(), ".crewbit.test.locks"); +const TEST_LOCKS_DIR = join(process.cwd(), ".crewbit/test.locks"); function cleanTestDir() { if (existsSync(TEST_LOCKS_DIR)) { From 9ae2646e5fa585f3553e8c93c105feda7dbcac14 Mon Sep 17 00:00:00 2001 From: Duke Date: Thu, 9 Apr 2026 22:03:14 -0300 Subject: [PATCH 3/3] refactor: reduce cognitive complexity of runDaemonCommand Extract reapFinishedSessions() and startNewSessions() to bring the main loop cognitive complexity below the allowed threshold. Co-Authored-By: Claude Sonnet 4.6 --- src/commands/daemon.ts | 147 +++++++++++++++++++++-------------------- 1 file changed, 75 insertions(+), 72 deletions(-) diff --git a/src/commands/daemon.ts b/src/commands/daemon.ts index 53d2627..fa4a231 100644 --- a/src/commands/daemon.ts +++ b/src/commands/daemon.ts @@ -31,10 +31,7 @@ async function runSession( dryRun: boolean, lockDir: string, ): Promise { - const logPrefix = `[${action.issueKey}]`; - const prefixedLog = (message: string) => { - log(`${logPrefix} ${message}`); - }; + const prefixedLog = (message: string) => log(`[${action.issueKey}] ${message}`); const runner = createRunner(config, REPO_ROOT, prefixedLog); try { return await runner.run(action, config, dryRun); @@ -43,6 +40,72 @@ async function runSession( } } +function reapFinishedSessions(sessions: Map): 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, + lockDir: string, + dryRun: boolean, + maxConcurrent: number, +): Promise { + 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; @@ -51,26 +114,20 @@ export async function runDaemonCommand(args: { let emptyBackoffExp = 1; let failureBackoffExp = 1; const lockDir = join(REPO_ROOT, LOCK_DIR); + const sessions = new Map(); log(`crewbit starting${dryRun ? " (dry-run)" : ""}`); log(`Config: ${configPath}`); - const sessions = new Map(); - - process.on("SIGINT", () => { - log("Stopping... waiting for in-flight sessions to finish."); - Promise.allSettled([...sessions.values()].map((session) => session.promise)).then(() => { - log("Stopped."); - process.exit(0); - }); - }); - process.on("SIGTERM", () => { + const shutdown = () => { log("Stopping... waiting for in-flight sessions to finish."); - Promise.allSettled([...sessions.values()].map((session) => session.promise)).then(() => { + 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 { @@ -79,62 +136,8 @@ export async function runDaemonCommand(args: { const waitSeconds = Number(process.env.WAIT_SECONDS ?? config.daemon?.waitSeconds ?? 60); const pollSeconds = 2; - 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); - } - - const lockedKeys = new Set(getLockedKeys(lockDir)); - for (const issueKey of sessions.keys()) { - lockedKeys.add(issueKey); - } - - const slotsAvailable = Math.max(0, maxConcurrent - sessions.size); - let startedCount = 0; - - if (slotsAvailable > 0) { - const provider = createProvider(config); - const actions = await resolveNextActions(config, provider, slotsAvailable, lockedKeys); - - 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, - }; - - const sessionPromise = 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; - }); - - tracked.promise = sessionPromise; - sessions.set(action.issueKey, tracked); - startedCount += 1; - } - } + const hadFailure = reapFinishedSessions(sessions); + const startedCount = await startNewSessions(config, sessions, lockDir, dryRun, maxConcurrent); if (startedCount > 0) { emptyBackoffExp = 1;