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

depend on pi defaults, and global yaml

Browse Source
This commit is contained in:
Michael Freno
2026-05-31 01:35:00 -04:00
parent ead5d9be3a
commit 8e2e24d0e3
6 changed files with 1073 additions and 873 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
README.md
View File

@@ -98,16 +98,34 @@ tasks:
## Configuration ## Configuration
Create `.ralpi/config.yaml`: Create config files. Both are optional:
| Scope | Path |
|-------|------|
| **Global** | `~/.pi/ralpi/config.yaml` |
| **Project** | `./.ralpi/config.yaml` |
```yaml ```yaml
maxRetries: 3 execution:
retryDelayMs: 5000 maxParallel: 3 # ralpi-level concurrency only
timeoutMs: 1800000 prompts:
maxParallel: 3 projectContext: "Additional context for all tasks"
projectContext: "Additional context for all tasks"
``` ```
> ralpi deliberately does **not** set timeouts or retries — those are inherited
> from Pi's own settings. Tasks run until they complete or Pi's own flow stops them.
The keys mirror the nested structure of `RalpiConfig` in `src/types.ts`.
### Precedence (highest wins)
| Priority | Source |
|----------|--------|
| **1st** | In-memory overrides (`model`, `thinkingLevel` from parent Pi session) |
| **2nd** | `./.ralpi/config.yaml` — project-level |
| **3rd** | `~/.pi/ralpi/config.yaml` — global, shared across projects |
| **4th** | `DEFAULT_CONFIG` in `src/types.ts` |
### Task-Level Timeout ### Task-Level Timeout
You can set a timeout for individual tasks using a meta block in the task file: You can set a timeout for individual tasks using a meta block in the task file:

1003
index.ts
View File

File diff suppressed because it is too large Load Diff

6
package.json
View File

