Mike/omp-ralpi
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

port: sync from Mike/ralpi@c870efa1

Browse Source
This commit is contained in:
omp-port
2026-08-12 14:13:36 -04:00
parent 479bb55d2d
commit 470c660ad8
7 changed files with 282 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
AGENTS.md
View File

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

2
package.json
View File

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

115
src/executor.ts
View File

@@ -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<ReturnType<typeof runAgentSession>>;
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) {

22
src/progress.ts
View File

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

57
src/prompts.ts
View File

@@ -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?",

10
src/types.ts
View File

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

88
src/utils.ts
View File

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