feat(FN-882): add loop detection recovery with compact-and-resume
- Add ContextLimitDetector to detect agent loops via repeated tool call patterns - Implement compact-and-resume strategy: summarize conversation and restart agent from current step - Add loop recovery to StuckTaskDetector with configurable attempt tracking and retry limits - Extend executor with automatic loop recovery on context limit detection - Add loop recovery support to pi executor with same compact-and-resume pattern - Add comprehensive tests for context-limit-detector, stuck-task-detector loop detection, executor, and pi recovery - Add changeset for patch bump to @gsxdsm/fusion - Update README with loop detection and recovery documentation
This commit is contained in:
gsxdsm
@@ -19,7 +19,7 @@
|
||||
"build": "tsc",
|
||||
"typecheck": "tsc --noEmit",
|
||||
"test": "vitest run --exclude src/executor.test.ts && pnpm run test:executor",
|
||||
"test:executor": "vitest run src/executor.test.ts -t \"TaskExecutor with semaphore|TaskExecutor worktreeInitCommand|TaskExecutor worktree naming\" && vitest run src/executor.test.ts -t \"TaskExecutor worktree recovery\" && vitest run src/executor.test.ts -t \"TaskExecutor dependency-based worktree creation\" && vitest run src/executor.test.ts -t \"TaskExecutor worktree pool integration|WorktreePool capacity|Merger worktree pool integration\" && vitest run src/executor.test.ts -t \"buildExecutionPrompt|summarizeToolArgs|TaskExecutor pause behavior|TaskExecutor global pause behavior|TaskExecutor enginePaused soft pause\" && vitest run src/executor.test.ts -t \"Code review verdict|RETHINK verdict handling|Plan RETHINK verdict handling|E2E review pipeline|task_add_dep tool|TaskExecutor usage limit detection|Per-task model overrides|Invalid transition error handling|TaskExecutor task_done with summary|Workflow Steps Execution|Real-time steering injection\""
|
||||
"test:executor": "vitest run src/executor.test.ts -t \"TaskExecutor with semaphore|TaskExecutor worktreeInitCommand|TaskExecutor worktree naming\" && vitest run src/executor.test.ts -t \"TaskExecutor worktree recovery\" && vitest run src/executor.test.ts -t \"TaskExecutor dependency-based worktree creation\" && vitest run src/executor.test.ts -t \"TaskExecutor worktree pool integration|WorktreePool capacity|Merger worktree pool integration\" && vitest run src/executor.test.ts -t \"buildExecutionPrompt|summarizeToolArgs|TaskExecutor pause behavior|TaskExecutor global pause behavior|TaskExecutor enginePaused soft pause\" && vitest run src/executor.test.ts -t \"Code review verdict|RETHINK verdict handling|Plan RETHINK verdict handling|E2E review pipeline|task_add_dep tool|TaskExecutor usage limit detection|Per-task model overrides|Invalid transition error handling|TaskExecutor task_done with summary|Workflow Steps Execution|Real-time steering injection|TaskExecutor loop recovery\""
|
||||
},
|
||||
"dependencies": {
|
||||
"@fusion/core": "workspace:*",
|
||||
|
||||
117
packages/engine/src/context-limit-detector.test.ts
Normal file
117
packages/engine/src/context-limit-detector.test.ts
Normal file
@@ -0,0 +1,117 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { isContextLimitError } from "./context-limit-detector.js";
|
||||
|
||||
describe("isContextLimitError", () => {
|
||||
// ── Positive matches: known provider patterns ──────────────────────
|
||||
|
||||
it("matches Anthropic 'prompt is too long' error", () => {
|
||||
expect(isContextLimitError("prompt is too long: 210000 tokens > 200000 maximum")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches OpenAI 'exceeds the context window' error", () => {
|
||||
expect(isContextLimitError("This model's maximum context length is 128000 tokens. Your input exceeds the context window.")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches Google Gemini 'input token count exceeds' error", () => {
|
||||
expect(isContextLimitError("input token count exceeds the maximum limit of 1000000")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches xAI 'maximum prompt length' error", () => {
|
||||
expect(isContextLimitError("maximum prompt length is 131072 but request contains 150000")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches Groq 'reduce the length of the messages' error", () => {
|
||||
expect(isContextLimitError("Please reduce the length of the messages")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches Mistral context length error", () => {
|
||||
expect(isContextLimitError("Prompt contains 35000 tokens ... too large for model with 32000 maximum context length")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches OpenRouter context length error", () => {
|
||||
expect(isContextLimitError("maximum context length is 128000 tokens")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches llama.cpp 'exceeds the available context size' error", () => {
|
||||
expect(isContextLimitError("attempt to access position 8193 exceeds the available context size")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches LM Studio 'greater than the context length' error", () => {
|
||||
expect(isContextLimitError("request has 9000 tokens which is greater than the context length of 8192")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches Kimi 'exceeded model token limit' error", () => {
|
||||
expect(isContextLimitError("exceeded model token limit: 131072 (requested: 150000)")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches generic 'context length exceeded'", () => {
|
||||
expect(isContextLimitError("context length exceeded")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches generic 'context window exceeded'", () => {
|
||||
expect(isContextLimitError("context window exceeded")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches generic 'context size exceeded'", () => {
|
||||
expect(isContextLimitError("context size exceeded")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches 'too many tokens'", () => {
|
||||
expect(isContextLimitError("too many tokens in request")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches Anthropic 'would exceed' variant", () => {
|
||||
expect(isContextLimitError("messages with that many tokens would exceed the limit")).toBe(true);
|
||||
});
|
||||
|
||||
it("matches 'token limit ... context' pattern", () => {
|
||||
expect(isContextLimitError("token limit reached for context window")).toBe(true);
|
||||
});
|
||||
|
||||
// ── Negative matches: must NOT trigger ─────────────────────────────
|
||||
|
||||
it("returns false for empty string", () => {
|
||||
expect(isContextLimitError("")).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false for undefined-ish empty input", () => {
|
||||
expect(isContextLimitError("")).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false for generic 'Aborted' error", () => {
|
||||
expect(isContextLimitError("Aborted")).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false for rate limit error", () => {
|
||||
expect(isContextLimitError("429 Too Many Requests")).toBe(false);
|
||||
expect(isContextLimitError("rate_limit_error: Rate limit exceeded")).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false for generic 'limit exceeded' without context keywords", () => {
|
||||
expect(isContextLimitError("limit exceeded")).toBe(false);
|
||||
expect(isContextLimitError("quota exceeded")).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false for server error", () => {
|
||||
expect(isContextLimitError("500 Internal Server Error")).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false for connection error", () => {
|
||||
expect(isContextLimitError("ECONNREFUSED")).toBe(false);
|
||||
expect(isContextLimitError("connection refused")).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false for usage/billing error", () => {
|
||||
expect(isContextLimitError("billing limit reached")).toBe(false);
|
||||
expect(isContextLimitError("usage cap exceeded")).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false for transient network error", () => {
|
||||
expect(isContextLimitError("fetch failed")).toBe(false);
|
||||
expect(isContextLimitError("ETIMEDOUT")).toBe(false);
|
||||
});
|
||||
|
||||
it("returns false for overloaded error (not context)", () => {
|
||||
expect(isContextLimitError("overloaded_error: Overloaded")).toBe(false);
|
||||
});
|
||||
});
|
||||
60
packages/engine/src/context-limit-detector.ts
Normal file
60
packages/engine/src/context-limit-detector.ts
Normal file
@@ -0,0 +1,60 @@
|
||||
/**
|
||||
* Context limit error detection.
|
||||
*
|
||||
* Classifies errors from LLM providers that indicate the conversation context
|
||||
* has grown too large for the model's window. Used by the executor to trigger
|
||||
* compact-and-resume recovery before falling back to kill/requeue.
|
||||
*
|
||||
* Patterns are intentionally conservative — we only match errors that
|
||||
* explicitly reference context/token overflow, NOT generic rate limits or
|
||||
* server errors (those are handled by usage-limit-detector and transient-error-detector).
|
||||
*/
|
||||
|
||||
/** Patterns that indicate a context-window overflow from the LLM provider. */
|
||||
const CONTEXT_OVERFLOW_PATTERNS: RegExp[] = [
|
||||
// Anthropic: "prompt is too long: X tokens > Y maximum"
|
||||
/prompt is too long/i,
|
||||
// OpenAI (Completions & Responses): "exceeds the context window"
|
||||
/exceeds?\s+the\s+context\s+window/i,
|
||||
// Google Gemini: "input token count exceeds the maximum"
|
||||
/input token count exceeds/i,
|
||||
// xAI (Grok): "maximum prompt length is X but request contains Y"
|
||||
/maximum prompt length/i,
|
||||
// Groq: "reduce the length of the messages"
|
||||
/reduce the length of the messages/i,
|
||||
// Mistral: "too large for model with Y maximum context length"
|
||||
/too large for model with.*maximum context length/i,
|
||||
// OpenRouter (all backends): "maximum context length is X tokens"
|
||||
/maximum context length is \d+ tokens/i,
|
||||
// llama.cpp: "exceeds the available context size"
|
||||
/exceeds?\s+the\s+available\s+context\s+size/i,
|
||||
// LM Studio: "greater than the context length"
|
||||
/greater than the context length/i,
|
||||
// Kimi: "exceeded model token limit"
|
||||
/exceeded model token limit/i,
|
||||
// Generic catch-all: "context length exceeded" / "context window exceeded"
|
||||
/context (?:length|window|size) exceeded/i,
|
||||
// Token limit patterns with context keywords
|
||||
/token limit.*context/i,
|
||||
/too many tokens/i,
|
||||
// Anthropic variant: "messages with that many tokens would exceed"
|
||||
/tokens? would exceed/i,
|
||||
];
|
||||
|
||||
/**
|
||||
* Check if an error message indicates a context-window overflow.
|
||||
*
|
||||
* Returns true only when the message explicitly references context overflow
|
||||
* from a known LLM provider pattern. Returns false for:
|
||||
* - Rate limit errors (handled by usage-limit-detector)
|
||||
* - Transient network errors (handled by transient-error-detector)
|
||||
* - Generic "limit exceeded" without context keywords (false positive prevention)
|
||||
* - "Aborted" errors without context signal
|
||||
*
|
||||
* @param message — The error message string to classify
|
||||
* @returns true if the message indicates a context overflow
|
||||
*/
|
||||
export function isContextLimitError(message: string): boolean {
|
||||
if (!message) return false;
|
||||
return CONTEXT_OVERFLOW_PATTERNS.some((pattern) => pattern.test(message));
|
||||
}
|
||||
@@ -5,6 +5,13 @@ import { AgentSemaphore } from "./concurrency.js";
|
||||
vi.mock("./pi.js", () => ({
|
||||
createKbAgent: vi.fn(),
|
||||
describeModel: vi.fn().mockReturnValue("mock-provider/mock-model"),
|
||||
compactSessionContext: vi.fn(async (session, instructions) => {
|
||||
// Delegate to session.compact if available (supports loop recovery tests)
|
||||
if (typeof (session as any).compact === "function") {
|
||||
return (session as any).compact(instructions);
|
||||
}
|
||||
return null;
|
||||
}),
|
||||
promptWithFallback: vi.fn(async (session, prompt, options) => {
|
||||
if (options === undefined) {
|
||||
await session.prompt(prompt);
|
||||
@@ -79,6 +86,7 @@ import { WorktreePool } from "./worktree-pool.js";
|
||||
import { generateWorktreeName, slugify } from "./worktree-names.js";
|
||||
import type { Column, Task, TaskDetail } from "@fusion/core";
|
||||
import { SessionManager } from "@mariozechner/pi-coding-agent";
|
||||
import { StuckTaskDetector } from "./stuck-task-detector.js";
|
||||
|
||||
const mockedCreateHaiAgent = vi.mocked(createKbAgent);
|
||||
const mockedSessionManager = vi.mocked(SessionManager);
|
||||
@@ -6740,3 +6748,137 @@ describe("Real-time steering injection", () => {
|
||||
await executePromise;
|
||||
});
|
||||
});
|
||||
|
||||
// ── Loop recovery (compact-and-resume) integration tests ────────────
|
||||
|
||||
describe("TaskExecutor loop recovery", () => {
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
});
|
||||
|
||||
function createMockSessionForLoopRecovery(overrides?: { compactResult?: any }) {
|
||||
const defaultResult = {
|
||||
summary: "Compacted conversation",
|
||||
tokensBefore: 150000,
|
||||
};
|
||||
const compactRetVal = overrides && "compactResult" in overrides ? overrides.compactResult : defaultResult;
|
||||
const compact = vi.fn(async () => compactRetVal);
|
||||
const steer = vi.fn(async () => {});
|
||||
|
||||
return {
|
||||
prompt: vi.fn(async () => {}),
|
||||
dispose: vi.fn(),
|
||||
subscribe: vi.fn(),
|
||||
setThinkingLevel: vi.fn(),
|
||||
steer,
|
||||
compact,
|
||||
sessionFile: "/tmp/test-session.json",
|
||||
model: { provider: "mock", id: "mock-model", name: "Mock" },
|
||||
sessionManager: { getLeafId: vi.fn().mockReturnValue("leaf-1") },
|
||||
state: {},
|
||||
};
|
||||
}
|
||||
|
||||
function setupExecutorWithActiveSession(mockSession: ReturnType<typeof createMockSessionForLoopRecovery>) {
|
||||
const store = createMockStore();
|
||||
(store.getSettings as any).mockResolvedValue({
|
||||
maxConcurrent: 2,
|
||||
maxWorktrees: 4,
|
||||
pollIntervalMs: 15000,
|
||||
groupOverlappingFiles: false,
|
||||
autoMerge: false,
|
||||
});
|
||||
|
||||
const executor = new TaskExecutor(store, "/tmp/test-root");
|
||||
|
||||
// Directly inject an active session (avoids full execute() chain)
|
||||
(executor as any).activeSessions.set("FN-001", {
|
||||
session: mockSession,
|
||||
seenSteeringIds: new Set(),
|
||||
});
|
||||
|
||||
return { store, executor, mockSession };
|
||||
}
|
||||
|
||||
it("handleLoopDetected returns true and compacts session when active session exists", async () => {
|
||||
const mockSession = createMockSessionForLoopRecovery();
|
||||
const { store, executor } = setupExecutorWithActiveSession(mockSession);
|
||||
|
||||
const result = await executor.handleLoopDetected({
|
||||
taskId: "FN-001",
|
||||
reason: "loop",
|
||||
noProgressMs: 600000,
|
||||
inactivityMs: 0,
|
||||
activitySinceProgress: 100,
|
||||
shouldRequeue: true,
|
||||
});
|
||||
|
||||
expect(result).toBe(true);
|
||||
expect(mockSession.compact).toHaveBeenCalled();
|
||||
expect(mockSession.steer).toHaveBeenCalled();
|
||||
expect(store.logEntry).toHaveBeenCalledWith(
|
||||
"FN-001",
|
||||
expect.stringContaining("compact-and-resume"),
|
||||
);
|
||||
});
|
||||
|
||||
it("handleLoopDetected returns false when no active session", async () => {
|
||||
const store = createMockStore();
|
||||
const executor = new TaskExecutor(store, "/tmp/test-root");
|
||||
|
||||
// No session active (activeSessions is empty)
|
||||
const result = await executor.handleLoopDetected({
|
||||
taskId: "FN-001",
|
||||
reason: "loop",
|
||||
noProgressMs: 600000,
|
||||
inactivityMs: 0,
|
||||
activitySinceProgress: 100,
|
||||
shouldRequeue: true,
|
||||
});
|
||||
|
||||
expect(result).toBe(false);
|
||||
});
|
||||
|
||||
it("handleLoopDetected returns false when attempt ceiling reached", async () => {
|
||||
const mockSession = createMockSessionForLoopRecovery();
|
||||
const { executor } = setupExecutorWithActiveSession(mockSession);
|
||||
|
||||
// First call succeeds
|
||||
const result1 = await executor.handleLoopDetected({
|
||||
taskId: "FN-001",
|
||||
reason: "loop",
|
||||
noProgressMs: 600000,
|
||||
inactivityMs: 0,
|
||||
activitySinceProgress: 100,
|
||||
shouldRequeue: true,
|
||||
});
|
||||
expect(result1).toBe(true);
|
||||
|
||||
// Second call hits ceiling (max 1 attempt per execute() lifecycle)
|
||||
const result2 = await executor.handleLoopDetected({
|
||||
taskId: "FN-001",
|
||||
reason: "loop",
|
||||
noProgressMs: 600000,
|
||||
inactivityMs: 0,
|
||||
activitySinceProgress: 200,
|
||||
shouldRequeue: true,
|
||||
});
|
||||
expect(result2).toBe(false);
|
||||
});
|
||||
|
||||
it("handleLoopDetected returns false when compaction fails", async () => {
|
||||
const mockSession = createMockSessionForLoopRecovery({ compactResult: null });
|
||||
const { executor } = setupExecutorWithActiveSession(mockSession);
|
||||
|
||||
const result = await executor.handleLoopDetected({
|
||||
taskId: "FN-001",
|
||||
reason: "loop",
|
||||
noProgressMs: 600000,
|
||||
inactivityMs: 0,
|
||||
activitySinceProgress: 100,
|
||||
shouldRequeue: true,
|
||||
});
|
||||
|
||||
expect(result).toBe(false);
|
||||
});
|
||||
});
|
||||
|
||||
@@ -5,7 +5,7 @@ import type { TaskStore, Task, TaskDetail, StepStatus, Settings, WorkflowStep, M
|
||||
import { findWorktreeUser } from "./merger.js";
|
||||
import { generateWorktreeName, slugify } from "./worktree-names.js";
|
||||
import { Type, type Static } from "@mariozechner/pi-ai";
|
||||
import { createKbAgent, describeModel, promptWithFallback } from "./pi.js";
|
||||
import { createKbAgent, describeModel, promptWithFallback, compactSessionContext } from "./pi.js";
|
||||
import { reviewStep, type ReviewVerdict } from "./reviewer.js";
|
||||
import { SessionManager, type ToolDefinition, type AgentSession } from "@mariozechner/pi-coding-agent";
|
||||
import { PRIORITY_EXECUTE, type AgentSemaphore } from "./concurrency.js";
|
||||
@@ -16,7 +16,8 @@ import { isUsageLimitError, checkSessionError, type UsageLimitPauser } from "./u
|
||||
import { isTransientError } from "./transient-error-detector.js";
|
||||
import { withRateLimitRetry } from "./rate-limit-retry.js";
|
||||
import { computeRecoveryDecision, formatDelay, MAX_RECOVERY_RETRIES } from "./recovery-policy.js";
|
||||
import type { StuckTaskDetector } from "./stuck-task-detector.js";
|
||||
import type { StuckTaskDetector, StuckTaskEvent } from "./stuck-task-detector.js";
|
||||
import { isContextLimitError } from "./context-limit-detector.js";
|
||||
|
||||
// Re-export for backward compatibility (tests import from executor.ts)
|
||||
export { summarizeToolArgs } from "./agent-logger.js";
|
||||
@@ -186,6 +187,10 @@ export class TaskExecutor {
|
||||
private depAborted = new Set<string>();
|
||||
/** Tasks killed by stuck task detector. Value = shouldRequeue (budget not exhausted). */
|
||||
private stuckAborted = new Map<string, boolean>();
|
||||
/** In-memory loop recovery state per task. Keyed by taskId, not persisted.
|
||||
* Tracks compact-and-resume attempt count per execute() lifecycle.
|
||||
* Reset at execute() lifecycle end (finally block). */
|
||||
private loopRecoveryState = new Map<string, { attempts: number; pending: boolean }>();
|
||||
|
||||
/**
|
||||
* @param store — Task store instance (also used to listen for events)
|
||||
@@ -685,6 +690,37 @@ export class TaskExecutor {
|
||||
// the error is stored on session.state.error instead of being thrown.
|
||||
checkSessionError(session);
|
||||
|
||||
// If loop recovery is pending (compact-and-resume was triggered by
|
||||
// handleLoopDetected), consume the pending state and resume with a
|
||||
// deterministic prompt. The session has already been compacted, so
|
||||
// we just need to send a fresh prompt to continue execution.
|
||||
const loopState = this.loopRecoveryState.get(task.id);
|
||||
if (loopState?.pending) {
|
||||
loopState.pending = false;
|
||||
executorLog.log(`${task.id} consuming loop recovery — resuming with fresh context`);
|
||||
await this.store.logEntry(task.id, "Resuming execution after context compaction — taking a different approach");
|
||||
|
||||
// Reset activity tracking so the detector doesn't immediately re-trigger
|
||||
stuckDetector?.recordProgress(task.id);
|
||||
|
||||
const resumePrompt = [
|
||||
"Your conversation was compacted because you were looping without making progress.",
|
||||
"Review the current state of the worktree carefully:",
|
||||
"1. Check `git log --oneline` to see what's already been committed",
|
||||
"2. Read the files you were working on to understand current state",
|
||||
"3. Review the PROMPT.md steps to see which are still pending",
|
||||
"",
|
||||
"Take a DIFFERENT approach from what you were doing before.",
|
||||
"If the current step is complete, call task_update to mark it done and move to the next step.",
|
||||
"If you're stuck on a problem, try a simpler or alternative solution.",
|
||||
"",
|
||||
"Continue the task from where you left off.",
|
||||
].join("\n");
|
||||
|
||||
await promptWithFallback(session, resumePrompt);
|
||||
checkSessionError(session);
|
||||
}
|
||||
|
||||
// If dependency was added during execution, discard worktree and move to triage
|
||||
if (this.depAborted.has(task.id)) {
|
||||
this.depAborted.delete(task.id);
|
||||
@@ -874,8 +910,52 @@ export class TaskExecutor {
|
||||
this.stuckAborted.delete(task.id);
|
||||
executorLog.log(`${task.id} terminated by stuck task detector — will ${stuckRequeue ? "retry" : "not retry (budget exhausted)"}`);
|
||||
} else {
|
||||
// Check if the error is a usage-limit error and trigger global pause
|
||||
if (this.options.usageLimitPauser && isUsageLimitError(err.message)) {
|
||||
// Check if the error is a context-limit error and attempt compact-and-resume
|
||||
// before falling through to the normal failure path. This catches context
|
||||
// overflow errors from the LLM provider that occur during prompt execution.
|
||||
const loopState = this.loopRecoveryState.get(task.id);
|
||||
const loopAttempts = loopState?.attempts ?? 0;
|
||||
|
||||
if (isContextLimitError(err.message) && loopAttempts < 1) {
|
||||
const activeEntry = this.activeSessions.get(task.id);
|
||||
if (activeEntry) {
|
||||
executorLog.log(`${task.id} context limit error — attempting compact-and-resume`);
|
||||
await this.store.logEntry(task.id, `Context limit error — attempting compact-and-resume: ${err.message}`);
|
||||
|
||||
const compactResult = await compactSessionContext(activeEntry.session);
|
||||
if (compactResult) {
|
||||
this.loopRecoveryState.set(task.id, { attempts: loopAttempts + 1, pending: true });
|
||||
executorLog.log(`${task.id} context compaction succeeded — resuming`);
|
||||
|
||||
try {
|
||||
this.options.stuckTaskDetector?.recordProgress(task.id);
|
||||
const resumePrompt = [
|
||||
"Your conversation hit the context window limit and has been compacted.",
|
||||
"Review the current state of the worktree and continue from where you left off.",
|
||||
"Check git log and current files to understand what's already been done.",
|
||||
"Take a different, more efficient approach if needed.",
|
||||
"",
|
||||
"Continue the task.",
|
||||
].join("\n");
|
||||
await promptWithFallback(activeEntry.session, resumePrompt);
|
||||
checkSessionError(activeEntry.session);
|
||||
|
||||
// Check for loop recovery pending from the compact-and-resume
|
||||
const updatedState = this.loopRecoveryState.get(task.id);
|
||||
if (updatedState?.pending) {
|
||||
updatedState.pending = false;
|
||||
await promptWithFallback(activeEntry.session, "Continue working on the remaining steps.");
|
||||
checkSessionError(activeEntry.session);
|
||||
}
|
||||
} catch (resumeErr: any) {
|
||||
// Resume after context compaction failed — fall through to normal failure
|
||||
executorLog.error(`${task.id} resume after context compaction failed: ${resumeErr.message}`);
|
||||
}
|
||||
} else {
|
||||
executorLog.log(`${task.id} context compaction failed — falling through to normal failure`);
|
||||
}
|
||||
}
|
||||
} else if (this.options.usageLimitPauser && isUsageLimitError(err.message)) {
|
||||
await this.options.usageLimitPauser.onUsageLimitHit("executor", task.id, err.message);
|
||||
} else if (isTransientError(err.message)) {
|
||||
// Transient network/infrastructure error — use bounded recovery policy
|
||||
@@ -917,6 +997,10 @@ export class TaskExecutor {
|
||||
} finally {
|
||||
this.executing.delete(task.id);
|
||||
|
||||
// Reset loop recovery state at end of execute() lifecycle.
|
||||
// State is in-memory and per-run — should not persist across attempts.
|
||||
this.loopRecoveryState.delete(task.id);
|
||||
|
||||
// Requeue stuck-killed task AFTER this.executing is cleared.
|
||||
// This prevents the race where the scheduler re-dispatches the task
|
||||
// (via task:moved → execute()) while the old execution guard is still set,
|
||||
@@ -2152,6 +2236,72 @@ If issues are found that need attention, describe them clearly.`;
|
||||
this.stuckAborted.set(taskId, shouldRequeue);
|
||||
}
|
||||
|
||||
/**
|
||||
* Handle a loop-detected event from the stuck task detector.
|
||||
* Attempts an in-process compact-and-resume before falling back to kill/requeue.
|
||||
*
|
||||
* This method is the `onLoopDetected` callback wired through the dashboard.
|
||||
* It:
|
||||
* 1. Checks if the task has an active session
|
||||
* 2. Rejects if the one-attempt ceiling has been reached
|
||||
* 3. Calls `compactSessionContext()` to compact the conversation
|
||||
* 4. Sets recovery-pending state so the execution flow can resume
|
||||
*
|
||||
* @returns true if the executor accepted recovery ownership (detector skips kill),
|
||||
* false if recovery should not be attempted (detector proceeds with kill/requeue)
|
||||
*/
|
||||
async handleLoopDetected(event: StuckTaskEvent): Promise<boolean> {
|
||||
const { taskId } = event;
|
||||
const activeEntry = this.activeSessions.get(taskId);
|
||||
|
||||
// No active session — can't compact, let detector kill/requeue
|
||||
if (!activeEntry) {
|
||||
executorLog.log(`${taskId} loop detected but no active session — falling back to kill/requeue`);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Check attempt ceiling (max 1 compact-and-resume per execute() lifecycle)
|
||||
const state = this.loopRecoveryState.get(taskId);
|
||||
if (state && state.attempts >= 1) {
|
||||
executorLog.log(`${taskId} loop detected but compact ceiling reached — falling back to kill/requeue`);
|
||||
return false;
|
||||
}
|
||||
|
||||
// Attempt compaction
|
||||
const attempt = (state?.attempts ?? 0) + 1;
|
||||
executorLog.log(`${taskId} loop detected (attempt ${attempt}) — attempting compact-and-resume`);
|
||||
await this.store.logEntry(taskId, `Loop detected (${event.activitySinceProgress} events since last progress) — attempting compact-and-resume (attempt ${attempt})`);
|
||||
|
||||
const compactResult = await compactSessionContext(activeEntry.session);
|
||||
if (!compactResult) {
|
||||
executorLog.log(`${taskId} compaction failed or unavailable — falling back to kill/requeue`);
|
||||
await this.store.logEntry(taskId, "Context compaction failed or unavailable — falling back to kill/requeue");
|
||||
return false;
|
||||
}
|
||||
|
||||
executorLog.log(`${taskId} compaction succeeded (freed ${compactResult.tokensBefore} tokens) — setting recovery-pending`);
|
||||
await this.store.logEntry(taskId, `Context compacted successfully — will resume with fresh context`);
|
||||
|
||||
// Mark recovery-pending so the execution flow can consume it
|
||||
this.loopRecoveryState.set(taskId, { attempts: attempt, pending: true });
|
||||
|
||||
// Steer the session with a resume prompt to break the loop
|
||||
try {
|
||||
await activeEntry.session.steer(
|
||||
"⚠️ Loop detected: you were repeating actions without making progress. " +
|
||||
"The conversation has been compacted. Review the current state carefully, " +
|
||||
"check what's already been done (git log, file contents), and take a different " +
|
||||
"approach. Do NOT repeat the same actions. Advance to the next step if the " +
|
||||
"current work is complete.",
|
||||
);
|
||||
} catch (err: any) {
|
||||
executorLog.error(`${taskId} failed to steer after compaction: ${err.message}`);
|
||||
// Recovery-pending is still set — the execution flow will handle it
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
getWorktreePath(taskId: string): string | undefined {
|
||||
return this.activeWorktrees.get(taskId);
|
||||
}
|
||||
|
||||
@@ -1,5 +1,5 @@
|
||||
import { describe, it, expect } from "vitest";
|
||||
import { describeModel } from "./pi.js";
|
||||
import { describeModel, compactSessionContext, COMPACTION_FALLBACK_INSTRUCTIONS } from "./pi.js";
|
||||
import type { AgentSession } from "@mariozechner/pi-coding-agent";
|
||||
|
||||
describe("describeModel", () => {
|
||||
@@ -35,3 +35,83 @@ describe("describeModel", () => {
|
||||
expect(describeModel(fakeSession)).toBe("openai/gpt-4o");
|
||||
});
|
||||
});
|
||||
|
||||
describe("COMPACTION_FALLBACK_INSTRUCTIONS", () => {
|
||||
it("is a non-empty string", () => {
|
||||
expect(COMPACTION_FALLBACK_INSTRUCTIONS).toBeTruthy();
|
||||
expect(typeof COMPACTION_FALLBACK_INSTRUCTIONS).toBe("string");
|
||||
expect(COMPACTION_FALLBACK_INSTRUCTIONS.length).toBeGreaterThan(0);
|
||||
});
|
||||
|
||||
it("mentions summarizing completed steps", () => {
|
||||
expect(COMPACTION_FALLBACK_INSTRUCTIONS).toContain("completed steps");
|
||||
});
|
||||
});
|
||||
|
||||
describe("compactSessionContext", () => {
|
||||
it("returns null when session does not have compact method", async () => {
|
||||
const session = {} as AgentSession;
|
||||
const result = await compactSessionContext(session);
|
||||
expect(result).toBeNull();
|
||||
});
|
||||
|
||||
it("calls session.compact with default instructions when no custom instructions provided", async () => {
|
||||
const compact = async (instructions: string) => ({
|
||||
summary: "Compacted",
|
||||
tokensBefore: 100000,
|
||||
});
|
||||
const session = { compact } as unknown as AgentSession;
|
||||
|
||||
const result = await compactSessionContext(session);
|
||||
|
||||
expect(result).toEqual({
|
||||
summary: "Compacted",
|
||||
tokensBefore: 100000,
|
||||
});
|
||||
});
|
||||
|
||||
it("calls session.compact with custom instructions when provided", async () => {
|
||||
let capturedInstructions: string | undefined;
|
||||
const compact = async (instructions: string) => {
|
||||
capturedInstructions = instructions;
|
||||
return { summary: "Custom", tokensBefore: 50000 };
|
||||
};
|
||||
const session = { compact } as unknown as AgentSession;
|
||||
|
||||
const result = await compactSessionContext(session, "Focus on step 3");
|
||||
|
||||
expect(capturedInstructions).toBe("Focus on step 3");
|
||||
expect(result).toEqual({
|
||||
summary: "Custom",
|
||||
tokensBefore: 50000,
|
||||
});
|
||||
});
|
||||
|
||||
it("returns null when session.compact throws", async () => {
|
||||
const compact = async () => { throw new Error("compaction failed"); };
|
||||
const session = { compact } as unknown as AgentSession;
|
||||
|
||||
const result = await compactSessionContext(session);
|
||||
|
||||
expect(result).toBeNull();
|
||||
});
|
||||
|
||||
it("returns null when session.compact returns null", async () => {
|
||||
const compact = async () => null;
|
||||
const session = { compact } as unknown as AgentSession;
|
||||
|
||||
const result = await compactSessionContext(session);
|
||||
|
||||
expect(result).toBeNull();
|
||||
});
|
||||
|
||||
it("returns result with empty summary when session.compact returns object without summary", async () => {
|
||||
const compact = async () => ({});
|
||||
const session = { compact } as unknown as AgentSession;
|
||||
|
||||
const result = await compactSessionContext(session);
|
||||
|
||||
// Should still return a result with empty summary since the guard checks for object
|
||||
expect(result).toEqual({ summary: "", tokensBefore: 0 });
|
||||
});
|
||||
});
|
||||
|
||||
@@ -59,6 +59,56 @@ export function describeModel(session: AgentSession): string {
|
||||
return `${model.provider}/${model.id}`;
|
||||
}
|
||||
|
||||
/**
|
||||
* Default instructions used when calling `session.compact()` for loop recovery.
|
||||
* These guide the compaction summary to preserve essential context while
|
||||
* freeing up the context window for continued work.
|
||||
*/
|
||||
export const COMPACTION_FALLBACK_INSTRUCTIONS = [
|
||||
"Summarize all completed steps concisely.",
|
||||
"Preserve the current step number and any in-progress work details.",
|
||||
"Keep references to key files, decisions, and error states.",
|
||||
"Discard verbose tool output, repeated attempts, and exploration history.",
|
||||
].join(" ");
|
||||
|
||||
/**
|
||||
* Compact an agent session's context to free up the context window.
|
||||
*
|
||||
* Uses the SDK's native `session.compact()` method when available (the
|
||||
* preferred path — it produces structured, LLM-generated summaries).
|
||||
*
|
||||
* @param session — The agent session to compact
|
||||
* @param customInstructions — Optional instructions for the compaction summary.
|
||||
* When not provided, uses COMPACTION_FALLBACK_INSTRUCTIONS.
|
||||
* @returns The compaction result with summary and token metrics, or null if
|
||||
* compaction was not available or failed.
|
||||
*/
|
||||
export async function compactSessionContext(
|
||||
session: AgentSession,
|
||||
customInstructions?: string,
|
||||
): Promise<{ summary: string; tokensBefore: number } | null> {
|
||||
const instructions = customInstructions ?? COMPACTION_FALLBACK_INSTRUCTIONS;
|
||||
|
||||
// Check if session.compact is available (runtime capability detection)
|
||||
if (typeof (session as any).compact !== "function") {
|
||||
return null;
|
||||
}
|
||||
|
||||
try {
|
||||
const result = await (session as any).compact(instructions);
|
||||
if (result && typeof result === "object") {
|
||||
return {
|
||||
summary: result.summary ?? "",
|
||||
tokensBefore: result.tokensBefore ?? 0,
|
||||
};
|
||||
}
|
||||
return null;
|
||||
} catch {
|
||||
// Compaction failed — return null so caller can fall through to kill/requeue
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
export interface AgentOptions {
|
||||
cwd: string;
|
||||
systemPrompt: string;
|
||||
|
||||
@@ -808,4 +808,174 @@ describe("StuckTaskDetector", () => {
|
||||
vi.useRealTimers();
|
||||
});
|
||||
});
|
||||
|
||||
describe("onLoopDetected pre-kill callback", () => {
|
||||
it("calls onLoopDetected before onStuck when reason is loop", async () => {
|
||||
const callOrder: string[] = [];
|
||||
const onLoopDetected = vi.fn(async () => { callOrder.push("onLoopDetected"); return false; });
|
||||
const onStuck = vi.fn(() => { callOrder.push("onStuck"); });
|
||||
const customDetector = new StuckTaskDetector(store, { onLoopDetected, onStuck });
|
||||
const session = createMockSession();
|
||||
|
||||
vi.useFakeTimers({ shouldAdvanceTime: true });
|
||||
customDetector.trackTask("FN-001", session);
|
||||
vi.advanceTimersByTime(61000);
|
||||
for (let i = 0; i < 80; i++) {
|
||||
customDetector.recordActivity("FN-001");
|
||||
}
|
||||
|
||||
await customDetector.killAndRetry("FN-001", 60000);
|
||||
|
||||
// onLoopDetected should be called first
|
||||
expect(onLoopDetected).toHaveBeenCalledTimes(1);
|
||||
expect(onLoopDetected).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ taskId: "FN-001", reason: "loop" }),
|
||||
);
|
||||
expect(callOrder).toEqual(["onLoopDetected", "onStuck"]);
|
||||
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("skips kill/requeue when onLoopDetected returns true", async () => {
|
||||
const onLoopDetected = vi.fn().mockResolvedValue(true);
|
||||
const onStuck = vi.fn();
|
||||
const customDetector = new StuckTaskDetector(store, { onLoopDetected, onStuck });
|
||||
const session = createMockSession();
|
||||
|
||||
vi.useFakeTimers({ shouldAdvanceTime: true });
|
||||
customDetector.trackTask("FN-001", session);
|
||||
vi.advanceTimersByTime(61000);
|
||||
for (let i = 0; i < 80; i++) {
|
||||
customDetector.recordActivity("FN-001");
|
||||
}
|
||||
|
||||
await customDetector.killAndRetry("FN-001", 60000);
|
||||
|
||||
// onLoopDetected accepted recovery — no kill/requeue
|
||||
expect(onLoopDetected).toHaveBeenCalledTimes(1);
|
||||
expect(onStuck).not.toHaveBeenCalled();
|
||||
expect(session.dispose).not.toHaveBeenCalled();
|
||||
expect(customDetector.trackedCount).toBe(0); // untracked
|
||||
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("does NOT call onLoopDetected when reason is inactivity", async () => {
|
||||
const onLoopDetected = vi.fn().mockResolvedValue(true);
|
||||
const onStuck = vi.fn();
|
||||
const customDetector = new StuckTaskDetector(store, { onLoopDetected, onStuck });
|
||||
const session = createMockSession();
|
||||
|
||||
vi.useFakeTimers({ shouldAdvanceTime: true });
|
||||
customDetector.trackTask("FN-001", session);
|
||||
vi.advanceTimersByTime(61000);
|
||||
// No activity — pure inactivity
|
||||
|
||||
await customDetector.killAndRetry("FN-001", 60000);
|
||||
|
||||
// onLoopDetected should NOT be called for inactivity
|
||||
expect(onLoopDetected).not.toHaveBeenCalled();
|
||||
// Normal kill path should still execute
|
||||
expect(onStuck).toHaveBeenCalled();
|
||||
expect(session.dispose).toHaveBeenCalled();
|
||||
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("falls through to normal kill when onLoopDetected returns false", async () => {
|
||||
const onLoopDetected = vi.fn().mockResolvedValue(false);
|
||||
const onStuck = vi.fn();
|
||||
const customDetector = new StuckTaskDetector(store, { onLoopDetected, onStuck });
|
||||
const session = createMockSession();
|
||||
|
||||
vi.useFakeTimers({ shouldAdvanceTime: true });
|
||||
customDetector.trackTask("FN-001", session);
|
||||
vi.advanceTimersByTime(61000);
|
||||
for (let i = 0; i < 80; i++) {
|
||||
customDetector.recordActivity("FN-001");
|
||||
}
|
||||
|
||||
await customDetector.killAndRetry("FN-001", 60000);
|
||||
|
||||
// Callback declined — normal kill path
|
||||
expect(onLoopDetected).toHaveBeenCalledTimes(1);
|
||||
expect(onStuck).toHaveBeenCalled();
|
||||
expect(session.dispose).toHaveBeenCalled();
|
||||
expect(store.moveTask).not.toHaveBeenCalled(); // executor handles move
|
||||
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("falls through to normal kill when onLoopDetected throws", async () => {
|
||||
const onLoopDetected = vi.fn().mockRejectedValue(new Error("callback exploded"));
|
||||
const onStuck = vi.fn();
|
||||
const customDetector = new StuckTaskDetector(store, { onLoopDetected, onStuck });
|
||||
const session = createMockSession();
|
||||
|
||||
vi.useFakeTimers({ shouldAdvanceTime: true });
|
||||
customDetector.trackTask("FN-001", session);
|
||||
vi.advanceTimersByTime(61000);
|
||||
for (let i = 0; i < 80; i++) {
|
||||
customDetector.recordActivity("FN-001");
|
||||
}
|
||||
|
||||
await customDetector.killAndRetry("FN-001", 60000);
|
||||
|
||||
// Error in callback — normal kill path
|
||||
expect(onLoopDetected).toHaveBeenCalledTimes(1);
|
||||
expect(onStuck).toHaveBeenCalled();
|
||||
expect(session.dispose).toHaveBeenCalled();
|
||||
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("does not call onLoopDetected when callback is not registered", async () => {
|
||||
// No onLoopDetected callback — normal kill path
|
||||
const onStuck = vi.fn();
|
||||
const customDetector = new StuckTaskDetector(store, { onStuck });
|
||||
const session = createMockSession();
|
||||
|
||||
vi.useFakeTimers({ shouldAdvanceTime: true });
|
||||
customDetector.trackTask("FN-001", session);
|
||||
vi.advanceTimersByTime(61000);
|
||||
for (let i = 0; i < 80; i++) {
|
||||
customDetector.recordActivity("FN-001");
|
||||
}
|
||||
|
||||
await customDetector.killAndRetry("FN-001", 60000);
|
||||
|
||||
// No callback registered — goes straight to onStuck + dispose
|
||||
expect(onStuck).toHaveBeenCalled();
|
||||
expect(session.dispose).toHaveBeenCalled();
|
||||
|
||||
vi.useRealTimers();
|
||||
});
|
||||
|
||||
it("receives correct event payload with shouldRequeue from beforeRequeue", async () => {
|
||||
const beforeRequeue = vi.fn().mockResolvedValue(false); // budget exhausted
|
||||
const onLoopDetected = vi.fn().mockResolvedValue(true);
|
||||
const customDetector = new StuckTaskDetector(store, { beforeRequeue, onLoopDetected });
|
||||
const session = createMockSession();
|
||||
|
||||
vi.useFakeTimers({ shouldAdvanceTime: true });
|
||||
customDetector.trackTask("FN-001", session);
|
||||
vi.advanceTimersByTime(61000);
|
||||
for (let i = 0; i < 80; i++) {
|
||||
customDetector.recordActivity("FN-001");
|
||||
}
|
||||
|
||||
await customDetector.killAndRetry("FN-001", 60000);
|
||||
|
||||
// onLoopDetected should receive shouldRequeue=false from beforeRequeue
|
||||
expect(onLoopDetected).toHaveBeenCalledWith(
|
||||
expect.objectContaining({
|
||||
taskId: "FN-001",
|
||||
reason: "loop",
|
||||
shouldRequeue: false,
|
||||
}),
|
||||
);
|
||||
|
||||
vi.useRealTimers();
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@@ -66,6 +66,18 @@ export interface StuckTaskDetectorOptions {
|
||||
* (caller is responsible for marking the task as terminally failed).
|
||||
* Used by SelfHealingManager to enforce stuck kill budgets. */
|
||||
beforeRequeue?: (taskId: string) => Promise<boolean>;
|
||||
/** Pre-kill callback invoked ONLY when reason is "loop".
|
||||
* Called BEFORE session.dispose() / moveTask("todo") so the caller can
|
||||
* attempt in-process recovery (e.g. compact-and-resume) without killing
|
||||
* the agent session.
|
||||
*
|
||||
* Return `true` to signal "executor accepted ownership of recovery for this
|
||||
* run" — the detector will skip dispose/requeue and remove the task from
|
||||
* tracking (the caller is now responsible for the task's fate).
|
||||
* Return `false` to let the detector proceed with the normal kill/requeue path.
|
||||
*
|
||||
* Errors in this callback fall through to the normal kill path (treated as `false`). */
|
||||
onLoopDetected?: (event: StuckTaskEvent) => Promise<boolean>;
|
||||
}
|
||||
|
||||
export class StuckTaskDetector {
|
||||
@@ -74,6 +86,7 @@ export class StuckTaskDetector {
|
||||
private pollIntervalMs: number;
|
||||
private onStuck?: (event: StuckTaskEvent) => void;
|
||||
private beforeRequeue?: (taskId: string) => Promise<boolean>;
|
||||
private onLoopDetected?: (event: StuckTaskEvent) => Promise<boolean>;
|
||||
|
||||
constructor(
|
||||
private store: TaskStore,
|
||||
@@ -82,6 +95,7 @@ export class StuckTaskDetector {
|
||||
this.pollIntervalMs = options.pollIntervalMs ?? 30_000;
|
||||
this.onStuck = options.onStuck;
|
||||
this.beforeRequeue = options.beforeRequeue;
|
||||
this.onLoopDetected = options.onLoopDetected;
|
||||
}
|
||||
|
||||
/**
|
||||
@@ -284,6 +298,32 @@ export class StuckTaskDetector {
|
||||
shouldRequeue,
|
||||
};
|
||||
|
||||
// ── Pre-kill loop interception ──────────────────────────────────
|
||||
// When reason is "loop" and an onLoopDetected callback is registered,
|
||||
// give the caller a chance to handle recovery in-process (e.g.
|
||||
// compact-and-resume) before falling through to the kill/requeue path.
|
||||
//
|
||||
// If the callback returns true, the caller owns the task — we skip
|
||||
// dispose/requeue and just untrack. Errors fall through to normal kill.
|
||||
if (reason === "loop" && this.onLoopDetected) {
|
||||
try {
|
||||
const handled = await this.onLoopDetected(event);
|
||||
if (handled) {
|
||||
stuckLog.log(
|
||||
`${taskId} loop recovery accepted by onLoopDetected callback — ` +
|
||||
`skipping kill/requeue (caller owns recovery)`,
|
||||
);
|
||||
// The caller is now responsible for the task; remove from tracking
|
||||
// so we don't double-trigger.
|
||||
this.tracked.delete(taskId);
|
||||
return;
|
||||
}
|
||||
} catch (err) {
|
||||
stuckLog.error(`onLoopDetected callback failed for ${taskId}:`, err);
|
||||
// Fall through to normal kill path
|
||||
}
|
||||
}
|
||||
|
||||
// Notify listeners before disposing the session so executor cleanup can
|
||||
// mark the abort as intentional before the disposed session unwinds.
|
||||
this.onStuck?.(event);
|
||||
|
||||
Reference in New Issue
Block a user