feat: hang detection and resumable task sessions
Some checks failed
port-to-omp / port (push) Failing after 4s
Some checks failed
port-to-omp / port (push) Failing after 4s
- runAgentSession gains an inactivity watchdog (execution.inactivityTimeoutMs, default 0 = off): when no session event arrives within the window, the session is aborted (agent abort + bash subprocess kill) with a clear "inactivity timeout" error - agent sessions persist to .ralpi/sessions/*.jsonl; resume reopens the JSONL via SessionManager.open so an interrupted task continues with its prior conversation instead of restarting from scratch - progress.json tracks sessionFile per task (persisted at session creation, so kill/reload mid-run is resumable); the first attempt after resume reuses it, failover retries stay fresh; corrupt/missing files fall back to a fresh session with a warning - bump version to 0.6.0
This commit is contained in:
@@ -245,6 +245,13 @@ export async function runTask(
|
||||
/** Review feedback from a rejected review — injected when re-executing
|
||||
* a task in review-gated mode so the agent knows what to fix. */
|
||||
reviewFeedback?: ReviewResult,
|
||||
/** Session JSONL from a prior interrupted run of this task. When set and
|
||||
* readable, the agent session reopens it and continues from the prior
|
||||
* conversation instead of starting fresh. */
|
||||
resumeSessionFile?: string,
|
||||
/** Called with the session file path as soon as the agent session is
|
||||
* created, so the caller can persist it for a later resume. */
|
||||
onSessionFile?: (sessionFile: string) => void,
|
||||
): Promise<{
|
||||
success: boolean;
|
||||
reflection?: Reflection;
|
||||
@@ -254,6 +261,12 @@ export async function runTask(
|
||||
outputPreview?: string;
|
||||
commitMessages?: string[];
|
||||
commitSummary?: string;
|
||||
/** Path to the JSONL session file backing this run (for resume). */
|
||||
sessionFile?: string;
|
||||
/** True when a resume was requested but the session could not be opened
|
||||
* from the file — the caller should clear the stored session file so
|
||||
* retries start fresh. */
|
||||
resumeFailed?: boolean;
|
||||
}> {
|
||||
const startMs = Date.now();
|
||||
|
||||
@@ -385,6 +398,9 @@ export async function runTask(
|
||||
config.thinkingLevel,
|
||||
false, // noSkills — task sessions need skills
|
||||
(ctx.modelRegistry as any).runtime as ModelRuntime,
|
||||
config.execution.inactivityTimeoutMs,
|
||||
resumeSessionFile,
|
||||
onSessionFile,
|
||||
);
|
||||
|
||||
const durationMs = Date.now() - startMs;
|
||||
@@ -409,6 +425,8 @@ export async function runTask(
|
||||
success: false,
|
||||
error: output.error,
|
||||
durationMs,
|
||||
sessionFile: output.sessionFile,
|
||||
resumeFailed: output.resumeFailed,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -439,6 +457,7 @@ export async function runTask(
|
||||
outputPreview,
|
||||
commitMessages,
|
||||
commitSummary,
|
||||
sessionFile: output.sessionFile,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -825,6 +844,12 @@ async function executeTask(
|
||||
: null;
|
||||
const worktreeDir = wt?.dir ?? projectDir;
|
||||
|
||||
// Session file from a prior interrupted run of this task. The first
|
||||
// attempt reopens it so the agent continues with its prior conversation
|
||||
// (tool calls, findings) instead of restarting from scratch; failover
|
||||
// retries start fresh.
|
||||
const resumeSessionFile = progress.getSessionFile(task.id);
|
||||
|
||||
while (modelAttempt < maxModelAttempts) {
|
||||
// Model advancement happens in the cycling branch below (not here) so a
|
||||
// same-model retry `continue` doesn't re-advance and accidentally swap
|
||||
@@ -885,6 +910,11 @@ async function executeTask(
|
||||
currentModel,
|
||||
batchRender,
|
||||
priorReview,
|
||||
modelAttempt === 0 && sameModelAttempt === 0
|
||||
? resumeSessionFile
|
||||
: undefined,
|
||||
(sessionFile) =>
|
||||
progress.setSessionFile(task.id, sessionFile),
|
||||
);
|
||||
|
||||
if (result.success) {
|
||||
@@ -1009,6 +1039,7 @@ async function executeTask(
|
||||
`review-${task.id}`,
|
||||
config.execution.reviewTimeoutMs,
|
||||
reviewModels,
|
||||
config.execution.inactivityTimeoutMs,
|
||||
);
|
||||
|
||||
if (!reviewResult.success) {
|
||||
@@ -1301,6 +1332,7 @@ async function executeTask(
|
||||
finalCommitSummary,
|
||||
finalReview,
|
||||
reviewRetries,
|
||||
result.sessionFile,
|
||||
);
|
||||
// Auto-update the PRD source file checkbox
|
||||
try {
|
||||
@@ -1314,6 +1346,12 @@ async function executeTask(
|
||||
|
||||
// Agent session failed (provider error).
|
||||
// Pi's built-in in-call retry already exhausted for this attempt.
|
||||
// A resumed session that couldn't be opened (corrupt/missing
|
||||
// JSONL) must not be retried — forget it so later attempts (and
|
||||
// future resumes) start fresh.
|
||||
if (result.resumeFailed) {
|
||||
progress.setSessionFile(task.id, undefined);
|
||||
}
|
||||
// Reattempt on the SAME model a few more times before cycling — a
|
||||
// transient outage can outlast pi's per-prompt backoff window.
|
||||
sameModelAttempt++;
|
||||
@@ -1446,6 +1484,8 @@ async function runFollowUpSession(
|
||||
widgetKeySuffix: string,
|
||||
timeoutMs: number,
|
||||
models: unknown[],
|
||||
/** Inactivity timeout for the session (see runAgentSession). */
|
||||
inactivityTimeoutMs = 0,
|
||||
): Promise<{
|
||||
result: Awaited<ReturnType<typeof runAgentSession>>;
|
||||
toolCalls: ToolCallEntry[];
|
||||
@@ -1544,6 +1584,7 @@ async function runFollowUpSession(
|
||||
config.thinkingLevel,
|
||||
false, // noSkills=false — follow-up sessions load skills too
|
||||
(ctx.modelRegistry as any).runtime as ModelRuntime,
|
||||
inactivityTimeoutMs,
|
||||
);
|
||||
|
||||
if (result.success) break;
|
||||
@@ -1690,6 +1731,7 @@ async function runCommitSession(
|
||||
`commit-${task.id}`,
|
||||
config.execution.commitTimeoutMs,
|
||||
commitModels,
|
||||
config.execution.inactivityTimeoutMs,
|
||||
);
|
||||
|
||||
if (commitResult.success) {
|
||||
@@ -1862,6 +1904,7 @@ async function resolveConflictsSession(
|
||||
`resolve-${task.id}`,
|
||||
config.execution.commitTimeoutMs,
|
||||
models,
|
||||
config.execution.inactivityTimeoutMs,
|
||||
);
|
||||
|
||||
if (!result.success) {
|
||||
|
||||
@@ -185,11 +185,13 @@ export class ProgressTracker {
|
||||
}
|
||||
|
||||
/** Mark a task as in progress */
|
||||
markInProgress(taskId: string): void {
|
||||
markInProgress(taskId: string, sessionFile?: string): void {
|
||||
const prd = this.getPRD();
|
||||
this.ensureTask(prd, taskId);
|
||||
prd.tasks[taskId].status = "in_progress";
|
||||
prd.tasks[taskId].startedAt = new Date().toISOString();
|
||||
if (sessionFile !== undefined)
|
||||
prd.tasks[taskId].sessionFile = sessionFile;
|
||||
this.save();
|
||||
}
|
||||
|
||||
@@ -204,6 +206,7 @@ export class ProgressTracker {
|
||||
commitSummary?: string,
|
||||
review?: ReviewResult,
|
||||
reviewRetries?: number,
|
||||
sessionFile?: string,
|
||||
): void {
|
||||
const prd = this.getPRD();
|
||||
this.ensureTask(prd, taskId);
|
||||
@@ -218,6 +221,7 @@ export class ProgressTracker {
|
||||
if (review) prd.tasks[taskId].review = review;
|
||||
if (reviewRetries !== undefined)
|
||||
prd.tasks[taskId].reviewRetries = reviewRetries;
|
||||
if (sessionFile !== undefined) prd.tasks[taskId].sessionFile = sessionFile;
|
||||
this.save();
|
||||
}
|
||||
|
||||
@@ -236,6 +240,22 @@ export class ProgressTracker {
|
||||
return prd.tasks[taskId]?.status ?? "pending";
|
||||
}
|
||||
|
||||
/** Get the persisted session file path for a task (for resume), if any. */
|
||||
getSessionFile(taskId: string): string | undefined {
|
||||
const prd = this.getPRD();
|
||||
return prd.tasks[taskId]?.sessionFile;
|
||||
}
|
||||
|
||||
/** Persist the session file path for a task without changing its status.
|
||||
* Called as soon as an agent session is created so an interrupted run
|
||||
* can be resumed from the JSONL history. Pass undefined to clear. */
|
||||
setSessionFile(taskId: string, sessionFile: string | undefined): void {
|
||||
const prd = this.getPRD();
|
||||
this.ensureTask(prd, taskId);
|
||||
prd.tasks[taskId].sessionFile = sessionFile;
|
||||
this.save();
|
||||
}
|
||||
|
||||
/** Get IDs of all completed tasks */
|
||||
getCompletedTaskIds(): string[] {
|
||||
const prd = this.getPRD();
|
||||
|
||||
10
src/types.ts
10
src/types.ts
@@ -161,6 +161,10 @@ export interface TaskProgressInfo {
|
||||
commitSummary?: string;
|
||||
/** Number of review-fix re-execution attempts made (review-gated mode) */
|
||||
reviewRetries?: number;
|
||||
/** Path to the JSONL session file backing this task's agent session,
|
||||
* persisted so a resume can reopen it and continue where the
|
||||
* interrupted session left off. */
|
||||
sessionFile?: string;
|
||||
}
|
||||
|
||||
export interface ProgressState {
|
||||
@@ -205,6 +209,11 @@ export interface RalpiConfig {
|
||||
execution: {
|
||||
/** Task execution timeout in milliseconds */
|
||||
timeoutMs: number;
|
||||
/** Inactivity timeout in milliseconds — if no agent session event
|
||||
* arrives within this window, the task is considered hung (e.g. a
|
||||
* bash subprocess that never returns) and the session is aborted.
|
||||
* 0 = disabled. */
|
||||
inactivityTimeoutMs: number;
|
||||
/** Maximum parallel tasks (0 = unlimited) */
|
||||
maxParallel: number;
|
||||
/** Round-robin model list for parallel tasks (empty = inherit parent model) */
|
||||
@@ -301,6 +310,7 @@ export const DEFAULT_CONFIG: RalpiConfig = {
|
||||
},
|
||||
execution: {
|
||||
timeoutMs: 0, // 0 = inherit Pi's own defaults (no ralpi-level timeout)
|
||||
inactivityTimeoutMs: 0, // 0 = disabled (no inactivity hang detection)
|
||||
maxParallel: 3,
|
||||
models: [],
|
||||
autoCommit: true,
|
||||
|
||||
86
src/utils.ts
86
src/utils.ts
@@ -611,6 +611,18 @@ export async function runAgentSession(
|
||||
* rate-limit normalization) are available. When omitted, the SDK creates
|
||||
* a fresh runtime from models.json only — extension providers are lost. */
|
||||
modelRuntime?: ModelRuntime,
|
||||
/** Inactivity timeout in milliseconds — if no agent session event arrives
|
||||
* within this window, the task is considered hung (e.g. a bash subprocess
|
||||
* that never returns) and the session is aborted. 0 = disabled. */
|
||||
inactivityTimeoutMs = 0,
|
||||
/** Existing session JSONL file to resume. The session reopens the file and
|
||||
* appends to it, so the agent sees the full prior conversation and can
|
||||
* continue rather than redo prior tool calls. When the file is missing,
|
||||
* a fresh session is created instead (with a warning). */
|
||||
resumeSessionFile?: string,
|
||||
/** Called with the session file path as soon as the session is created, so
|
||||
* callers can persist it for resume before the session completes. */
|
||||
onSessionFile?: (sessionFile: string) => void,
|
||||
): Promise<{
|
||||
success: boolean;
|
||||
text: string;
|
||||
@@ -618,6 +630,13 @@ export async function runAgentSession(
|
||||
toolUsage: ToolUsage;
|
||||
stopReason?: string;
|
||||
events: AgentSessionEvent[];
|
||||
/** Path to the JSONL session file backing this session (set once the
|
||||
* session is created; enables resume). */
|
||||
sessionFile?: string;
|
||||
/** True when a resume was requested but the session could not be created
|
||||
* from the file (corrupt/unreadable JSONL). Callers should clear the
|
||||
* stored session file so retries start fresh. */
|
||||
resumeFailed?: boolean;
|
||||
}> {
|
||||
const toolUsage: ToolUsage = {
|
||||
read: 0,
|
||||
@@ -638,6 +657,16 @@ export async function runAgentSession(
|
||||
session?: Awaited<ReturnType<typeof createAgentSession>>["session"];
|
||||
} = {};
|
||||
|
||||
let sessionFile: string | undefined;
|
||||
let sessionCreated = false;
|
||||
// Inactivity watchdog: aborts the session when no events arrive within
|
||||
// inactivityTimeoutMs. The SDK emits an event for every tool start/end/
|
||||
// update and message start/end, so silence means the agent is stuck
|
||||
// (typically a hung bash subprocess producing no output).
|
||||
let inactivityInterval: NodeJS.Timeout | null = null;
|
||||
let inactivityAborted = false;
|
||||
let lastEventTime = 0;
|
||||
|
||||
try {
|
||||
// Loop sessions load the full normal pi context: extensions (so all
|
||||
// extension-provided tools register), skills, and project context
|
||||
@@ -653,9 +682,30 @@ export async function runAgentSession(
|
||||
});
|
||||
await loader.reload();
|
||||
|
||||
// Persist sessions under the ralpi project's `.ralpi/sessions/` so they
|
||||
// survive worktree removal and are findable from the main repo on resume.
|
||||
// Worktrees live inside `<project>/.ralpi/worktrees/...`, so walking up
|
||||
// from the agent's cwd always finds the main project's `.ralpi` first.
|
||||
const ralpiDir = findRalpiDir(cwd);
|
||||
const sessionDir = ralpiDir
|
||||
? path.join(ralpiDir, ".ralpi", "sessions")
|
||||
: path.join(cwd, ".ralpi", "sessions");
|
||||
|
||||
let sessionManager: SessionManager;
|
||||
if (resumeSessionFile && fs.existsSync(resumeSessionFile)) {
|
||||
sessionManager = SessionManager.open(resumeSessionFile, sessionDir, cwd);
|
||||
} else {
|
||||
if (resumeSessionFile) {
|
||||
console.warn(
|
||||
`[ralpi] resume session file not found (${resumeSessionFile}) — starting a fresh session`,
|
||||
);
|
||||
}
|
||||
sessionManager = SessionManager.create(cwd, sessionDir);
|
||||
}
|
||||
|
||||
const result = await createAgentSession({
|
||||
cwd,
|
||||
sessionManager: SessionManager.inMemory(),
|
||||
sessionManager,
|
||||
resourceLoader: loader,
|
||||
settingsManager: SettingsManager.create(cwd, getAgentDir()),
|
||||
modelRuntime,
|
||||
@@ -663,8 +713,12 @@ export async function runAgentSession(
|
||||
model: model as any,
|
||||
thinkingLevel: thinkingLevel as any,
|
||||
});
|
||||
sessionCreated = true;
|
||||
sessionRef.session = result.session;
|
||||
|
||||
sessionFile = result.session.sessionFile;
|
||||
if (sessionFile) onSessionFile?.(sessionFile);
|
||||
|
||||
// Wire external abort signal
|
||||
const abortHandler = () => result.session.agent.abort();
|
||||
signal?.addEventListener("abort", abortHandler, { once: true });
|
||||
@@ -672,8 +726,26 @@ export async function runAgentSession(
|
||||
let finalText = "";
|
||||
let errorMessage: string | undefined;
|
||||
let stopReason: string | undefined;
|
||||
lastEventTime = Date.now();
|
||||
|
||||
// Inactivity watchdog: check the silence window on an interval and abort
|
||||
// (plus kill any hung bash subprocess) when it is exceeded.
|
||||
if (inactivityTimeoutMs > 0) {
|
||||
const intervalMs = Math.min(inactivityTimeoutMs, 5000);
|
||||
inactivityInterval = setInterval(() => {
|
||||
if (!sessionRef.session) return;
|
||||
if (Date.now() - lastEventTime <= inactivityTimeoutMs) return;
|
||||
inactivityAborted = true;
|
||||
sessionRef.session.agent.abort();
|
||||
sessionRef.session.abortBash();
|
||||
errorMessage = `Task aborted: inactivity timeout (no events for ${Math.round(inactivityTimeoutMs / 1000)}s)`;
|
||||
if (inactivityInterval) clearInterval(inactivityInterval);
|
||||
inactivityInterval = null;
|
||||
}, intervalMs);
|
||||
}
|
||||
|
||||
const unsubscribe = result.session.subscribe((event) => {
|
||||
lastEventTime = Date.now();
|
||||
onEvent?.(event);
|
||||
|
||||
if (event.type === "message_end") {
|
||||
@@ -685,7 +757,10 @@ export async function runAgentSession(
|
||||
};
|
||||
if (message.role !== "assistant") return;
|
||||
if (message.stopReason) stopReason = message.stopReason;
|
||||
if (message.errorMessage) errorMessage = message.errorMessage;
|
||||
// Keep the inactivity-timeout message: the abort's own errorMessage
|
||||
// would otherwise clobber the (more useful) hang explanation.
|
||||
if (message.errorMessage && !inactivityAborted)
|
||||
errorMessage = message.errorMessage;
|
||||
const text = extractAssistantText(message.content);
|
||||
if (text) finalText = text;
|
||||
}
|
||||
@@ -718,6 +793,7 @@ export async function runAgentSession(
|
||||
toolUsage,
|
||||
stopReason,
|
||||
events: [], // streamed to file
|
||||
sessionFile,
|
||||
};
|
||||
}
|
||||
|
||||
@@ -727,6 +803,7 @@ export async function runAgentSession(
|
||||
toolUsage,
|
||||
stopReason,
|
||||
events: [],
|
||||
sessionFile,
|
||||
};
|
||||
} catch (error) {
|
||||
if (timeoutHandle) clearTimeout(timeoutHandle);
|
||||
@@ -736,9 +813,14 @@ export async function runAgentSession(
|
||||
error: error instanceof Error ? error.message : String(error),
|
||||
toolUsage,
|
||||
events: [],
|
||||
sessionFile,
|
||||
// A requested resume that failed to open (corrupt/unreadable file)
|
||||
// should not be retried — callers clear the stored file and go fresh.
|
||||
resumeFailed: resumeSessionFile !== undefined && !sessionCreated,
|
||||
};
|
||||
} finally {
|
||||
sessionRef.session?.dispose();
|
||||
if (inactivityInterval) clearInterval(inactivityInterval);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
Reference in New Issue
Block a user