@@ -1,7 +1,7 @@
{ {
"name": "ralpi-loop", "name": "ralpi",
"version": "1.0.0", "version": "0.1.0",
"description": "Execute tasks from task files using DAG-based dependency resolution with persistent progress tracking", "description": "Execute tasks from task files/PRD's using DAG-based dependency resolution with persistent progress tracking",
"main": "dist/index.js", "main": "dist/index.js",
"keywords": [ "keywords": [
"pi-package", "pi-package",

778
src/executor.ts
View File

@@ -5,27 +5,47 @@ import type { ProgressTracker } from "./progress";
import type { ExtensionContext } from "@earendil-works/pi-coding-agent"; import type { ExtensionContext } from "@earendil-works/pi-coding-agent";
import { buildTaskPrompt } from "./prompts"; import { buildTaskPrompt } from "./prompts";
import { extractReflection } from "./reflection"; import { extractReflection } from "./reflection";
import { WidgetBatcher } from "./widget-batcher";
import { import {
runAgentSession, runAgentSession,
writeFileSafe, writeFileSafe,
ensureDir, ensureDir,
captureGitCommits, captureGitCommits,
formatDuration, formatDuration,
} from "./utils"; } from "./utils";
/** Optional callback to post a progress message into the chat history. */ /** Optional callback to post a progress message into the chat history. */
export type SendChatMessage = ( export type SendChatMessage = (
content: string, content: string,
/** Extra data passed to the message renderer for the expanded view. */ /** Extra data passed to the message renderer for the expanded view. */
meta?: { toolCalls?: ToolCallEntry[] }, meta?: { toolCalls?: ToolCallEntry[] },
) => void; ) => void;
export interface ToolCallEntry { export interface ToolCallEntry {
name: string; name: string;
label: string; label: string;
} }
// ─── Widget Expand/Collapse ───────────────────────────────────────────────
/** Max tool calls shown in a live widget before truncating. Widgets don't
* support message-style Ctrl+O expansion (that's only for chat-history
* messages rendered by registerMessageRenderer). */
const MAX_COLLAPSED = 3;
const SPINNER_FRAMES = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"];
/** Shared state for parallel-batch widget. Each running task writes its
* tool calls and spinner frame; the batch widget reads them in task-ID order. */
interface ParallelWidgetEntry {
taskHeader: string;
frameIndex: number;
done: boolean;
success: boolean;
toolCalls: ToolCallEntry[];
}
type ParallelWidgetState = Map<string, ParallelWidgetEntry>;
// ─── Run Single Task ──────────────────────────────────────────────────────── // ─── Run Single Task ────────────────────────────────────────────────────────
/** /**
@@ -33,176 +53,204 @@ export interface ToolCallEntry {
* Non-blocking — the TUI remains responsive throughout. * Non-blocking — the TUI remains responsive throughout.
*/ */
export async function runTask( export async function runTask(
task: Task, task: Task,
project: Project, project: Project,
config: RalpiConfig, config: RalpiConfig,
depReflections: Reflection[], depReflections: Reflection[],
ctx: ExtensionContext, ctx: ExtensionContext,
sendChatMessage?: SendChatMessage, sendChatMessage?: SendChatMessage,
projectDir: string = project.sourceDir, projectDir: string = project.sourceDir,
batcher?: WidgetBatcher, parallelState?: ParallelWidgetState,
): Promise<{ ): Promise<{
success: boolean; success: boolean;
reflection?: Reflection; reflection?: Reflection;
error?: string; error?: string;
durationMs: number; durationMs: number;
toolUsage?: ToolUsage; toolUsage?: ToolUsage;
outputPreview?: string; outputPreview?: string;
sessionFile?: string; sessionFile?: string;
commitMessages?: string[]; commitMessages?: string[];
commitSummary?: string; commitSummary?: string;
}> { }> {
const startMs = Date.now(); const startMs = Date.now();
// Build prompt // Build prompt
const prompt = buildTaskPrompt( const prompt = buildTaskPrompt(
task, task,
project, project,
depReflections, depReflections,
config.prompts.projectContext, config.prompts.projectContext,
); );
// Write prompt to .ralpi/ with timestamp (for debugging) // Write prompt to .ralpi/ with timestamp (for debugging)
const ralpiDir = path.join(projectDir, ".ralpi"); const ralpiDir = path.join(projectDir, ".ralpi");
ensureDir(ralpiDir); ensureDir(ralpiDir);
const promptFile = path.join(ralpiDir, `prompt-${startMs}.md`); const promptFile = path.join(ralpiDir, `prompt-${startMs}.md`);
writeFileSafe(promptFile, prompt); writeFileSafe(promptFile, prompt);
// Footer shows just the task title (no batch prefix) const taskHeader = `${task.id} · ${task.title}`;
ctx.ui.setStatus("ralpi", task.title);
const taskHeader = `${task.id} · ${task.title}`; // When running in parallel, all tasks share a single widget so ordering
// is deterministic (sorted by task ID). In sequential mode each task gets
// its own widget.
const isParallel = !!parallelState;
const widgetKey = `ralpi-task-${task.id}`;
let frameIndex = 0;
const toolCalls: ToolCallEntry[] = [];
let widgetTui: { requestRender(): void } | null = null;
// Live progress widget above the editor — animated spinner + tool call tree if (isParallel) {
// Using setWidget instead of setWorkingMessage because the working message area parallelState!.set(task.id, {
// is only visible during parent agent streaming, not during extension command execution. taskHeader,
// Widget key is unique per task so parallel tasks each get their own widget. frameIndex: 0,
const widgetKey = `ralpi-task-${task.id}`; done: false,
const SPINNER_FRAMES = ["⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏"]; success: false,
let frameIndex = 0; toolCalls: [],
const theme = ctx.ui.theme; });
const MAX_COLLAPSED = 3; } else {
// Build widget lines from current state. Live widgets can't expand/collapse
// like chat messages, so we always truncate to MAX_COLLAPSED recent calls.
const buildLines = (t: typeof ctx.ui.theme): string[] => {
const frame = t.fg("accent", SPINNER_FRAMES[frameIndex]);
const lines = [`${frame} ${taskHeader}`];
const toolCalls: ToolCallEntry[] = []; if (toolCalls.length > 0) {
if (toolCalls.length <= MAX_COLLAPSED) {
for (let i = 0; i < toolCalls.length; i++) {
const entry = toolCalls[i];
const isLast = i === toolCalls.length - 1;
const branch = isLast ? " └── " : " ├── ";
const tag = t.fg("accent", `[${entry.name}]`);
lines.push(`${branch}${tag} ${entry.label}`);
}
} else {
const shown = toolCalls.slice(-MAX_COLLAPSED);
const remaining = toolCalls.length - shown.length;
lines.push(t.fg("dim", ` ├── …${remaining} earlier`));
for (let i = 0; i < shown.length; i++) {
const entry = shown[i];
const isLast = i === shown.length - 1;
const branch = isLast ? " └── " : " ├── ";
const tag = t.fg("accent", `[${entry.name}]`);
lines.push(`${branch}${tag} ${entry.label}`);
}
}
}
return lines;
};
const updateWidget = () => { ctx.ui.setWidget(widgetKey, (tui, t) => {
const frame = theme.fg("accent", SPINNER_FRAMES[frameIndex]); widgetTui = tui;
const lines = [`${frame} ${taskHeader}`]; return {
render: () => buildLines(t),
invalidate: () => widgetTui?.requestRender(),
};
});
}
if (toolCalls.length > 0) { const requestRender = () => widgetTui?.requestRender();
const shown = toolCalls.slice(-MAX_COLLAPSED);
const remaining = toolCalls.length - shown.length;
if (remaining > 0) { // Spinner animation (sequential only — parallel uses a single batch timer)
lines.push(theme.fg("dim", ` ├── ${remaining} more`)); let spinnerTimer: NodeJS.Timeout | undefined;
} if (!isParallel) {
spinnerTimer = setInterval(() => {
frameIndex = (frameIndex + 1) % SPINNER_FRAMES.length;
requestRender();
}, 100);
}
for (let i = 0; i < shown.length; i++) { // Use task-level timeout if set, otherwise fall back to config
const entry = shown[i]; const timeoutMs = task.timeoutMs ?? config.execution.timeoutMs;
const isLast = i === shown.length - 1;
const branch = isLast ? " └── " : " ├── ";
const tag = theme.fg("accent", `[${entry.name}]`);
lines.push(`${branch}${tag} ${entry.label}`);
}
}
if (batcher) { // Pre-create session file path so events stream to disk (avoids 300+ MB in-memory accumulation)
batcher.schedule(widgetKey, lines); const sessionsDir = path.join(ralpiDir, "sessions");
} else { ensureDir(sessionsDir);
ctx.ui.setWidget(widgetKey, lines); const sessionFilePath = path.join(sessionsDir, `${task.id}-${startMs}.txt`);
}
};
// Smooth spinner animation at 100ms intervals // Run task asynchronously via Pi SDK — event loop stays responsive
const spinnerTimer = setInterval(() => { const output = await runAgentSession(
frameIndex = (frameIndex + 1) % SPINNER_FRAMES.length; prompt,
updateWidget(); projectDir,
}, 100); timeoutMs,
(event) => {
if (event.type === "tool_execution_start") {
const label = formatToolArg(event.toolName, event.args);
toolCalls.push({
name: event.toolName,
label,
});
if (isParallel) {
const entry = parallelState!.get(task.id);
if (entry) {
entry.toolCalls.push({ name: event.toolName, label });
}
}
requestRender();
}
},
undefined, // no abort signal
sessionFilePath, // stream events to file
config.model,
config.thinkingLevel,
);
// Initial display const durationMs = Date.now() - startMs;
updateWidget();
// Use task-level timeout if set, otherwise fall back to config // Clear progress widget and status after task finishes
const timeoutMs = task.timeoutMs ?? config.execution.timeoutMs; if (spinnerTimer) clearInterval(spinnerTimer);
if (isParallel) {
const entry = parallelState!.get(task.id);
if (entry) {
entry.done = true;
entry.success = output.success;
}
} else {
ctx.ui.setWidget(widgetKey, undefined);
}
// Pre-create session file path so events stream to disk (avoids 300+ MB in-memory accumulation) if (!output.success) {
const sessionsDir = path.join(ralpiDir, "sessions"); sendChatMessage?.(`✗ ${taskHeader} — ${output.error}`);
ensureDir(sessionsDir); ctx.ui.notify(`Task ${task.id} failed: ${output.error}`, "error");
const sessionFilePath = path.join(sessionsDir, `${task.id}-${startMs}.txt`); return {
success: false,
error: output.error,
durationMs,
sessionFile: sessionFilePath, // events streamed to file for debugging
};
}
// Run task asynchronously via Pi SDK — event loop stays responsive const agentText = output.text;
const output = await runAgentSession( const toolUsage = output.toolUsage;
prompt,
projectDir,
timeoutMs,
(event) => {
if (event.type === "tool_execution_start") {
const label = formatToolArg(event.toolName, event.args);
toolCalls.push({
name: event.toolName,
label,
});
updateWidget();
}
},
undefined, // no abort signal
sessionFilePath, // stream events to file
);
const durationMs = Date.now() - startMs; // Capture git commits made during this task
const { commitMessages, commitSummary } = captureGitCommits(projectDir);
// Clear progress widget and status after task finishes // Session file already written by runAgentSession (events streamed to disk)
clearInterval(spinnerTimer); const sessionFile = sessionFilePath;
if (batcher) {
batcher.scheduleRemove(widgetKey);
} else {
ctx.ui.setWidget(widgetKey, undefined);
}
ctx.ui.setStatus("ralpi", undefined);
if (!output.success) { // Build output preview (first 500 chars of agent text)
sendChatMessage?.(`✗ ${taskHeader} — ${output.error}`); const outputPreview =
ctx.ui.notify(`Task ${task.id} failed: ${output.error}`, "error"); agentText.length > 500
return { ? agentText.slice(0, 500) + "\n... (truncated, see session file)"
success: false, : agentText;
error: output.error,
durationMs,
sessionFile: sessionFilePath, // events streamed to file for debugging
};
}
const agentText = output.text; // Extract reflection from agent output
const toolUsage = output.toolUsage; const reflection = extractReflection(agentText, task.id, task.title);
// Capture git commits made during this task // Post completion chat message — header only, renderer builds the expandable tree
const { commitMessages, commitSummary } = captureGitCommits(projectDir); const dur = formatDuration(durationMs);
sendChatMessage?.(`✓ ${taskHeader} (${dur})`, { toolCalls });
// Session file already written by runAgentSession (events streamed to disk) return {
const sessionFile = sessionFilePath; success: true,
reflection: reflection ?? undefined,
// Build output preview (first 500 chars of agent text) durationMs,
const outputPreview = toolUsage,
agentText.length > 500 outputPreview,
? agentText.slice(0, 500) + "\n... (truncated, see session file)" sessionFile,
: agentText; commitMessages,
commitSummary,
// Extract reflection from agent output };
const reflection = extractReflection(agentText, task.id, task.title);
// Post completion chat message — header only, renderer builds the expandable tree
const dur = formatDuration(durationMs);
sendChatMessage?.(`✓ ${taskHeader} (${dur})`, { toolCalls });
return {
success: true,
reflection: reflection ?? undefined,
durationMs,
toolUsage,
outputPreview,
sessionFile,
commitMessages,
commitSummary,
};
} }
// ─── Execute Batch ─────────────────────────────────────────────────────────── // ─── Execute Batch ───────────────────────────────────────────────────────────
@@ -211,198 +259,260 @@ export async function runTask(
* Execute a batch of tasks (sequentially or in parallel) * Execute a batch of tasks (sequentially or in parallel)
*/ */
export async function executeBatch( export async function executeBatch(
tasks: Task[], tasks: Task[],
project: Project, project: Project,
config: RalpiConfig, config: RalpiConfig,
progress: ProgressTracker, progress: ProgressTracker,
ctx: ExtensionContext, ctx: ExtensionContext,
options?: { parallel?: boolean }, options?: { parallel?: boolean },
sendChatMessage?: SendChatMessage, sendChatMessage?: SendChatMessage,
projectDir?: string, projectDir?: string,
): Promise<void> { ): Promise<void> {
// Defensive: ensure tasks is an iterable array // Defensive: ensure tasks is an iterable array
if (!Array.isArray(tasks)) { if (!Array.isArray(tasks)) {
throw new Error( throw new Error(
`executeBatch received invalid tasks: expected array, got ${typeof tasks}`, `executeBatch received invalid tasks: expected array, got ${typeof tasks}`,
); );
} }
// Check if we should run parallel // Check if we should run parallel
const shouldParallel = const shouldParallel =
options?.parallel && tasks.length > 1 && config.execution.maxParallel > 0; options?.parallel && tasks.length > 1 && config.execution.maxParallel > 0;
if (shouldParallel) { if (shouldParallel) {
await executeBatchParallel( await executeBatchParallel(
tasks, tasks,
project, project,
config, config,
progress, progress,
ctx, ctx,
sendChatMessage, sendChatMessage,
projectDir, projectDir,
); );
return; return;
} }
// Execute sequentially // Execute sequentially
for (const task of tasks) { for (const task of tasks) {
await executeTask( await executeTask(
task, task,
project, project,
config, config,
progress, progress,
ctx, ctx,
sendChatMessage, sendChatMessage,
projectDir, projectDir,
); );
} }
} }
/** /**
* Execute tasks in parallel using child processes * Execute tasks in parallel using child processes
*/ */
async function executeBatchParallel( async function executeBatchParallel(
tasks: Task[], tasks: Task[],
project: Project, project: Project,
config: RalpiConfig, config: RalpiConfig,
progress: ProgressTracker, progress: ProgressTracker,
ctx: ExtensionContext, ctx: ExtensionContext,
sendChatMessage?: SendChatMessage, sendChatMessage?: SendChatMessage,
projectDir?: string, projectDir?: string,
): Promise<void> { ): Promise<void> {
const maxParallel = config.execution.maxParallel; const maxParallel = config.execution.maxParallel;
const batcher = new WidgetBatcher(ctx); const sharedState: ParallelWidgetState = new Map();
const results: Array<{ task: Task; result: Promise<any> }> = [];
for (const task of tasks) { // Register a single batch widget that renders ALL parallel tasks in ID order.
results.push({ const widgetKey = `ralpi-parallel-${Date.now()}`;
task, let widgetTui: { requestRender(): void } | null = null;
result: executeTask(
task,
project,
config,
progress,
ctx,
sendChatMessage,
projectDir,
batcher,
),
});
// Limit concurrency const buildBatchLines = (t: typeof ctx.ui.theme): string[] => {
if (results.length >= maxParallel) { const lines: string[] = [];
const first = results.shift(); const sortedIds = Array.from(sharedState.keys()).sort();
if (first) await first.result;
}
}
// Wait for remaining tasks for (const id of sortedIds) {
for (const { result } of results) { const entry = sharedState.get(id)!;
await result; const frame = entry.done
} ? entry.success
? "✓"
: "✗"
: t.fg("accent", SPINNER_FRAMES[entry.frameIndex]);
lines.push(`${frame} ${entry.taskHeader}`);
// Flush and stop the batcher after all tasks complete if (entry.toolCalls.length > 0) {
batcher.stop(); if (entry.toolCalls.length <= MAX_COLLAPSED) {
for (let i = 0; i < entry.toolCalls.length; i++) {
const tc = entry.toolCalls[i];
const isLast = i === entry.toolCalls.length - 1;
const branch = isLast ? " └── " : " ├── ";
const tag = t.fg("accent", `[${tc.name}]`);
lines.push(`${branch}${tag} ${tc.label}`);
}
} else {
const shown = entry.toolCalls.slice(-MAX_COLLAPSED);
const remaining = entry.toolCalls.length - shown.length;
lines.push(t.fg("dim", ` ├── …${remaining} earlier`));
for (let i = 0; i < shown.length; i++) {
const tc = shown[i];
const isLast = i === shown.length - 1;
const branch = isLast ? " └── " : " ├── ";
const tag = t.fg("accent", `[${tc.name}]`);
lines.push(`${branch}${tag} ${tc.label}`);
}
}
}
}
return lines;
};
ctx.ui.setWidget(widgetKey, (tui, t) => {
widgetTui = tui;
return {
render: () => buildBatchLines(t),
invalidate: () => widgetTui?.requestRender(),
};
});
// Single spinner timer drives all tasks in the batch
const spinnerTimer = setInterval(() => {
for (const entry of sharedState.values()) {
if (!entry.done) {
entry.frameIndex = (entry.frameIndex + 1) % SPINNER_FRAMES.length;
}
}
widgetTui?.requestRender();
}, 100);
const results: Array<{ task: Task; result: Promise<any> }> = [];
for (const task of tasks) {
results.push({
task,
result: executeTask(
task,
project,
config,
progress,
ctx,
sendChatMessage,
projectDir,
sharedState,
),
});
// Limit concurrency
if (results.length >= maxParallel) {
const first = results.shift();
if (first) await first.result;
}
}
// Wait for remaining tasks
for (const { result } of results) {
await result;
}
clearInterval(spinnerTimer);
ctx.ui.setWidget(widgetKey, undefined);
} }
// ─── Execute Single Task with Retry ────────────────────────────────────────── // ─── Execute Single Task with Retry ──────────────────────────────────────────
async function executeTask( async function executeTask(
task: Task, task: Task,
project: Project, project: Project,
config: RalpiConfig, config: RalpiConfig,
progress: ProgressTracker, progress: ProgressTracker,
ctx: ExtensionContext, ctx: ExtensionContext,
sendChatMessage?: SendChatMessage, sendChatMessage?: SendChatMessage,
projectDir: string = project.sourceDir, projectDir: string = project.sourceDir,
batcher?: WidgetBatcher, parallelState?: ParallelWidgetState,
): Promise<void> { ): Promise<void> {
const maxRetries = config.execution.maxRetries; const maxRetries = config.execution.maxRetries;
let retries = 0; let retries = 0;
while (retries <= maxRetries) { while (retries <= maxRetries) {
try { try {
// Mark as in progress // Mark as in progress
progress.markInProgress(task.id); progress.markInProgress(task.id);
// Get dependency reflections // Get dependency reflections
const depReflections = progress.getDependencyReflections( const depReflections = progress.getDependencyReflections(
task.dependencies || [], task.dependencies || [],
); );
// Run the task // Run the task
const result = await runTask( const result = await runTask(
task, task,
project, project,
config, config,
depReflections, depReflections,
ctx, ctx,
sendChatMessage, sendChatMessage,
projectDir, projectDir,
batcher, parallelState,
); );
if (result.success) { if (result.success) {
// Save reflection // Save reflection
if (result.reflection) { if (result.reflection) {
saveReflectionToFile(projectDir, config, result.reflection); saveReflectionToFile(projectDir, config, result.reflection);
} }
// Mark completed with all metadata // Mark completed with all metadata
progress.markCompleted( progress.markCompleted(
task.id, task.id,
result.durationMs, result.durationMs,
result.reflection, result.reflection,
result.toolUsage, result.toolUsage,
result.sessionFile, result.sessionFile,
result.outputPreview, result.outputPreview,
result.commitMessages, result.commitMessages,
result.commitSummary, result.commitSummary,
); );
return; return;
} }
// Task failed, check if we should retry // Task failed, check if we should retry
if (retries < maxRetries) { if (retries < maxRetries) {
retries = progress.incrementRetry(task.id); retries = progress.incrementRetry(task.id);
ctx.ui.notify( ctx.ui.notify(
`Retrying task ${task.id} (${retries}/${maxRetries}): ${result.error}`, `Retrying task ${task.id} (${retries}/${maxRetries}): ${result.error}`,
"warning", "warning",
); );
// Exponential backoff // Exponential backoff
const delay = config.execution.retryDelayMs * 2 ** (retries - 1); const delay = config.execution.retryDelayMs * 2 ** (retries - 1);
await sleep(delay); await sleep(delay);
} else { } else {
// Max retries exceeded // Max retries exceeded
progress.markFailed(task.id, result.error || "Unknown error"); progress.markFailed(task.id, result.error || "Unknown error");
throw new Error(`Task ${task.id} failed: ${result.error}`); throw new Error(`Task ${task.id} failed: ${result.error}`);
} }
} catch (error) { } catch (error) {
const errorMsg = error instanceof Error ? error.message : String(error); const errorMsg = error instanceof Error ? error.message : String(error);
progress.markFailed(task.id, errorMsg); progress.markFailed(task.id, errorMsg);
throw error; throw error;
} }
} }
} }
// ─── Save Reflection to File ──────────────────────────────────────────────── // ─── Save Reflection to File ────────────────────────────────────────────────
function saveReflectionToFile( function saveReflectionToFile(
sourceDir: string, sourceDir: string,
config: RalpiConfig, config: RalpiConfig,
reflection: Reflection, reflection: Reflection,
): void { ): void {
const reflectionsDir = path.join(sourceDir, config.paths.reflectionsDir); const reflectionsDir = path.join(sourceDir, config.paths.reflectionsDir);
ensureDir(reflectionsDir); ensureDir(reflectionsDir);
const filePath = path.join(reflectionsDir, `${reflection.taskId}.json`); const filePath = path.join(reflectionsDir, `${reflection.taskId}.json`);
writeFileSafe(filePath, JSON.stringify(reflection, null, 2)); writeFileSafe(filePath, JSON.stringify(reflection, null, 2));
} }
// ─── Helpers ───────────────────────────────────────────────────────────────── // ─── Helpers ─────────────────────────────────────────────────────────────────
function sleep(ms: number): Promise<void> { function sleep(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms)); return new Promise((resolve) => setTimeout(resolve, ms));
} }
// ─── Tool Call Formatting ──────────────────────────────────────────────── // ─── Tool Call Formatting ────────────────────────────────────────────────
@@ -411,34 +521,34 @@ function sleep(ms: number): Promise<void> {
* Format a tool call argument into a short label. * Format a tool call argument into a short label.
*/ */
function formatToolArg(name: string, args: unknown): string { function formatToolArg(name: string, args: unknown): string {
const a = args as Record<string, unknown>; const a = args as Record<string, unknown>;
switch (name) { switch (name) {
case "bash": case "bash":
return truncateMiddle(String(a.command ?? ""), 70); return truncateMiddle(String(a.command ?? ""), 70);
case "write": case "write":
case "read": case "read":
return truncateMiddle(String(a.path ?? ""), 60); return truncateMiddle(String(a.path ?? ""), 60);
case "edit": case "edit":
return truncateMiddle(String(a.path ?? ""), 60); return truncateMiddle(String(a.path ?? ""), 60);
case "grep": case "grep":
return `${a.pattern ?? "?"} — ${truncateMiddle( return `${a.pattern ?? "?"} — ${truncateMiddle(
String(a.path ?? ""), String(a.path ?? ""),
40, 40,
)}`; )}`;
case "find": case "find":
return `${a.path ?? "."} — ${a.glob ?? "*"}`; return `${a.path ?? "."} — ${a.glob ?? "*"}`;
case "ls": case "ls":
return truncateMiddle(String(a.path ?? "."), 60); return truncateMiddle(String(a.path ?? "."), 60);
default: default:
return name; return name;
} }
} }
/** /**
* Truncate a long string in the middle, keeping start and end visible. * Truncate a long string in the middle, keeping start and end visible.
*/ */
function truncateMiddle(s: string, maxLen: number): string { function truncateMiddle(s: string, maxLen: number): string {
if (s.length <= maxLen) return s; if (s.length <= maxLen) return s;
const half = Math.floor((maxLen - 3) / 2); const half = Math.floor((maxLen - 3) / 2);
return s.slice(0, half) + "…" + s.slice(s.length - half); return s.slice(0, half) + "…" + s.slice(s.length - half);
} }

10
src/types.ts
View File

@@ -160,6 +160,10 @@ export interface RalpiConfig {
/** Custom prompt suffix for reflection extraction */ /** Custom prompt suffix for reflection extraction */
reflectionPrompt: string; reflectionPrompt: string;
}; };
/** Parent session model to inherit in child agent sessions */
model?: unknown;
/** Parent session thinking level to inherit in child agent sessions */
thinkingLevel?: unknown;
} }
export const DEFAULT_CONFIG: RalpiConfig = { export const DEFAULT_CONFIG: RalpiConfig = {
@@ -168,9 +172,9 @@ export const DEFAULT_CONFIG: RalpiConfig = {
reflectionsDir: ".ralpi/reflections", reflectionsDir: ".ralpi/reflections",
}, },
execution: { execution: {
maxRetries: 3, maxRetries: 0,
retryDelayMs: 5000, retryDelayMs: 0,
timeoutMs: 30 * 60 * 1000, // 30 minutes timeoutMs: 0, // 0 = inherit Pi's own defaults (no ralpi-level timeout)
maxParallel: 3, maxParallel: 3,
}, },
prompts: { prompts: {

119
src/utils.ts
View File

@@ -82,32 +82,35 @@ export function findProgressFile(
// ─── Config ────────────────────────────────────────────────────────────────── // ─── Config ──────────────────────────────────────────────────────────────────
function parseSimpleYaml(content: string): Record<string, any> { /** Try to use the `yaml` package (real dependency in package.json).
const result: Record<string, any> = {}; * Falls back to a flat key:value parser when unavailable. */
const lines = content.split("\n"); const parseSimpleYaml: (content: string) => Record<string, any> = (() => {
try {
for (const line of lines) { // eslint-disable-next-line @typescript-eslint/no-var-requires
const trimmed = line.trim(); const { parse } = require("yaml");
if (!trimmed || trimmed.startsWith("#")) continue; return (content: string) => parse(content) ?? {};
} catch {
const match = trimmed.match(/^([^:]+):\s*(.+)$/); return (content: string) => {
if (match) { const result: Record<string, any> = {};
const key = match[1].trim(); for (const line of content.split("\n")) {
let value: string | boolean | number = match[2].trim(); const trimmed = line.trim();
if (!trimmed || trimmed.startsWith("#")) continue;
// Parse booleans const match = trimmed.match(/^([^:]+):\s*(.*)$/);
if (value === "true") value = true; if (match) {
else if (value === "false") value = false; const value = match[2].trim();
// Parse numbers if (value === "true") result[match[1].trim()] = true;
else if (/^\d+$/.test(value)) value = parseInt(value, 10); else if (value === "false") result[match[1].trim()] = false;
else if (/^\d+\.\d+$/.test(value)) value = parseFloat(value); else if (/^\d+$/.test(value))
result[match[1].trim()] = parseInt(value, 10);
result[key] = value; else if (/^\d+\.\d+$/.test(value))
} result[match[1].trim()] = parseFloat(value);
else result[match[1].trim()] = value;
}
}
return result;
};
} }
})();
return result;
}
/** /**
* Deep merge configuration objects * Deep merge configuration objects
@@ -129,25 +132,44 @@ function mergeConfig(
return result as RalpiConfig; return result as RalpiConfig;
} }
/** Path to the global ralpi config under the user's Pi home directory. */
const GLOBAL_CONFIG_PATH = path.join(
process.env.HOME || "/tmp",
".pi",
"ralpi",
"config.yaml",
);
/** /**
* Load configuration from .ralpi/config.yaml or return defaults * Load and merge config from global and project sources.
*
* Precedence (highest wins):
* 1. Project-level: `<projectDir>/.ralpi/config.yaml`
* 2. Global: `~/.pi/ralpi/config.yaml`
* 3. `DEFAULT_CONFIG` in `src/types.ts`
*/ */
export function loadConfig(projectDir: string): RalpiConfig { export function loadConfig(projectDir: string): RalpiConfig {
const configPath = path.join(projectDir, ".ralpi", "config.yaml"); // Start with defaults
const merged: RalpiConfig = { ...DEFAULT_CONFIG };
// Return defaults silently when config file does not exist // Layer 1: global config (~/.pi/ralpi/config.yaml)
if (!fs.existsSync(configPath)) { tryLoadConfigFile(GLOBAL_CONFIG_PATH, merged);
return { ...DEFAULT_CONFIG };
}
try { // Layer 2: project config (.ralpi/config.yaml) — overrides global
const content = fs.readFileSync(configPath, "utf-8"); tryLoadConfigFile(path.join(projectDir, ".ralpi", "config.yaml"), merged);
// Simple YAML parsing (key: value format)
const config = parseSimpleYaml(content); return merged;
return mergeConfig(DEFAULT_CONFIG, config);
} catch { /** Attempt to load a single config file and merge into `acc` in place. */
// Malformed config — fall back to defaults silently function tryLoadConfigFile(filePath: string, acc: RalpiConfig): void {
return { ...DEFAULT_CONFIG }; if (!fs.existsSync(filePath)) return;
try {
const content = fs.readFileSync(filePath, "utf-8");
const parsed = parseSimpleYaml(content);
Object.assign(acc, mergeConfig(acc, parsed));
} catch {
// Malformed config — skip silently
}
} }
} }
@@ -339,6 +361,8 @@ export async function runAgentSession(
onEvent?: (event: AgentSessionEvent) => void, onEvent?: (event: AgentSessionEvent) => void,
signal?: AbortSignal, signal?: AbortSignal,
sessionFile?: string, sessionFile?: string,
model?: unknown,
thinkingLevel?: unknown,
): Promise<{ ): Promise<{
success: boolean; success: boolean;
text: string; text: string;
@@ -361,10 +385,13 @@ export async function runAgentSession(
? fs.createWriteStream(sessionFile, { flags: "a" }) ? fs.createWriteStream(sessionFile, { flags: "a" })
: null; : null;
// Wire timeout via abort signal // Wire timeout via abort signal (only when set; 0 means inherit Pi's defaults)
const timeoutHandle = setTimeout(() => { let timeoutHandle: NodeJS.Timeout | null = null;
if (sessionRef?.session) sessionRef.session.agent.abort(); if (timeoutMs > 0) {
}, timeoutMs); timeoutHandle = setTimeout(() => {
if (sessionRef?.session) sessionRef.session.agent.abort();
}, timeoutMs);
}
const sessionRef: { const sessionRef: {
session?: Awaited<ReturnType<typeof createAgentSession>>["session"]; session?: Awaited<ReturnType<typeof createAgentSession>>["session"];
@@ -387,6 +414,8 @@ export async function runAgentSession(
sessionManager: SessionManager.inMemory(), sessionManager: SessionManager.inMemory(),
resourceLoader: loader, resourceLoader: loader,
tools: ["read", "bash", "edit", "write", "grep", "find", "ls"], tools: ["read", "bash", "edit", "write", "grep", "find", "ls"],
model: model as any,
thinkingLevel: thinkingLevel as any,
}); });
sessionRef.session = result.session; sessionRef.session = result.session;
@@ -437,7 +466,7 @@ export async function runAgentSession(
unsubscribe(); unsubscribe();
result.session.dispose(); result.session.dispose();
signal?.removeEventListener("abort", abortHandler); signal?.removeEventListener("abort", abortHandler);
clearTimeout(timeoutHandle); if (timeoutHandle) clearTimeout(timeoutHandle);
// Flush and close the event stream before returning // Flush and close the event stream before returning
if (eventStream) { if (eventStream) {
@@ -463,7 +492,7 @@ export async function runAgentSession(
events: [], // streamed to file events: [], // streamed to file
}; };
} catch (error) { } catch (error) {
clearTimeout(timeoutHandle); if (timeoutHandle) clearTimeout(timeoutHandle);
if (eventStream && !eventStream.destroyed) { if (eventStream && !eventStream.destroyed) {
eventStream.end(); eventStream.end();
} }