root/fusion
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(engine): close executor/merger concurrency races and reviewer pause TOCTOU

Browse Source 
FN-2910 surfaced concurrent reviewer + merger activity on the same task.
Root cause: asymmetric in-flight guards let an unpause-resume kick off a
fresh executor session while a recovery path was already running, and the
auto-merge handoff fired before the executor's finally block finished
cleanup. This sweeps the surrounding lifecycle paths for similar races and
tightens the reviewer pause gate against TOCTOU through runtime setup.

- Symmetric in-flight tracking across `executing`, `recoveringCompleted`,
  and `resumingUnpaused`; `recoverCompletedTask` bails when any are set.
- Atomic claim of the recovery slot in the completed-task watchdog before
  any awaited work.
- Workflow-rerun bounce returns "bounced" | "skipped-pending" so the
  watchdog can no longer log a false-success retry when the original
  bounce is still mid-flight.
- Self-healing's completed-task scan re-checks executing IDs inside the
  loop instead of trusting a pre-await snapshot.
- 300ms grace period before auto-merge enqueue, giving the executor's
  finally block (session disposal, child cleanup) time to drain and
  eliminating the residual log-overlap symptom from FN-2910. Test uses
  fake timers, no real sleep added.
- New AgentSemaphore.runNested for synchronously nested helper agents
  (reviewers): bumps activeCount for honest observability while bypassing
  the wait queue, preserving forward-progress fairness for the parent at
  low maxConcurrent. Both createReviewStepTool and triage's
  createReviewSpecTool now use it.
- New beforeSpawnSession hook on AgentRuntimeOptions/AgentOptions fired
  inside createFnAgent immediately before createAgentSession, past every
  awaited setup step. Reviewer wires a pause re-check that throws a
  sentinel error converted to UNAVAILABLE, closing the TOCTOU window
  where pause flipped during runtime resolution or resource loading.

All 2887 engine tests pass; engine + core + cli + dashboard + plugin-sdk
+ pi-claude-cli + desktop typecheck clean.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsxdsm
2026-04-29 11:26:39 -07:00
parent 327ddd613c
commit 800c283e57
10 changed files with 342 additions and 90 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
packages/engine/src/__tests__/project-engine.test.ts
View File

@@ -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"),
);

15
packages/engine/src/agent-runtime.ts
View File

@@ -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> | void;
}
/**

14
packages/engine/src/agent-session-helpers.ts
View File

@@ -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

37
packages/engine/src/concurrency.ts
View File

@@ -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<T>(fn: () => Promise<T>): Promise<T> {
this._active++;
try {
return await fn();
} finally {
this._active--;
this._drain();
}
}
/**
* Unblock waiters while slots are available.
*

187
packages/engine/src/executor.ts
View File

@@ -483,6 +483,10 @@ export class TaskExecutor {
private resumingUnpaused = new Set<string>();
/** Completed orphan recovery tasks currently running during startup. */
private recoveringCompleted = new Set<string>();
/** 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<string>();
/** Active agent sessions per task, used to terminate on pause and inject steering. */
private activeSessions = new Map<string, {
session: AgentSession;
@@ -553,7 +557,7 @@ export class TaskExecutor {
/** Returns the set of task IDs currently being executed. */
getExecutingTaskIds(): Set<string> {
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<void> {
// 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<boolean> {
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,

11
packages/engine/src/pi.ts
View File

@@ -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> | void;
}
function resolveConfiguredModel(
@@ -1108,6 +1111,14 @@ export async function createFnAgent(options: AgentOptions): Promise<AgentResult>
...(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,

47
packages/engine/src/project-engine.ts
View File

@@ -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);
}

95
packages/engine/src/reviewer.ts
View File

@@ -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 };
}

8
packages/engine/src/self-healing.ts
View File

@@ -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<string>();
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++;

10
packages/engine/src/triage.ts
View File

@@ -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;