From 470c660ad80d59ea4fbbd62f18881390eb285ef4 Mon Sep 17 00:00:00 2001 From: omp-port Date: Wed, 12 Aug 2026 14:13:36 -0400 Subject: [PATCH] port: sync from Mike/ralpi@c870efa1 --- AGENTS.md | 3 ++ package.json | 2 +- src/executor.ts | 115 ++++++++++++++++++++++++++++++++++++++++++++---- src/progress.ts | 22 ++++++++- src/prompts.ts | 57 +++++++++++++++++++++++- src/types.ts | 10 +++++ src/utils.ts | 88 +++++++++++++++++++++++++++++++++++- 7 files changed, 282 insertions(+), 15 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 2957e8e..9a34040 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "@mikefreno/omp-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": [ "omp", diff --git a/src/executor.ts b/src/executor.ts index 3232b8e..e86aedd 100644 --- a/src/executor.ts +++ b/src/executor.ts @@ -244,6 +244,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; @@ -253,6 +260,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(); @@ -384,6 +397,9 @@ export async function runTask( config.thinkingLevel, false, // noSkills — task sessions need skills ctx.modelRegistry, + config.execution.inactivityTimeoutMs, + resumeSessionFile, + onSessionFile, ); const durationMs = Date.now() - startMs; @@ -408,6 +424,8 @@ export async function runTask( success: false, error: output.error, durationMs, + sessionFile: output.sessionFile, + resumeFailed: output.resumeFailed, }; } @@ -438,6 +456,7 @@ export async function runTask( outputPreview, commitMessages, commitSummary, + sessionFile: output.sessionFile, }; } @@ -824,6 +843,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 @@ -884,6 +909,11 @@ async function executeTask( currentModel, batchRender, priorReview, + modelAttempt === 0 && sameModelAttempt === 0 + ? resumeSessionFile + : undefined, + (sessionFile) => + progress.setSessionFile(task.id, sessionFile), ); if (result.success) { @@ -935,6 +965,10 @@ async function executeTask( // A FAILED range computation (broken/stale base ref, git error) is // logged as a distinct warning and is never treated as a clean, // verified task — only a GENUINE "no changes" skips review. + // Accumulated rejected reviews from earlier passes — injected + // into the next review prompt so the reviewer sees prior + // findings and can verify they were addressed. + const priorReviews: ReviewResult[] = []; while (true) { if (!baseRef) { sendChatMessage?.( @@ -972,16 +1006,17 @@ async function executeTask( reviewInfo.hash, reviewInfo.subject, reviewInfo.diff, - { - projectContext: config.prompts.projectContext, - focus: config.prompts.reviewFocus, - diffOptions: { - extraPatterns: compileIgnorePatterns( - config.review.extraIgnorePatterns, - ), - ignorePaths: config.review.ignorePaths, - }, + { + projectContext: config.prompts.projectContext, + focus: config.prompts.reviewFocus, + priorReviews, + diffOptions: { + extraPatterns: compileIgnorePatterns( + config.review.extraIgnorePatterns, + ), + ignorePaths: config.review.ignorePaths, }, + }, ); const reviewModel = resolveFollowUpModel( @@ -1003,6 +1038,7 @@ async function executeTask( `review-${task.id}`, config.execution.reviewTimeoutMs, reviewModels, + config.execution.inactivityTimeoutMs, ); if (!reviewResult.success) { @@ -1096,6 +1132,9 @@ async function executeTask( break; // changes already committed — merge proceeds } + // Accumulate the rejected review so the next review pass + // sees prior findings and can verify they were addressed. + if (review) priorReviews.push(review); attempt++; reviewRetries++; sendChatMessage?.( @@ -1292,6 +1331,7 @@ async function executeTask( finalCommitSummary, finalReview, reviewRetries, + result.sessionFile, ); // Auto-update the PRD source file checkbox try { @@ -1305,6 +1345,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++; @@ -1437,6 +1483,8 @@ async function runFollowUpSession( widgetKeySuffix: string, timeoutMs: number, models: unknown[], + /** Inactivity timeout for the session (see runAgentSession). */ + inactivityTimeoutMs = 0, ): Promise<{ result: Awaited>; toolCalls: ToolCallEntry[]; @@ -1535,6 +1583,7 @@ async function runFollowUpSession( config.thinkingLevel, false, // noSkills=false — follow-up sessions load skills too ctx.modelRegistry, + inactivityTimeoutMs, ); if (result.success) break; @@ -1681,6 +1730,7 @@ async function runCommitSession( `commit-${task.id}`, config.execution.commitTimeoutMs, commitModels, + config.execution.inactivityTimeoutMs, ); if (commitResult.success) { @@ -1778,6 +1828,52 @@ async function resolveConflictsSession( return; } + // Merge failed but produced no unmerged paths — no real conflicts to + // resolve. This can happen when the branch tip was already merged by the + // first attempt (mergeWorktree) before it aborted, or when the merge + // fails for a non-conflict reason (dirty index, stale ref). Don't spawn + // an agent session for zero conflicts — abort or complete and finish. + if (attempt.conflicts.length === 0) { + // Try to complete whatever merge state exists; if there's nothing to + // commit, abort to leave the working tree clean. + if (completeMerge(projectDir)) { + sendChatMessage?.( + `✓ conflicts auto-resolved for ${task.id} · ${task.title}`, + ); + removeWorktree(projectDir, worktree); + progress.markCompleted( + task.id, + 0, + undefined, + undefined, + undefined, + [], + "", + undefined, + 0, + ); + try { + updateTaskInFile(project.sourcePath, task.id, "completed"); + } catch { + // Best-effort + } + return; + } + abortMerge(projectDir); + sendChatMessage?.( + `~ conflict resolution for ${task.id} · ${task.title} — merge produced no conflicts but could not be completed`, + ); + progress.markFailed( + task.id, + `Merge of ${branch} produced no conflicts but could not be completed`, + ); + try { + updateTaskInFile(project.sourcePath, task.id, "failed"); + } catch { + // Best-effort + } + return; + } // Conflicts exist — spawn a resolution agent session. const prompt = buildConflictResolutionPrompt( task, @@ -1807,6 +1903,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/prompts.ts b/src/prompts.ts index 79088a8..09641ed 100644 --- a/src/prompts.ts +++ b/src/prompts.ts @@ -29,6 +29,12 @@ export interface ReviewPromptOptions { focus?: string; /** Noise-filter overrides (config.review.*). */ diffOptions?: DiffOptions; + /** Prior review results from earlier passes in a review-gated loop. + * Each rejected review is injected into the next review prompt so the + * reviewer can verify prior findings were addressed and catch new + * regressions introduced by the fix attempt — instead of re-reviewing + * from scratch. */ + priorReviews?: ReviewResult[]; } // ─── Task Prompt ───────────────────────────────────────────────────────────── @@ -258,8 +264,12 @@ export function buildReviewPrompt( parts.push(renderDiffSection(summary, filtered, "### Diff")); parts.push(""); - // ── Custom Review Focus ── + // ── Prior Review History (review-gated re-review) ── + parts.push(renderPriorReviews(opts.priorReviews ?? [])); + if (opts.priorReviews && opts.priorReviews.length > 0) parts.push(""); + + // ── Custom Review Focus ── if (opts.focus) { parts.push("## Custom Review Focus"); parts.push(opts.focus); @@ -360,7 +370,11 @@ export function buildReviewPromptUncommitted( parts.push( renderDiffSection(summary, filtered, "### Current Tracked Diff (git diff)"), ); - parts.push(""); + + // ── Prior Review History (review-gated re-review) ── + + parts.push(renderPriorReviews(opts.priorReviews ?? [])); + if (opts.priorReviews && opts.priorReviews.length > 0) parts.push(""); // ── Custom Review Focus ── @@ -469,6 +483,45 @@ function renderDiffSection( return lines.join("\n"); } +/** + * Render a "Prior Review History" section from earlier rejected reviews. + * Returns an empty string when there are no prior reviews so callers omit + * the section entirely. + * + * Each prior review's verdict, summary, and findings are listed so the + * reviewer can verify the developer addressed them and watch for new + * regressions — instead of re-reviewing from scratch on each pass. + */ +function renderPriorReviews(priorReviews: ReviewResult[]): string { + if (priorReviews.length === 0) return ""; + const lines: string[] = []; + lines.push("## Prior Review History"); + lines.push( + "Previous review pass(es) rejected this task. Verify each finding was", + "addressed in the current diff and watch for new regressions:", + ); + lines.push(""); + for (let i = 0; i < priorReviews.length; i++) { + const r = priorReviews[i]; + if (!r) continue; + lines.push(`### Review ${i + 1} — ${r.verdict.toUpperCase()}`); + lines.push(`Summary: ${r.summary}`); + if (r.findings.length > 0) { + lines.push("Findings:"); + for (const f of r.findings) { + const loc = f.file + ? f.line + ? ` (${f.file}:${f.line})` + : ` (${f.file})` + : ""; + lines.push(`- [${f.severity}]${loc} ${f.message}`); + } + } + lines.push(""); + } + return lines.join("\n"); +} + function reviewInstructions(): string[] { return [ "- **Correctness**: Does the implementation fulfill the task requirements?", 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 cefa34b..df571f8 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 registry from models.json only — extension providers are lost. */ modelRegistry?: ModelRegistry, + /** 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,12 +657,45 @@ 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 omp context: extensions (so all // extension-provided tools register) and project context (AGENTS.md). + // 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 = await SessionManager.open(resumeSessionFile, sessionDir, undefined, { + initialCwd: 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(cwd), + sessionManager, settingsManager: Settings.init({ cwd, agentDir: getAgentDir() }), // Loop sessions intentionally load extensions (no disableExtensionDiscovery), // plus skills and project context via default discovery. @@ -655,8 +707,12 @@ export async function runAgentSession( modelRegistry, agentRegistry: new AgentRegistry(), }); + 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 }); @@ -664,8 +720,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") { @@ -677,7 +751,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; } @@ -710,6 +787,7 @@ export async function runAgentSession( toolUsage, stopReason, events: [], // streamed to file + sessionFile, }; } @@ -719,6 +797,7 @@ export async function runAgentSession( toolUsage, stopReason, events: [], + sessionFile, }; } catch (error) { if (timeoutHandle) clearTimeout(timeoutHandle); @@ -728,9 +807,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); } }