fix(phased-execution): resume failed session on retry, surface errors, thinking/compaction indicators
- retries now resume the failed executor's session (pre/post session tracking replaces the broken mtime-vs-ref check) - report recovery from the session file when the JSON stream loses its tail (child signaled mid-flush) - progress.mjs: thinking indicators (◐ thinking… / ◑ thought for Ns), compaction (⧉ …) and provider auto-retry lines; trims trailing whitespace so no blank lines after LLM text; exits 1 on truncated stream, model error, or missing final text (pi --mode json always exits 0 even on errors) - explicit ✗ ERROR lines + exit codes: 0 ok, 1 phase failed, 130/143 interrupted (INT/TERM traps; post-pipeline check covers the case where bash suppresses the INT trap after a job dies from SIGINT) - PIPESTATUS captured in a single statement (any following command resets it) - skills dir: .gitignore README.md so pi's skill scanner (which honors .gitignore) stops warning 'description is required'
This commit is contained in:
@@ -6,6 +6,15 @@
|
||||
//
|
||||
// Quiet mode still writes the report file (for harness logs); it only
|
||||
// suppresses the progress display.
|
||||
//
|
||||
// Shows: tool calls, assistant text, thinking indicators, context
|
||||
// compaction, and provider auto-retries.
|
||||
//
|
||||
// Exit code: 0 for a clean run (agent_end seen, no model errors, final
|
||||
// assistant text received). 1 when the stream ended without agent_end
|
||||
// (child crashed or was killed), the model reported an error, or no final
|
||||
// assistant text was produced. `pi --mode json` itself exits 0 even on
|
||||
// model errors, so the harness relies on this exit code to detect failures.
|
||||
|
||||
import { writeFileSync } from "node:fs";
|
||||
|
||||
@@ -95,12 +104,39 @@ let sessionId = "";
|
||||
let lastUsage = null;
|
||||
let lastAssistantText = "";
|
||||
let sawError = false;
|
||||
let sawAgentEnd = false;
|
||||
let thinkingSince = null;
|
||||
const failures = [];
|
||||
|
||||
// Streaming deltas (text / thinking / tool-call args) arrive inside
|
||||
// message_update.assistantMessageEvent; only thinking is surfaced live.
|
||||
function handleUpdate(ame) {
|
||||
if (!ame || typeof ame !== "object") return;
|
||||
switch (ame.type) {
|
||||
case "thinking_start":
|
||||
thinkingSince = Date.now();
|
||||
out(" " + c.dim("◐ thinking…"));
|
||||
break;
|
||||
case "thinking_end":
|
||||
if (thinkingSince !== null) {
|
||||
const secs = Math.max(1, Math.round((Date.now() - thinkingSince) / 1000));
|
||||
out(c.dim(` ◑ thought for ${secs}s`));
|
||||
thinkingSince = null;
|
||||
}
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
function handle(ev) {
|
||||
switch (ev.type) {
|
||||
case "session":
|
||||
sessionId = ev.id ?? "";
|
||||
break;
|
||||
case "agent_end":
|
||||
sawAgentEnd = true;
|
||||
break;
|
||||
case "tool_execution_start":
|
||||
out(" " + c.cyan("⏺ " + ev.toolName) + " " + c.dim(summarizeTool(ev.toolName, ev.args)));
|
||||
break;
|
||||
@@ -109,52 +145,104 @@ function handle(ev) {
|
||||
break;
|
||||
case "message_update":
|
||||
if (ev.usage) lastUsage = ev.usage;
|
||||
handleUpdate(ev.assistantMessageEvent);
|
||||
break;
|
||||
case "message_end": {
|
||||
const m = ev.message ?? {};
|
||||
if (m.role === "assistant") {
|
||||
const text = textOf(m);
|
||||
// Trim so trailing newlines in model output don't render as blank
|
||||
// lines after the message.
|
||||
const text = textOf(m).trim();
|
||||
if (text) {
|
||||
lastAssistantText = text;
|
||||
out(wrap(text, 2));
|
||||
}
|
||||
if (m.stopReason === "error") {
|
||||
if (m.stopReason === "error" || m.stopReason === "aborted") {
|
||||
sawError = true;
|
||||
out(c.red(" ✗ " + (m.errorMessage ?? "assistant error")));
|
||||
failures.push(`assistant ${m.stopReason}: ${m.errorMessage ?? ""}`);
|
||||
out(c.red(" ✗ " + (m.errorMessage ?? `assistant ${m.stopReason}`)));
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "compaction_start":
|
||||
out(" " + c.yellow("⧉ compacting context (" + (ev.reason ?? "auto") + ")…"));
|
||||
break;
|
||||
case "compaction_end":
|
||||
if (ev.aborted || ev.errorMessage) {
|
||||
sawError = true;
|
||||
failures.push(`compaction ${ev.aborted ? "aborted" : "failed"}: ${ev.errorMessage ?? ""}`);
|
||||
out(
|
||||
c.red(
|
||||
" ✗ compaction " + (ev.aborted ? "aborted" : "failed") + (ev.errorMessage ? ": " + ev.errorMessage : ""),
|
||||
),
|
||||
);
|
||||
} else {
|
||||
out(c.dim(" ✓ context compacted"));
|
||||
}
|
||||
break;
|
||||
case "auto_retry_start":
|
||||
out(
|
||||
c.yellow(
|
||||
` ↻ provider error, retrying ${ev.attempt}/${ev.maxAttempts} in ${Math.round((ev.delayMs ?? 0) / 1000)}s: ${truncate(
|
||||
ev.errorMessage ?? "",
|
||||
100,
|
||||
)}`,
|
||||
),
|
||||
);
|
||||
break;
|
||||
case "auto_retry_end":
|
||||
if (ev.success) {
|
||||
out(c.dim(" ✓ retry succeeded"));
|
||||
} else {
|
||||
sawError = true;
|
||||
failures.push(`retries exhausted: ${ev.finalError ?? ""}`);
|
||||
out(c.red(" ✗ retries exhausted: " + truncate(ev.finalError ?? "", 120)));
|
||||
}
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
function processLine(line) {
|
||||
line = line.trim();
|
||||
if (!line) return;
|
||||
let ev;
|
||||
try {
|
||||
ev = JSON.parse(line);
|
||||
} catch {
|
||||
out(c.yellow(" ? " + truncate(line, 120)));
|
||||
return;
|
||||
}
|
||||
handle(ev);
|
||||
}
|
||||
|
||||
let buf = "";
|
||||
process.stdin.setEncoding("utf8");
|
||||
process.stdin.on("data", (chunk) => {
|
||||
buf += chunk;
|
||||
let idx;
|
||||
while ((idx = buf.indexOf("\n")) >= 0) {
|
||||
const line = buf.slice(0, idx).trim();
|
||||
const line = buf.slice(0, idx);
|
||||
buf = buf.slice(idx + 1);
|
||||
if (!line) continue;
|
||||
let ev;
|
||||
try {
|
||||
ev = JSON.parse(line);
|
||||
} catch {
|
||||
out(c.yellow(" ? " + truncate(line, 120)));
|
||||
continue;
|
||||
}
|
||||
handle(ev);
|
||||
processLine(line);
|
||||
}
|
||||
});
|
||||
|
||||
process.stdin.on("end", () => {
|
||||
// A trailing line without a final newline (child killed mid-flush) still
|
||||
// counts — process whatever is left in the buffer.
|
||||
if (buf.trim()) {
|
||||
processLine(buf);
|
||||
buf = "";
|
||||
}
|
||||
|
||||
writeFileSync(
|
||||
reportPath,
|
||||
lastAssistantText ? lastAssistantText + "\n" : "(no final assistant message — see the .err log)\n",
|
||||
);
|
||||
|
||||
const u = lastUsage ?? {};
|
||||
const bits = [];
|
||||
if (u.input) bits.push(`↑${u.input}`);
|
||||
@@ -162,5 +250,17 @@ process.stdin.on("end", () => {
|
||||
if (u.cacheRead) bits.push(`R${u.cacheRead}`);
|
||||
if (sessionId) bits.push("session " + String(sessionId).slice(0, 8));
|
||||
if (bits.length) out(c.dim(" · " + bits.join(" ")));
|
||||
out(sawError ? c.red(" · child finished with errors") : c.green(" · child done"));
|
||||
|
||||
if (!sawAgentEnd) {
|
||||
failures.push("stream ended without agent_end (child crashed or was killed)");
|
||||
out(c.red(" ✗ stream ended without agent_end — child crashed or was killed (see .err log)"));
|
||||
} else if (sawError) {
|
||||
out(c.red(" ✗ child finished with errors: " + failures.join("; ")));
|
||||
} else if (!lastAssistantText) {
|
||||
failures.push("no final assistant text");
|
||||
out(c.red(" ✗ no final assistant text — no report available (see .err log)"));
|
||||
} else {
|
||||
out(c.green(" · child done"));
|
||||
}
|
||||
process.exitCode = sawError || !sawAgentEnd || !lastAssistantText ? 1 : 0;
|
||||
});
|
||||
|
||||
Reference in New Issue
Block a user