From 100262b94f903cba7e4c7dd01cf68febf41b831b Mon Sep 17 00:00:00 2001 From: Michael Freno Date: Wed, 12 Aug 2026 12:55:44 -0400 Subject: [PATCH] feat: hang detection and resumable task sessions - 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 --- AGENTS.md | 3 ++ package.json | 2 +- src/executor.ts | 43 +++++++++++++++++++++++++ src/progress.ts | 22 ++++++++++++- src/types.ts | 10 ++++++ src/utils.ts | 86 +++++++++++++++++++++++++++++++++++++++++++++++-- 6 files changed, 162 insertions(+), 4 deletions(-) diff --git a/AGENTS.md b/AGENTS.md index e8d8b13..c1b62f2 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -124,6 +124,9 @@ Key config fields in `execution`: - `autoCommit` / `autoReview` / `saveReviews` — loop options (selectable at loop startup via `selectLoopOptions`; review is asked FIRST, commit is mandated when review is on) +- `inactivityTimeoutMs` — hang detection: if no agent session event arrives + within this window (e.g. a bash subprocess that never returns), the task is + aborted (agent abort + bash subprocess kill). `0` = disabled (default) - `models` — slot-aware round-robin model list for parallel mode, with failover to the next model per task (only after exhausting same-model retries, see `maxSameModelAttempts`) diff --git a/package.json b/package.json index c9817e9..9b95383 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@mikefreno/ralpi", - "version": "0.5.0", + "version": "0.6.0", "description": "Execute tasks from task files/PRD's using DAG-based dependency resolution with persistent progress tracking", "keywords": [ "pi-package", diff --git a/src/executor.ts b/src/executor.ts index f5cd049..581ef0e 100644 --- a/src/executor.ts +++ b/src/executor.ts @@ -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>; 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) { diff --git a/src/progress.ts b/src/progress.ts index 804fc6d..1f51cc6 100644 --- a/src/progress.ts +++ b/src/progress.ts @@ -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(); diff --git a/src/types.ts b/src/types.ts index ffa93ba..53bfd50 100644 --- a/src/types.ts +++ b/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, diff --git a/src/utils.ts b/src/utils.ts index 9921be7..5c04340 100644 --- a/src/utils.ts +++ b/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>["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 `/.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); } }