diff --git a/packages/engine/src/__tests__/project-engine.test.ts b/packages/engine/src/__tests__/project-engine.test.ts index 6a38ca07d..6f49a0b80 100644 --- a/packages/engine/src/__tests__/project-engine.test.ts +++ b/packages/engine/src/__tests__/project-engine.test.ts @@ -1,7 +1,6 @@ import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; import { ProjectEngine } from "../project-engine.js"; import { runtimeLog } from "../logger.js"; -import { aiMergeTask } from "../merger.js"; import { TunnelProcessManager } from "../remote-access/tunnel-process-manager.js"; const mocks = vi.hoisted(() => ({ @@ -1224,11 +1223,18 @@ describe("ProjectEngine swallowed error hardening", () => { expect(handler).toBeTypeOf("function"); if (!handler) throw new Error("task:moved handler was not registered"); + // Auto-merge enqueue runs inside a setTimeout grace period (~300ms) to + // let the executor's finally block complete before the merger starts. + // Use fake timers so the test doesn't actually sleep 300ms. + vi.useFakeTimers(); + await handler({ task: { id: "FN-001", column: "in-review" }, to: "in-review", }); + await vi.advanceTimersByTimeAsync(500); + expect(warnSpy).toHaveBeenCalledWith( expect.stringContaining("Auto-merge: failed to read settings for task:moved on FN-001"), ); diff --git a/packages/engine/src/agent-runtime.ts b/packages/engine/src/agent-runtime.ts index 100f326c2..e93045886 100644 --- a/packages/engine/src/agent-runtime.ts +++ b/packages/engine/src/agent-runtime.ts @@ -54,6 +54,21 @@ export interface AgentRuntimeOptions { skillSelection?: SkillSelectionContext; /** Convenience: skill names to include in the session */ skills?: string[]; + /** + * Last-chance abort hook fired by the runtime *immediately before* the + * underlying LLM session is instantiated — i.e., after all of the runtime's + * own awaited setup work (provider registration, resource loading, etc.). + * Throw from this callback to cancel session creation. + * + * Runtimes SHOULD invoke this hook at their latest synchronous decision + * point so callers can enforce time-sensitive predicates (notably the + * engine pause flag) without a TOCTOU window between an outer check and + * the actual session spawn. Runtimes that ignore this hook degrade + * gracefully — the caller's outer check still fires before + * `runtime.createSession()` is invoked, so the abort window is bounded by + * the runtime's internal setup latency rather than unbounded. + */ + beforeSpawnSession?: () => Promise | void; } /** diff --git a/packages/engine/src/agent-session-helpers.ts b/packages/engine/src/agent-session-helpers.ts index e6cce3b12..c91e81503 100644 --- a/packages/engine/src/agent-session-helpers.ts +++ b/packages/engine/src/agent-session-helpers.ts @@ -27,6 +27,15 @@ export interface ResolvedSessionOptions extends AgentRuntimeOptions { pluginRunner?: PluginRunner; /** Optional runtime hint from task/agent configuration */ runtimeHint?: string; + /** + * `beforeSpawnSession` is inherited from {@link AgentRuntimeOptions} — see + * its definition there for the contract. Callers (e.g. the reviewer's + * pause gate) throw from this callback to cancel session creation when + * external state changed during the async setup window. Forwarded + * verbatim to `runtime.createSession()`; the runtime is responsible for + * invoking it at its latest synchronous point before the underlying LLM + * session is instantiated. + */ } /** @@ -87,7 +96,10 @@ export async function createResolvedAgentSession( `[${sessionPurpose}] Using runtime "${resolved.runtimeId}" (configured=${resolved.wasConfigured})`, ); - // Create the session using the resolved runtime + // Forward `beforeSpawnSession` to the runtime so it fires at the true + // latest sync point (just before LLM session instantiation) rather than + // here, before the runtime's own awaited setup work runs. See + // AgentRuntimeOptions.beforeSpawnSession for the contract. const result = await resolved.runtime.createSession(runtimeOptions); // Attach the resolved runtime's promptWithFallback as a bound method on the diff --git a/packages/engine/src/concurrency.ts b/packages/engine/src/concurrency.ts index 3ffe28d15..8d0ba1f1b 100644 --- a/packages/engine/src/concurrency.ts +++ b/packages/engine/src/concurrency.ts @@ -15,8 +15,14 @@ interface PriorityWaiter { * A concurrency semaphore that gates all agentic activities (triage specification, * task execution, and merge operations) behind a shared slot limit. * - * The semaphore ensures that the total number of concurrently running AI agents - * never exceeds `maxConcurrent`, regardless of which subsystem spawned them. + * The semaphore ensures that the total number of concurrently running + * **top-level** AI agents never exceeds `maxConcurrent`, regardless of which + * subsystem spawned them. Nested helper agents (reviewers spawned from + * inside a parent's tool call) are admitted via {@link runNested} without + * entering the wait queue: they bump `activeCount` for honest observability + * and respect the parent's slot, but can transiently push the count above + * the configured limit. This is intentional — see {@link runNested} for the + * fairness/deadlock rationale. * * **Priority-based draining:** When a slot becomes available and multiple agents * are waiting, the waiter with the highest `priority` value is served first. @@ -131,6 +137,33 @@ export class AgentSemaphore { } } + /** + * Run a nested helper agent within the current caller's slot context. + * + * Unlike {@link run}, `runNested` does NOT enter the wait queue — it bumps + * `_active` directly so the helper begins immediately. The bump keeps + * {@link activeCount} an honest report of how many agent sessions exist + * right now, even though the helper bypasses the usual fairness queue. + * + * Intended use: a parent agent (executor, triage) is suspended awaiting a + * synchronous sub-agent's tool result (typically a reviewer). The parent + * makes no LLM calls while suspended, so the total number of LLM-active + * agents at any moment is still bounded by `maxConcurrent` — but two agent + * sessions exist, which `runNested` reflects in `activeCount`. This is + * intentionally a soft breach of the limit: it preserves forward-progress + * fairness for the in-flight task (no queue stealing) and avoids the + * deadlock that would occur if both parent and child needed a queued slot. + */ + async runNested(fn: () => Promise): Promise { + this._active++; + try { + return await fn(); + } finally { + this._active--; + this._drain(); + } + } + /** * Unblock waiters while slots are available. * diff --git a/packages/engine/src/executor.ts b/packages/engine/src/executor.ts index a15d0a992..7cd28a275 100644 --- a/packages/engine/src/executor.ts +++ b/packages/engine/src/executor.ts @@ -483,6 +483,10 @@ export class TaskExecutor { private resumingUnpaused = new Set(); /** Completed orphan recovery tasks currently running during startup. */ private recoveringCompleted = new Set(); + /** Tracks tasks whose workflow-rerun bounce is in flight (todo→in-progress). + * Prevents the task:moved handler from dispatching execute() before the + * bounce finishes its own dispatch. */ + private workflowRerunPending = new Set(); /** Active agent sessions per task, used to terminate on pause and inject steering. */ private activeSessions = new Map { - return new Set([...this.executing, ...this.recoveringCompleted]); + return new Set([...this.executing, ...this.recoveringCompleted, ...this.resumingUnpaused]); } /** @@ -700,7 +704,11 @@ export class TaskExecutor { && !this.activeSessions.has(task.id) && !this.activeStepExecutors.has(task.id) ) { - if (!this.executing.has(task.id) && !this.resumingUnpaused.has(task.id)) { + if ( + !this.executing.has(task.id) + && !this.resumingUnpaused.has(task.id) + && !this.recoveringCompleted.has(task.id) + ) { this.resumingUnpaused.add(task.id); executorLog.log(`Unpaused ${task.id} in-progress with no session — resuming execution`); try { @@ -979,36 +987,46 @@ export class TaskExecutor { const handle = setTimeout(async () => { this.completedTaskWatchdogs.delete(taskId); - let currentTask: Task | null = null; - try { - currentTask = await this.store.getTask(taskId); - } catch (err: unknown) { - const errorMessage = err instanceof Error ? err.message : String(err); - executorLog.warn(`${taskId}: completed-task watchdog could not read latest task state: ${errorMessage}`); + // Claim recovery slot atomically (synchronously) before any async work. + // Without this, two paths can pass the in-flight guards on the same + // event-loop turn and both call recoverCompletedTask() concurrently. + if ( + this.recoveringCompleted.has(taskId) + || this.executing.has(taskId) + || this.activeSessions.has(taskId) + || this.activeStepExecutors.has(taskId) + || this.resumingUnpaused.has(taskId) + ) { return; } - - if (!currentTask || currentTask.column !== "in-progress" || currentTask.paused) { - return; - } - if (this.activeSessions.has(taskId) || this.activeStepExecutors.has(taskId) || this.recoveringCompleted.has(taskId)) { - return; - } - if (!this.isTaskWorkComplete(currentTask)) { - return; - } - - executorLog.warn( - `${taskId}: completed-task watchdog fired after ${COMPLETED_TASK_WATCHDOG_MS / 1000}s ` + - `(${trigger}) — attempting direct recovery to in-review`, - ); - await this.store.logEntry( - taskId, - `Watchdog: task remained in-progress ${COMPLETED_TASK_WATCHDOG_MS / 1000}s after ${trigger} — attempting direct recovery to in-review`, - ).catch(() => undefined); - this.recoveringCompleted.add(taskId); + try { + let currentTask: Task | null = null; + try { + currentTask = await this.store.getTask(taskId); + } catch (err: unknown) { + const errorMessage = err instanceof Error ? err.message : String(err); + executorLog.warn(`${taskId}: completed-task watchdog could not read latest task state: ${errorMessage}`); + return; + } + + if (!currentTask || currentTask.column !== "in-progress" || currentTask.paused) { + return; + } + if (!this.isTaskWorkComplete(currentTask)) { + return; + } + + executorLog.warn( + `${taskId}: completed-task watchdog fired after ${COMPLETED_TASK_WATCHDOG_MS / 1000}s ` + + `(${trigger}) — attempting direct recovery to in-review`, + ); + await this.store.logEntry( + taskId, + `Watchdog: task remained in-progress ${COMPLETED_TASK_WATCHDOG_MS / 1000}s after ${trigger} — attempting direct recovery to in-review`, + ).catch(() => undefined); + const recovered = await this.recoverCompletedTask(currentTask); if (!recovered) { await this.store.logEntry( @@ -1024,29 +1042,54 @@ export class TaskExecutor { this.completedTaskWatchdogs.set(taskId, handle); } - private async performWorkflowRerunBounce(taskId: string, worktreePath: string): Promise { - // moveTask(in-progress → todo) clears `task.worktree`; restore it before - // the return trip so the dashboard never renders the task under - // "Unassigned" and self-healing can't reclaim the worktree as idle. - const latestTask = await this.store.getTask(taskId); - if (!latestTask) { - throw new Error("task missing during workflow rerun bounce"); + /** + * Result of a workflow-rerun bounce attempt. + * + * - `bounced` — the move sequence completed successfully and the task is + * back in `in-progress` ready for re-execution. + * - `skipped-pending` — another bounce for the same task is mid-flight; + * this attempt is a no-op. Callers (notably the watchdog) must NOT log + * this as a successful retry, since the original bounce may itself be + * stuck. + */ + private async performWorkflowRerunBounce( + taskId: string, + worktreePath: string, + ): Promise<"bounced" | "skipped-pending"> { + // Re-entry guard: if a previous bounce for the same task is still + // mid-flight (e.g., the watchdog fired before the original sequence + // completed), skip rather than racing two concurrent moveTask sequences. + if (this.workflowRerunPending.has(taskId)) { + executorLog.warn(`${taskId}: workflow rerun bounce already in flight — skipping re-entry`); + return "skipped-pending"; } + this.workflowRerunPending.add(taskId); + try { + // moveTask(in-progress → todo) clears `task.worktree`; restore it before + // the return trip so the dashboard never renders the task under + // "Unassigned" and self-healing can't reclaim the worktree as idle. + const latestTask = await this.store.getTask(taskId); + if (!latestTask) { + throw new Error("task missing during workflow rerun bounce"); + } - if (latestTask.column === "in-progress") { - await this.store.moveTask(taskId, "todo"); - await this.store.updateTask(taskId, { worktree: worktreePath }); - await this.store.moveTask(taskId, "in-progress"); - return; + if (latestTask.column === "in-progress") { + await this.store.moveTask(taskId, "todo"); + await this.store.updateTask(taskId, { worktree: worktreePath }); + await this.store.moveTask(taskId, "in-progress"); + return "bounced"; + } + + if (latestTask.column === "todo") { + await this.store.updateTask(taskId, { worktree: worktreePath }); + await this.store.moveTask(taskId, "in-progress"); + return "bounced"; + } + + throw new Error(`task is in '${latestTask.column}', cannot bounce to in-progress`); + } finally { + this.workflowRerunPending.delete(taskId); } - - if (latestTask.column === "todo") { - await this.store.updateTask(taskId, { worktree: worktreePath }); - await this.store.moveTask(taskId, "in-progress"); - return; - } - - throw new Error(`task is in '${latestTask.column}', cannot bounce to in-progress`); } private scheduleWorkflowRerun(taskId: string, worktreePath: string, successMessage: string): void { @@ -1054,8 +1097,12 @@ export class TaskExecutor { setTimeout(async () => { try { - await this.performWorkflowRerunBounce(taskId, worktreePath); - executorLog.log(successMessage); + const outcome = await this.performWorkflowRerunBounce(taskId, worktreePath); + if (outcome === "bounced") { + executorLog.log(successMessage); + } else { + executorLog.warn(`${taskId}: rerun bounce skipped — another bounce already in flight`); + } } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err); executorLog.error(`${taskId}: failed to schedule rerun bounce: ${errorMessage}`); @@ -1089,8 +1136,21 @@ export class TaskExecutor { ).catch(() => undefined); try { - await this.performWorkflowRerunBounce(taskId, worktreePath); - executorLog.warn(`${taskId}: workflow rerun watchdog retry succeeded`); + const outcome = await this.performWorkflowRerunBounce(taskId, worktreePath); + if (outcome === "bounced") { + executorLog.warn(`${taskId}: workflow rerun watchdog retry succeeded`); + } else { + // The original bounce is still mid-flight, which means *it* is the + // one that's hung — not us. Log honestly so operators don't see a + // false "succeeded" message while the task is actually stranded. + executorLog.error( + `${taskId}: workflow rerun watchdog retry skipped — original bounce still in flight after ${WORKFLOW_RERUN_WATCHDOG_MS / 1000}s; task may be stuck`, + ); + await this.store.logEntry( + taskId, + `Workflow rerun watchdog retry skipped — original bounce still in flight after ${WORKFLOW_RERUN_WATCHDOG_MS / 1000}s; task may be stuck`, + ).catch(() => undefined); + } } catch (err: unknown) { const errorMessage = err instanceof Error ? err.message : String(err); executorLog.error(`${taskId}: workflow rerun watchdog retry failed: ${errorMessage}`); @@ -1281,6 +1341,15 @@ export class TaskExecutor { */ async recoverCompletedTask(task: Task): Promise { try { + if ( + this.executing.has(task.id) + || this.activeSessions.has(task.id) + || this.activeStepExecutors.has(task.id) + || this.resumingUnpaused.has(task.id) + ) { + executorLog.log(`${task.id}: skipping recoverCompletedTask — task has active execution in flight`); + return false; + } const settings = await this.store.getSettings(); // Capture modified files if the worktree still exists @@ -3210,7 +3279,16 @@ export class TaskExecutor { try { const settings = await store.getSettings(); - const result = await reviewStep( + // Run the reviewer via semaphore.runNested so its slot accounting + // is honest: activeCount transiently bumps to reflect the second + // agent session, but the reviewer doesn't enter the wait queue + // (avoiding a fairness regression where unrelated work could + // overtake this task at low maxConcurrent). The parent (this + // executor) makes no LLM calls while suspended awaiting the tool + // result, so the soft breach of `limit` does not push real + // LLM-active concurrency above the configured cap. + const sem = options.semaphore; + const invokeReviewer = () => reviewStep( worktreePath, taskId, step, step_name, reviewType, promptContent, baseline, { @@ -3245,6 +3323,9 @@ export class TaskExecutor { settings, }, ); + const result = sem + ? await sem.runNested(invokeReviewer) + : await invokeReviewer(); await store.logEntry( taskId, diff --git a/packages/engine/src/pi.ts b/packages/engine/src/pi.ts index 276f70534..b90f03c6a 100644 --- a/packages/engine/src/pi.ts +++ b/packages/engine/src/pi.ts @@ -425,6 +425,9 @@ export interface AgentOptions { * (and `skillSelection` is not), auto-constructs a SkillSelectionContext * from the cwd and these names. Ignored when `skillSelection` is set. */ skills?: string[]; + /** Last-chance abort hook fired immediately before `createAgentSession`. + * See `AgentRuntimeOptions.beforeSpawnSession`. */ + beforeSpawnSession?: () => Promise | void; } function resolveConfiguredModel( @@ -1108,6 +1111,14 @@ export async function createFnAgent(options: AgentOptions): Promise ...(wrappedTools as ToolDefinition[]), ...(options.customTools ?? []), ]; + // Last-chance abort hook. Fires *here* — after every awaited setup step + // in createFnAgent (provider registration, worktree validation, resource + // loader reload) and immediately before the actual LLM session spawn. + // This is the latest synchronous decision point where the engine can + // honor a pause that flipped during this function's setup window. + if (options.beforeSpawnSession) { + await options.beforeSpawnSession(); + } return createAgentSession({ cwd: options.cwd, authStorage, diff --git a/packages/engine/src/project-engine.ts b/packages/engine/src/project-engine.ts index 1be80c709..974da04d5 100644 --- a/packages/engine/src/project-engine.ts +++ b/packages/engine/src/project-engine.ts @@ -45,6 +45,15 @@ export type ProcessPullRequestMergeFn = ( const execFileAsync = promisify(execFile); +/** + * Delay between a task moving to in-review and auto-merge being enqueued. + * Gives the executor's finally block time to complete session disposal, + * child-agent termination, and any in-flight reviewer teardown so the merger + * doesn't start emitting logs while the executor is still cleaning up. See + * FN-2910 for the observed overlap symptom. + */ +const MERGE_HANDOFF_GRACE_MS = 300; + interface RemoteLifecycleEvaluation { provider: TunnelProvider; config?: TunnelProviderConfig; @@ -1476,16 +1485,34 @@ export class ProjectEngine { if (to !== "in-review") return; if (task.paused) return; if (this.options.getTaskMergeBlocker?.(task)) return; - try { - const settings = await store.getSettings(); - if (settings.globalPause || settings.enginePaused) return; - if (!settings.autoMerge) return; - this.internalEnqueueMerge(task.id); - } catch (err: unknown) { - runtimeLog.warn( - `Auto-merge: failed to read settings for task:moved on ${task.id}: ${err instanceof Error ? err.message : String(err)}`, - ); - } + + // Grace period before handing off to the merger. The executor's finally + // block (session disposal, child-agent termination, in-flight reviewer + // teardown) runs *after* the moveTask("in-review") that fires this + // event. Without a delay, the merger's session can start emitting logs + // while the executor is still cleaning up — observed in FN-2910 as + // overlapping [reviewer]/[merger] log streams. The delay is also a + // belt-and-braces guard against any in-flight reviewer that the + // executor spawned just before transitioning. + setTimeout(async () => { + try { + // Re-validate eligibility after the grace period — the task may + // have been paused, moved, or had its merge blocked. + const latestTask = await store.getTask(task.id).catch(() => null); + if (!latestTask) return; + if (latestTask.column !== "in-review") return; + if (latestTask.paused) return; + if (this.options.getTaskMergeBlocker?.(latestTask)) return; + const settings = await store.getSettings(); + if (settings.globalPause || settings.enginePaused) return; + if (!settings.autoMerge) return; + this.internalEnqueueMerge(task.id); + } catch (err: unknown) { + runtimeLog.warn( + `Auto-merge: failed to read settings for task:moved on ${task.id}: ${err instanceof Error ? err.message : String(err)}`, + ); + } + }, MERGE_HANDOFF_GRACE_MS); }; store.on("task:moved", this.taskMovedHandler); } diff --git a/packages/engine/src/reviewer.ts b/packages/engine/src/reviewer.ts index 55dd9ae67..7fd55115d 100644 --- a/packages/engine/src/reviewer.ts +++ b/packages/engine/src/reviewer.ts @@ -404,26 +404,79 @@ export async function reviewStep( } : undefined), ] : undefined; - const { session } = await createResolvedAgentSession({ - sessionPurpose: "reviewer", - runtimeHint: extractRuntimeHint(memoryAgent?.runtimeConfig), - pluginRunner: options.pluginRunner, - cwd, - systemPrompt: reviewerSystemPrompt, - tools: "readonly", - customTools: memoryTools, - onText: agentLogger ? agentLogger.onText : (delta) => options.onText?.(delta), - onThinking: agentLogger?.onThinking, - onToolStart: agentLogger?.onToolStart, - onToolEnd: agentLogger?.onToolEnd, - defaultProvider: validatorProvider, - defaultModelId: validatorModelId, - fallbackProvider: validatorFallbackProvider, - fallbackModelId: validatorFallbackModelId, - defaultThinkingLevel: options.defaultThinkingLevel, - // Skill selection: use assigned agent skills if available, otherwise role fallback - ...(skillContext?.skillSelectionContext ? { skillSelection: skillContext.skillSelectionContext } : {}), - }); + // Sentinel error used by the beforeCreateSession hook to cancel session + // creation when a pause is detected after runtime resolution but before + // the LLM session is actually spawned. Caught locally and converted to an + // UNAVAILABLE verdict. + class ReviewerPauseAbortError extends Error { + constructor(public readonly reason: string) { + super(`reviewer aborted: ${reason}`); + } + } + + // Reviewers run within the parent agent's slot accounting via + // semaphore.runNested at the call site. The session spawn itself includes + // a last-chance pause check (beforeSpawnSession) that the runtime fires + // immediately before the underlying LLM session is instantiated — past + // every awaited setup step inside the runtime (provider registration, + // resource loading, etc.). This closes the TOCTOU window where a pause + // flipped during the setup chain. + let session: import("@mariozechner/pi-coding-agent").AgentSession; + try { + ({ session } = await createResolvedAgentSession({ + sessionPurpose: "reviewer", + runtimeHint: extractRuntimeHint(memoryAgent?.runtimeConfig), + pluginRunner: options.pluginRunner, + cwd, + systemPrompt: reviewerSystemPrompt, + tools: "readonly", + customTools: memoryTools, + onText: agentLogger ? agentLogger.onText : (delta) => options.onText?.(delta), + onThinking: agentLogger?.onThinking, + onToolStart: agentLogger?.onToolStart, + onToolEnd: agentLogger?.onToolEnd, + defaultProvider: validatorProvider, + defaultModelId: validatorModelId, + fallbackProvider: validatorFallbackProvider, + fallbackModelId: validatorFallbackModelId, + defaultThinkingLevel: options.defaultThinkingLevel, + // Skill selection: use assigned agent skills if available, otherwise role fallback + ...(skillContext?.skillSelectionContext ? { skillSelection: skillContext.skillSelectionContext } : {}), + beforeSpawnSession: async () => { + if (!options.store) return; + let finalSettings: Settings | undefined; + try { + finalSettings = await options.store.getSettings(); + } catch { + // Treat a transient store failure as "not paused" — better to + // proceed than to block reviews on a flaky read. + return; + } + if (finalSettings?.globalPause || finalSettings?.enginePaused) { + const reason = finalSettings.globalPause ? "Global pause" : "Engine paused"; + throw new ReviewerPauseAbortError(reason); + } + }, + })); + } catch (err) { + if (err instanceof ReviewerPauseAbortError) { + reviewerLog.log( + `${taskId}: ${reviewType} review for Step ${stepNumber} aborted before spawn — ${err.reason} active`, + ); + if (options.store && options.taskId) { + await options.store.logEntry( + options.taskId, + `${reviewType} review aborted before spawn — ${err.reason} active`, + ).catch(() => undefined); + } + return { + verdict: "UNAVAILABLE", + review: `${err.reason} active — reviewer not spawned. Stop calling fn_review_* and exit cleanly; the parent task will resume after unpause.`, + summary: `Skipped: ${err.reason}`, + }; + } + throw err; + } reviewerLog.log(`${taskId}: reviewer using model ${describeModel(session)}`); if (options.store && options.taskId) { @@ -451,10 +504,8 @@ export async function reviewStep( session.dispose(); } - // Extract verdict from the review text const verdict = extractVerdict(reviewText); const summary = extractSummary(reviewText); - return { verdict, review: reviewText, summary }; } diff --git a/packages/engine/src/self-healing.ts b/packages/engine/src/self-healing.ts index 187c689d1..8a53a88d5 100644 --- a/packages/engine/src/self-healing.ts +++ b/packages/engine/src/self-healing.ts @@ -745,6 +745,14 @@ export class SelfHealingManager { let recovered = 0; for (const task of stuckCompleted) { + // Re-check in-flight state inside the loop. The initial filter used a + // snapshot taken before any awaits; another path (executor resume, + // task:moved dispatch) may have claimed the task in between. + const latestExecutingIds = this.options.getExecutingTaskIds?.() ?? new Set(); + if (latestExecutingIds.has(task.id)) { + log.log(`${task.id} started executing concurrently — skipping recovery this cycle`); + continue; + } log.log(`Recovering completed task ${task.id}: ${task.title || task.description?.slice(0, 60) || "(untitled)"}`); const success = await recoverFn(task); if (success) recovered++; diff --git a/packages/engine/src/triage.ts b/packages/engine/src/triage.ts index 58630ee93..be6377fa3 100644 --- a/packages/engine/src/triage.ts +++ b/packages/engine/src/triage.ts @@ -1702,7 +1702,12 @@ export class TriageProcessor { // model changes made after the session started. const currentSettings = await store.getSettings(); - const result = await reviewStep( + // Spec reviewer runs via semaphore.runNested so it transiently + // bumps activeCount for honest observability while bypassing the + // wait queue (no fairness regression at low maxConcurrent). See + // concurrency.ts:runNested for the contract. + const sem = options.semaphore; + const invokeReviewer = () => reviewStep( rootDir, taskId, 0, @@ -1736,6 +1741,9 @@ export class TriageProcessor { rootDir, }, ); + const result = sem + ? await sem.runNested(invokeReviewer) + : await invokeReviewer(); // Track verdict for post-session enforcement specReviewVerdictRef.current = result.verdict;