diff --git a/.github/workflows/no-mistakes-required.yml b/.github/workflows/no-mistakes-required.yml index 65c5b2d1..609fde8d 100644 --- a/.github/workflows/no-mistakes-required.yml +++ b/.github/workflows/no-mistakes-required.yml @@ -16,6 +16,7 @@ on: permissions: contents: read + pull-requests: read # GitHub concurrency groups retain at most one pending run, replacing older # pending runs even when cancel-in-progress is false. Give body-bearing events @@ -40,16 +41,25 @@ jobs: steps: - name: Verify no-mistakes signature in PR body env: - PR_BODY: ${{ github.event.pull_request.body }} + GH_TOKEN: ${{ github.token }} PR_AUTHOR: ${{ github.event.pull_request.user.login }} PR_NUMBER: ${{ github.event.pull_request.number }} run: | set -eu marker='Updates from [git push no-mistakes](https://github.com/kunchenguid/no-mistakes)' - if printf '%s' "${PR_BODY:-}" | grep -qF -- "$marker"; then - echo "Found no-mistakes signature in PR #${PR_NUMBER} body." - exit 0 - fi + max_attempts=5 + attempt=1 + while [ "$attempt" -le "$max_attempts" ]; do + pr_body="$(gh api "repos/${GITHUB_REPOSITORY}/pulls/${PR_NUMBER}" --jq '.body // ""')" + if printf '%s' "$pr_body" | grep -qF -- "$marker"; then + echo "Found no-mistakes signature in PR #${PR_NUMBER} body." + exit 0 + fi + if [ "$attempt" -lt "$max_attempts" ]; then + sleep 2 + fi + attempt=$((attempt + 1)) + done { echo "::error::This PR was not raised through no-mistakes." echo diff --git a/README.md b/README.md index 7b061e82..5d40953b 100644 --- a/README.md +++ b/README.md @@ -151,7 +151,7 @@ After installing from npm, the skill is available under the installed package di ``` - **Incremental commits** - each successful iteration is a separate unsigned git commit, so you can cherry-pick or revert individual changes without GPG or SSH signing prompts blocking the run; if `git commit` fails, gnhf preserves the uncommitted work and asks the next agent iteration to repair it -- **Failure handling** - failed iterations are rolled back with `git reset --hard` except commit failures, which preserve uncommitted work for repair; agent-reported failures proceed to the next iteration immediately, retryable hard agent errors use exponential backoff, and permanent agent errors such as Claude low credit balance abort immediately and print the run log path. Complete no-op iterations are reported as failures and count toward the consecutive-failure abort limit. If the run exits with a pending commit failure, the exit summary warns that uncommitted changes were left for repair. +- **Failure handling** - failed iterations are rolled back with `git reset --hard` except commit failures, which preserve uncommitted work for repair; agent-reported failures proceed to the next iteration immediately, retryable hard agent errors use exponential backoff, and permanent agent errors such as Claude low credit balance abort immediately and print the run log path. When an agent completes a turn without a final answer, gnhf nudges it once inside the same session to continue before recording a failure, and records the nudge in the run log. Covered: OpenCode, Rovo Dev, and ACP targets reuse their live session, and `claude` and `codex` resume the exact session the empty turn used - they skip the nudge when that session cannot be identified or resumed (no session id reported, or agent args such as `--no-session-persistence` or `--ephemeral`). Not covered: `pi`, because gnhf runs it with `--no-session`, and `copilot`, because it has no verified exact-session resume contract ([#193](https://github.com/kunchenguid/gnhf/issues/193)). Whenever the nudge is skipped the failure names the reason, because a retry that cannot reach the original session could only invent a summary. Complete no-op iterations are reported as failures and count toward the consecutive-failure abort limit. If the run exits with a pending commit failure, the exit summary warns that uncommitted changes were left for repair. - **Runtime caps** - `--max-iterations` stops before the next iteration begins, `--max-tokens` can abort mid-iteration once reported usage reaches the cap, and `--stop-when` ends the loop after an iteration whose agent output reports the natural-language condition is met unless a commit failure needs repair first; resumed runs reuse the saved stop condition unless you pass a new value, or `--stop-when ""` to clear it; pending commit-failure repair work is preserved and other uncommitted work is rolled back, and in the interactive TUI the final state remains visible until you press Ctrl+C to exit - **Iteration finalization** - agents are expected to finish validation, stop any background processes they started, and only then emit the final JSON result for the iteration - **Graceful interrupts** - in the interactive TUI, the first Ctrl+C requests a graceful stop and lets the current iteration finish (or ends backoff early), the second Ctrl+C force-stops immediately, and `SIGTERM` also force-stops immediately diff --git a/e2e/e2e.test.ts b/e2e/e2e.test.ts index 434f5d22..25bb476b 100644 --- a/e2e/e2e.test.ts +++ b/e2e/e2e.test.ts @@ -234,6 +234,118 @@ describe("gnhf e2e", () => { expect(debugEvents).toContain("run:complete"); }, 30_000); + it("recovers one completed empty turn in the same session and records the nudge", async () => { + const cwd = createRepo(); + tempDirs.push(cwd); + const logDir = mkdtempSync(join(tmpdir(), "gnhf-e2e-logs-")); + tempDirs.push(logDir); + const mockLogPath = join(logDir, "mock-opencode.jsonl"); + + const result = await runCli( + cwd, + [ + "recover the empty response", + "--agent", + "opencode", + "--max-iterations", + "1", + "--prevent-sleep", + "off", + ], + { + env: { + ...createTestEnv(mockLogPath, tempDirs), + GNHF_MOCK_OPENCODE_EMPTY_ONCE: "1", + }, + }, + ); + + expect(result.code).toBe(0); + expect(git(["rev-list", "--count", "HEAD"], cwd)).toBe("2"); + + const messages = readJsonLines(mockLogPath).filter( + (entry) => entry.event === "message:start", + ); + expect(messages).toHaveLength(2); + expect(messages[1]?.sessionId).toBe(messages[0]?.sessionId); + expect(messages[1]?.prompt).toBe( + "You did not produce a final answer. Continue and provide your final summary now.", + ); + + const debugLogPath = findRunLogPath(cwd); + const continuationEvents = readJsonLines(debugLogPath).filter( + (entry) => entry.event === "opencode:output:continuation", + ); + expect(continuationEvents).toEqual([ + expect.objectContaining({ + attempt: 1, + sessionId: messages[0]?.sessionId, + prompt: + "You did not produce a final answer. Continue and provide your final summary now.", + }), + ]); + expect( + readFileSync(join(dirname(debugLogPath), "notes.md"), "utf-8"), + ).toContain("**Summary:** mocked objective complete"); + }, 30_000); + + it("fails after one continuation when both completed turns are empty", async () => { + const cwd = createRepo(); + tempDirs.push(cwd); + const logDir = mkdtempSync(join(tmpdir(), "gnhf-e2e-logs-")); + tempDirs.push(logDir); + const mockLogPath = join(logDir, "mock-opencode.jsonl"); + + const result = await runCli( + cwd, + [ + "stop after one empty-response retry", + "--agent", + "opencode", + "--max-iterations", + "1", + "--prevent-sleep", + "off", + ], + { + env: { + ...createTestEnv(mockLogPath, tempDirs), + GNHF_MOCK_OPENCODE_ALWAYS_EMPTY: "1", + }, + }, + ); + + expect(result.code).toBe(0); + expect(git(["rev-list", "--count", "HEAD"], cwd)).toBe("1"); + + const messages = readJsonLines(mockLogPath).filter( + (entry) => entry.event === "message:start", + ); + expect(messages).toHaveLength(2); + expect(messages[1]?.sessionId).toBe(messages[0]?.sessionId); + expect(messages[1]?.prompt).toBe( + "You did not produce a final answer. Continue and provide your final summary now.", + ); + + const debugLogPath = findRunLogPath(cwd); + const debugEntries = readJsonLines(debugLogPath); + expect( + debugEntries.filter( + (entry) => entry.event === "opencode:output:continuation", + ), + ).toHaveLength(1); + const iterationEnd = debugEntries.find( + (entry) => entry.event === "iteration:end", + ); + expect(iterationEnd).toMatchObject({ + success: false, + summary: "OpenCode produced no final answer", + }); + expect( + readFileSync(join(dirname(debugLogPath), "notes.md"), "utf-8"), + ).toContain("[ERROR] OpenCode produced no final answer"); + }, 30_000); + it("runs on the current branch and pushes each successful iteration", async () => { const cwd = createRepo(); tempDirs.push(cwd); diff --git a/e2e/fixtures/mock-opencode-server.mjs b/e2e/fixtures/mock-opencode-server.mjs index 8abf5d30..de05b2b5 100755 --- a/e2e/fixtures/mock-opencode-server.mjs +++ b/e2e/fixtures/mock-opencode-server.mjs @@ -157,6 +157,34 @@ function emitCompletedEvents(sessionId, summary) { return output; } +function emitEmptyCompletedEvents(sessionId) { + broadcast({ + directory: "/repo", + payload: { + type: "message.updated", + properties: { + sessionID: sessionId, + info: { + id: "msg-empty-1", + role: "assistant", + tokens: { + input: 3, + output: 0, + cache: { read: 0, write: 0 }, + }, + }, + }, + }, + }); + broadcast({ + directory: "/repo", + payload: { + type: "session.idle", + properties: { sessionID: sessionId }, + }, + }); +} + function applyWorkspaceChange(sessionId) { const session = sessions.get(sessionId); if (!session?.directory) return; @@ -282,6 +310,18 @@ const server = createServer(async (req, res) => { return; } + const shouldEmitEmpty = + process.env.GNHF_MOCK_OPENCODE_ALWAYS_EMPTY === "1" || + (process.env.GNHF_MOCK_OPENCODE_EMPTY_ONCE === "1" && + !session?.emittedEmptyTurn); + if (shouldEmitEmpty && session) { + session.emittedEmptyTurn = true; + emitEmptyCompletedEvents(sessionId); + res.writeHead(204); + res.end(); + return; + } + applyWorkspaceChange(sessionId); emitCompletedEvents(sessionId, "mocked objective complete"); res.writeHead(204); diff --git a/src/core/agents/acp.test.ts b/src/core/agents/acp.test.ts index 25947efe..134c147d 100644 --- a/src/core/agents/acp.test.ts +++ b/src/core/agents/acp.test.ts @@ -516,6 +516,169 @@ describe("AcpAgent", () => { expect(result.output).toEqual(VALID_OUTPUT); }); + it("nudges the same session once when a completed turn produced no output text", async () => { + const { runtime, calls } = createFakeRuntime([ + { events: [], result: { status: "completed" } }, + { + events: [textDelta(JSON.stringify(VALID_OUTPUT))], + result: { status: "completed" }, + }, + ]); + const agent = makeAgent(runtime); + const onUsage = vi.fn(); + + const result = await agent.run("p", "/w", { onUsage }); + + expect(result.output).toEqual(VALID_OUTPUT); + expect(calls.startTurnInputs).toHaveLength(2); + expect(calls.ensureSessionInputs).toHaveLength(1); + expect(calls.startTurnInputs[1]?.text).toContain( + "You did not produce a final answer", + ); + expect(calls.startTurnInputs[1]?.text).not.toContain( + "gnhf final output contract", + ); + expect(onUsage).toHaveBeenLastCalledWith(result.usage); + }); + + it("nudges when a completed turn produced only whitespace output", async () => { + const { runtime, calls } = createFakeRuntime([ + { + events: [textDelta(" \n\t")], + result: { status: "completed" }, + }, + { + events: [textDelta(JSON.stringify(VALID_OUTPUT))], + result: { status: "completed" }, + }, + ]); + const agent = makeAgent(runtime); + + const result = await agent.run("p", "/w"); + + expect(result.output).toEqual(VALID_OUTPUT); + expect(calls.startTurnInputs).toHaveLength(2); + expect(calls.startTurnInputs[1]?.text).toContain( + "You did not produce a final answer", + ); + }); + + it("reports usage across both the empty turn and its continuation", async () => { + // The first turn streams only reasoning, so it completes with no output + // text while still burning output tokens that must survive into the total. + const firstTurnText = "thinking hard"; + const secondTurnText = JSON.stringify(VALID_OUTPUT); + const { runtime } = createFakeRuntime([ + { + events: [textDelta(firstTurnText, "thought")], + result: { status: "completed" }, + }, + { + events: [textDelta(secondTurnText)], + result: { status: "completed" }, + }, + ]); + const agent = makeAgent(runtime); + + const result = await agent.run("p", "/w"); + + expect(result.usage.outputTokens).toBe( + Math.ceil(firstTurnText.length / 4) + + Math.ceil(secondTurnText.length / 4), + ); + }); + + it("does not add fallback input to a cumulative continuation update", async () => { + const { runtime } = createFakeRuntime([ + { + events: [ + { + type: "status", + text: "u", + tag: "usage_update", + used: 100, + size: 1000, + }, + textDelta(JSON.stringify(VALID_OUTPUT)), + ], + result: { status: "completed" }, + }, + { events: [], result: { status: "completed" } }, + { + events: [ + { + type: "status", + text: "u", + tag: "usage_update", + used: 160, + size: 1000, + }, + textDelta(JSON.stringify(VALID_OUTPUT)), + ], + result: { status: "completed" }, + }, + ]); + const agent = makeAgent(runtime); + + await agent.run("warmup", "/w"); + const onUsage = vi.fn(); + const result = await agent.run("recover", "/w", { onUsage }); + + expect(result.usage.inputTokens).toBe(60); + expect(onUsage).toHaveBeenLastCalledWith(result.usage); + }); + + it("continues when tool work leaves no final output message", async () => { + const { runtime, calls } = createFakeRuntime([ + { + events: [ + textDelta("I will inspect the file."), + { type: "tool_call", text: "Read file", toolCallId: "1" }, + ], + result: { status: "completed" }, + }, + { + events: [textDelta(JSON.stringify(VALID_OUTPUT))], + result: { status: "completed" }, + }, + ]); + const agent = makeAgent(runtime); + + const result = await agent.run("p", "/w"); + + expect(result.output).toEqual(VALID_OUTPUT); + expect(calls.startTurnInputs).toHaveLength(2); + }); + + it("rejects after the ACP continuation is also empty", async () => { + const { runtime, calls } = createFakeRuntime([ + { events: [], result: { status: "completed" } }, + { events: [], result: { status: "completed" } }, + ]); + const agent = makeAgent(runtime); + + await expect(agent.run("p", "/w")).rejects.toThrow( + "ACP agent returned no output text", + ); + expect(calls.startTurnInputs).toHaveLength(2); + }); + + it("does not nudge when the empty turn failed rather than completed", async () => { + const { runtime, calls } = createFakeRuntime([ + { + events: [], + result: { + status: "failed", + error: { message: "transient", retryable: true }, + }, + }, + ]); + const agent = makeAgent(runtime); + + await expect(agent.run("p", "/w")).rejects.toThrow("transient"); + expect(calls.startTurnInputs).toHaveLength(1); + }); + it("throws PermanentAgentError when the runtime reports a non-retryable failure", async () => { const { runtime } = createFakeRuntime([ { diff --git a/src/core/agents/acp.ts b/src/core/agents/acp.ts index 4737aac9..a8f50d9f 100644 --- a/src/core/agents/acp.ts +++ b/src/core/agents/acp.ts @@ -9,8 +9,14 @@ import { type AcpRuntimeTurnResult, type AcpxRuntime, } from "acpx/runtime"; +import type { WriteStream } from "node:fs"; import { appendDebugLog, serializeError } from "../debug-log.js"; import { redactAcpTargetForLogs } from "../config.js"; +import { + addTokenUsage, + EmptyAgentResponseError, + runTurnWithEmptyResponseRetry, +} from "./empty-response.js"; import { parseAgentJson } from "./json-extract.js"; import { PermanentAgentError, @@ -19,6 +25,8 @@ import { type AgentOutputSchema, type AgentResult, type AgentRunOptions, + type OnMessage, + type OnUsage, type TokenUsage, } from "./types.js"; @@ -207,6 +215,68 @@ export class AcpAgent implements Agent { } this.handle = handle; + const logStream = logPath ? createWriteStream(logPath) : null; + const turnUsageUpdates: boolean[] = []; + try { + // The ACP session is persistent, so a continuation turn still sees the + // first turn's reasoning, tool calls, and the output contract that + // buildAcpPrompt already delivered - the nudge stays bare. + return await runTurnWithEmptyResponseRetry({ + logEvent: "acp:turn:continuation", + logFields: { + target: redactAcpTargetForLogs(this.target), + sessionKey: this.runId, + }, + onUsage, + signal, + combineUsage: (firstTurnUsage, continuationUsage) => { + const combined = addTokenUsage(firstTurnUsage, continuationUsage); + if (!turnUsageUpdates[0] && turnUsageUpdates[1]) { + combined.inputTokens = continuationUsage.inputTokens; + if (!continuationUsage.estimated) { + delete combined.estimated; + } + } + return combined; + }, + initialText: buildAcpPrompt(prompt, this.schema), + runTurn: (text, onTurnUsage) => { + const turnIndex = turnUsageUpdates.length; + turnUsageUpdates.push(false); + return this.runTurn({ + runtime, + handle, + text, + cwd, + signal, + onMessage, + onUsage: onTurnUsage, + onUsageUpdate: () => { + turnUsageUpdates[turnIndex] = true; + }, + logStream, + }); + }, + }); + } finally { + logStream?.end(); + } + } + + private async runTurn(params: { + runtime: AcpxRuntimeLike; + handle: AcpRuntimeHandle; + text: string; + cwd: string; + signal?: AbortSignal; + onMessage?: OnMessage; + onUsage?: OnUsage; + onUsageUpdate?: () => void; + logStream: WriteStream | null; + }): Promise { + const { runtime, handle, text: acpPrompt, cwd, signal, logStream } = params; + const { onMessage, onUsage, onUsageUpdate } = params; + const requestId = randomUUID(); appendDebugLog("acp:turn:start", { target: redactAcpTargetForLogs(this.target), @@ -215,7 +285,6 @@ export class AcpAgent implements Agent { cwd, }); - const acpPrompt = buildAcpPrompt(prompt, this.schema); const promptTokenEstimate = estimateTokens(acpPrompt.length); const startedAt = Date.now(); @@ -261,12 +330,9 @@ export class AcpAgent implements Agent { // the turn, so this is the primary candidate to JSON.parse - separating // it from intermediate prose like "Let me examine the code...". let lastOutputMessage = ""; - // Concatenation of every output-stream chunk in the turn, used as a - // fallback when `lastOutputMessage` doesn't parse (e.g. when the agent - // streams the entire response as one continuous message without any - // tool_call to break it up). + // Concatenation of output-stream chunks since the most recent tool-call + // boundary, used as a fallback when `lastOutputMessage` doesn't parse. let outputBuf = ""; - const logStream = logPath ? createWriteStream(logPath) : null; const computeUsage = (): TokenUsage => { const usedDelta = Math.max(0, latestUsed - iterationStartUsed); @@ -300,148 +366,150 @@ export class AcpAgent implements Agent { pendingStream = null; }; - try { - // Surface an initial input-token estimate immediately so the renderer - // shows non-zero numbers as soon as the iteration starts. - onUsage?.(computeUsage()); - - try { - for await (const event of turn.events) { - logStream?.write(`${JSON.stringify(event)}\n`); - - if (event.type === "text_delta") { - const stream = event.stream ?? "output"; - const text = event.text; - if (!text) continue; - if (pendingStream !== null && pendingStream !== stream) { - flushPendingMessage(); - } - pendingStream = stream; - pendingMessage += text; - // Count both output and thought streams toward output tokens - - // reasoning is real generated text that consumes tokens. Without - // this, agents that stream reasoning before answering (Gemini, - // GPT-5, etc.) leave the renderer at 0 output tokens for the - // entire thinking phase. outputBuf stays output-only because it - // is used for JSON parsing and reasoning text would corrupt it. - if (stream === "output") { - outputBuf += text; - } - agentOutputChars += text.length; - onUsage?.(computeUsage()); - continue; - } + // Surface an initial input-token estimate immediately so the renderer + // shows non-zero numbers as soon as the iteration starts. + onUsage?.(computeUsage()); - if (event.type === "tool_call") { - // A tool_call ends the in-flight assistant message - flush - // whatever prose the assistant streamed so far, but don't surface - // the tool_call text itself. Tool descriptions like - // "tool call (completed)" are noisy and not useful in the TUI; - // the user wants to see assistant prose, not mechanics. + try { + for await (const event of turn.events) { + logStream?.write(`${JSON.stringify(event)}\n`); + + if (event.type === "text_delta") { + const stream = event.stream ?? "output"; + const text = event.text; + if (!text) continue; + if (pendingStream !== null && pendingStream !== stream) { flushPendingMessage(); - // Each tool call (not its many tool_call_update follow-ups) bumps - // the input-cost heuristic so the fallback estimate scales with - // actual work. Adapters tag the initial event "tool_call" and - // later updates "tool_call_update" - count only the former. - if (event.tag === "tool_call") { - toolCallCount += 1; - if (!usageUpdateReceived) onUsage?.(computeUsage()); - } - continue; } + pendingStream = stream; + pendingMessage += text; + // Count both output and thought streams toward output tokens - + // reasoning is real generated text that consumes tokens. Without + // this, agents that stream reasoning before answering (Gemini, + // GPT-5, etc.) leave the renderer at 0 output tokens for the + // entire thinking phase. outputBuf stays output-only because it + // is used for JSON parsing and reasoning text would corrupt it. + if (stream === "output") { + outputBuf += text; + } + agentOutputChars += text.length; + onUsage?.(computeUsage()); + continue; + } - if (event.type === "status") { - // Status events are metadata (usage_update, mode change, etc.) - // and fire frequently mid-stream. Don't surface their text via - // onMessage - it would flicker over the actual assistant message - // the user is reading. - if (typeof event.used === "number" && event.used !== latestUsed) { - latestUsed = event.used; - this.lastReportedUsed = latestUsed; - usageUpdateReceived = true; - onUsage?.(computeUsage()); - } - continue; + if (event.type === "tool_call") { + // A tool_call ends the in-flight assistant message - flush + // whatever prose the assistant streamed so far, but don't surface + // the tool_call text itself. Tool descriptions like + // "tool call (completed)" are noisy and not useful in the TUI; + // the user wants to see assistant prose, not mechanics. + flushPendingMessage(); + lastOutputMessage = ""; + outputBuf = ""; + // Each tool call (not its many tool_call_update follow-ups) bumps + // the input-cost heuristic so the fallback estimate scales with + // actual work. Adapters tag the initial event "tool_call" and + // later updates "tool_call_update" - count only the former. + if (event.tag === "tool_call") { + toolCallCount += 1; + if (!usageUpdateReceived) onUsage?.(computeUsage()); } + continue; } - flushPendingMessage(); - } catch (error) { - if (signal?.aborted || isAbortError(error)) { - await turn.cancel({ reason: "gnhf-aborted" }).catch(() => undefined); - appendDebugLog("acp:turn:aborted", { - target: redactAcpTargetForLogs(this.target), - requestId, - elapsedMs: Date.now() - startedAt, - }); - throw createAbortError(); + + if (event.type === "status") { + // Status events are metadata (usage_update, mode change, etc.) + // and fire frequently mid-stream. Don't surface their text via + // onMessage - it would flicker over the actual assistant message + // the user is reading. + if (typeof event.used === "number" && event.used !== latestUsed) { + latestUsed = event.used; + this.lastReportedUsed = latestUsed; + usageUpdateReceived = true; + onUsageUpdate?.(); + onUsage?.(computeUsage()); + } + continue; } - appendDebugLog("acp:turn:stream-error", { + } + flushPendingMessage(); + } catch (error) { + if (signal?.aborted || isAbortError(error)) { + await turn.cancel({ reason: "gnhf-aborted" }).catch(() => undefined); + appendDebugLog("acp:turn:aborted", { target: redactAcpTargetForLogs(this.target), requestId, elapsedMs: Date.now() - startedAt, - error: serializeAcpErrorForLog(error, this.target), }); - throw redactAcpErrorForThrow(error, this.target); + throw createAbortError(); } - - const result: AcpRuntimeTurnResult = await turn.result; - appendDebugLog("acp:turn:result", { + appendDebugLog("acp:turn:stream-error", { target: redactAcpTargetForLogs(this.target), requestId, - status: result.status, - stopReason: - result.status === "completed" || result.status === "cancelled" - ? result.stopReason - : undefined, - errorCode: result.status === "failed" ? result.error.code : undefined, - retryable: - result.status === "failed" ? result.error.retryable : undefined, elapsedMs: Date.now() - startedAt, - outputLength: outputBuf.length, + error: serializeAcpErrorForLog(error, this.target), }); + throw redactAcpErrorForThrow(error, this.target); + } - if (result.status === "cancelled") { - throw createAbortError(); - } - if (result.status === "failed") { - const message = redactRawAcpTargetInString( - result.error.message || "ACP turn failed", - this.target, - ); - if (result.error.retryable === false) { - throw new PermanentAgentError( - message, - result.error.code ?? "ACP_TURN_FAILED", - ); - } - throw new Error(message); - } - - if (lastOutputMessage.length === 0 && outputBuf.length === 0) { - throw new Error("ACP agent returned no output text"); - } + const result: AcpRuntimeTurnResult = await turn.result; + appendDebugLog("acp:turn:result", { + target: redactAcpTargetForLogs(this.target), + requestId, + status: result.status, + stopReason: + result.status === "completed" || result.status === "cancelled" + ? result.stopReason + : undefined, + errorCode: result.status === "failed" ? result.error.code : undefined, + retryable: + result.status === "failed" ? result.error.retryable : undefined, + elapsedMs: Date.now() - startedAt, + outputLength: outputBuf.length, + }); - // Try the most recent assistant message first - that's where the - // structured answer is supposed to live. Fall back to extracting a - // JSON object out of the full output stream if the last message - // alone doesn't parse (e.g. the agent streamed prose and JSON in - // one uninterrupted message, so we have to dig the JSON out). - let parsed = parseAgentJson(lastOutputMessage); - if (parsed === null && outputBuf !== lastOutputMessage) { - parsed = parseAgentJson(outputBuf); - } - if (parsed === null) { - const preview = (lastOutputMessage || outputBuf).slice(0, 200); - throw new Error( - `Failed to parse ACP agent output as JSON. Last assistant message started with: ${JSON.stringify(preview)}`, + if (result.status === "cancelled") { + throw createAbortError(); + } + if (result.status === "failed") { + const message = redactRawAcpTargetInString( + result.error.message || "ACP turn failed", + this.target, + ); + if (result.error.retryable === false) { + throw new PermanentAgentError( + message, + result.error.code ?? "ACP_TURN_FAILED", ); } + throw new Error(message); + } - const output = validateAgentOutput(parsed, this.schema); - return { output, usage: computeUsage() }; - } finally { - logStream?.end(); + if (!lastOutputMessage.trim() && !outputBuf.trim()) { + throw new EmptyAgentResponseError("ACP agent returned no output text", { + turnCompleted: result.status === "completed", + usage: computeUsage(), + }); } + + // Try the most recent assistant message first - that's where the + // structured answer is supposed to live. Fall back to extracting a + // JSON object out of the full output stream if the last message + // alone doesn't parse (e.g. the agent streamed prose and JSON in + // one uninterrupted message, so we have to dig the JSON out). + let parsed = parseAgentJson(lastOutputMessage); + if (parsed === null && outputBuf !== lastOutputMessage) { + parsed = parseAgentJson(outputBuf); + } + if (parsed === null) { + const preview = (lastOutputMessage || outputBuf).slice(0, 200); + throw new Error( + `Failed to parse ACP agent output as JSON. Last assistant message started with: ${JSON.stringify(preview)}`, + ); + } + + const output = validateAgentOutput(parsed, this.schema); + return { output, usage: computeUsage() }; } async close(): Promise { diff --git a/src/core/agents/claude.test.ts b/src/core/agents/claude.test.ts index 556876cf..ed2fb85d 100644 --- a/src/core/agents/claude.test.ts +++ b/src/core/agents/claude.test.ts @@ -6,11 +6,19 @@ vi.mock("node:child_process", () => ({ spawn: vi.fn(), })); +vi.mock("../debug-log.js", () => ({ + appendDebugLog: vi.fn(), + initDebugLog: vi.fn(), + serializeError: vi.fn(), +})); + import { execFileSync, spawn } from "node:child_process"; +import { appendDebugLog } from "../debug-log.js"; import { ClaudeAgent } from "./claude.js"; import { PermanentAgentError, buildAgentOutputSchema } from "./types.js"; const mockSpawn = vi.mocked(spawn); +const mockAppendDebugLog = vi.mocked(appendDebugLog); const STOP_SCHEMA = buildAgentOutputSchema({ includeStopField: true, @@ -1171,7 +1179,206 @@ describe("ClaudeAgent", () => { await expect(promise).rejects.toThrow("claude reported error"); }); - it("rejects when structured_output is null", async () => { + it("resumes the same session once with the bare nudge when structured_output is null", async () => { + const first = createMockProcess(); + const second = createMockProcess(); + mockSpawn.mockReturnValueOnce(first).mockReturnValueOnce(second); + + const promise = agent.run("prompt", "/cwd"); + + emitLine(first, { + type: "system", + subtype: "init", + session_id: "session-abc", + }); + emitLine(first, { + type: "result", + subtype: "success", + is_error: false, + session_id: "session-abc", + usage: { + input_tokens: 10, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + output_tokens: 5, + }, + structured_output: null, + }); + first.emit("close", 0); + + await vi.waitFor(() => expect(mockSpawn).toHaveBeenCalledTimes(2)); + + emitLine(second, { + type: "result", + subtype: "success", + is_error: false, + session_id: "session-abc", + usage: { + input_tokens: 3, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + output_tokens: 2, + }, + structured_output: { + success: true, + summary: "recovered", + key_changes_made: [], + key_learnings: [], + }, + }); + second.emit("close", 0); + + await expect(promise).resolves.toMatchObject({ + output: { success: true, summary: "recovered" }, + usage: { inputTokens: 13, outputTokens: 7 }, + }); + + const continuationArgs = mockSpawn.mock.calls[1]![1] as string[]; + expect(continuationArgs).toContain("--resume"); + expect(continuationArgs[continuationArgs.indexOf("--resume") + 1]).toBe( + "session-abc", + ); + expect(continuationArgs[continuationArgs.indexOf("-p") + 1]).toBe( + "You did not produce a final answer. Continue and provide your final summary now.", + ); + expect(continuationArgs).toContain("--json-schema"); + expect(mockAppendDebugLog).toHaveBeenCalledWith( + "claude:output:continuation", + expect.objectContaining({ attempt: 1 }), + ); + }); + + it.each(["-c", "--continue"])( + "replaces configured %s with the captured session id for recovery", + async (continuationArg) => { + const first = createMockProcess(); + const second = createMockProcess(); + mockSpawn.mockReturnValueOnce(first).mockReturnValueOnce(second); + const continuingAgent = new ClaudeAgent({ + extraArgs: [continuationArg], + }); + + const promise = continuingAgent.run("prompt", "/cwd"); + + emitLine(first, { + type: "result", + subtype: "success", + is_error: false, + session_id: "session-continued", + usage: { + input_tokens: 1, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + output_tokens: 1, + }, + structured_output: null, + }); + first.emit("close", 0); + + await vi.waitFor(() => expect(mockSpawn).toHaveBeenCalledTimes(2)); + + const initialArgs = mockSpawn.mock.calls[0]![1] as string[]; + const continuationArgs = mockSpawn.mock.calls[1]![1] as string[]; + expect(initialArgs).toContain(continuationArg); + expect(continuationArgs).not.toContain(continuationArg); + expect(continuationArgs).toContain("--resume"); + expect(continuationArgs[continuationArgs.indexOf("--resume") + 1]).toBe( + "session-continued", + ); + + emitLine(second, { + type: "result", + subtype: "success", + is_error: false, + session_id: "session-continued", + usage: { + input_tokens: 1, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + output_tokens: 1, + }, + structured_output: { + success: true, + summary: "recovered", + key_changes_made: [], + key_learnings: [], + }, + }); + second.emit("close", 0); + + await expect(promise).resolves.toMatchObject({ + output: { summary: "recovered" }, + }); + }, + ); + + it("rejects after exactly one continuation when structured_output is still null", async () => { + const first = createMockProcess(); + const second = createMockProcess(); + mockSpawn.mockReturnValueOnce(first).mockReturnValueOnce(second); + + const promise = agent.run("prompt", "/cwd"); + + const emptyResult = { + type: "result", + subtype: "success", + is_error: false, + session_id: "session-abc", + usage: { + input_tokens: 0, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + output_tokens: 0, + }, + structured_output: null, + }; + + emitLine(first, emptyResult); + first.emit("close", 0); + + await vi.waitFor(() => expect(mockSpawn).toHaveBeenCalledTimes(2)); + emitLine(second, emptyResult); + second.emit("close", 0); + + await expect(promise).rejects.toThrow( + "claude returned no structured_output", + ); + expect(mockSpawn).toHaveBeenCalledTimes(2); + }); + + it("does not continue when --no-session-persistence makes the session unresumable", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + const noPersistenceAgent = new ClaudeAgent({ + extraArgs: ["--no-session-persistence"], + }); + + const promise = noPersistenceAgent.run("prompt", "/cwd"); + + emitLine(proc, { + type: "result", + subtype: "success", + is_error: false, + session_id: "session-abc", + usage: { + input_tokens: 0, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + output_tokens: 0, + }, + structured_output: null, + }); + proc.emit("close", 0); + + await expect(promise).rejects.toThrow(/--no-session-persistence/); + expect(mockSpawn).toHaveBeenCalledTimes(1); + expect(mockAppendDebugLog).not.toHaveBeenCalledWith( + "claude:output:continuation", + expect.anything(), + ); + }); + + it("does not continue when claude reported no session id", async () => { const proc = createMockProcess(); mockSpawn.mockReturnValue(proc); @@ -1189,12 +1396,68 @@ describe("ClaudeAgent", () => { }, structured_output: null, }); + proc.emit("close", 0); + + await expect(promise).rejects.toThrow(/no session id/); + expect(mockSpawn).toHaveBeenCalledTimes(1); + }); + + it.each(["-c", "--continue"])( + "does not recover configured %s without a captured session id", + async (continuationArg) => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + const continuingAgent = new ClaudeAgent({ + extraArgs: [continuationArg], + }); + + const promise = continuingAgent.run("prompt", "/cwd"); + + emitLine(proc, { + type: "result", + subtype: "success", + is_error: false, + usage: { + input_tokens: 0, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + output_tokens: 0, + }, + structured_output: null, + }); + proc.emit("close", 0); + + await expect(promise).rejects.toThrow(/no session id/); + expect(mockSpawn).toHaveBeenCalledTimes(1); + expect(mockAppendDebugLog).not.toHaveBeenCalledWith( + "claude:output:continuation", + expect.anything(), + ); + }, + ); + + it("does not continue when claude reported an error result", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + const promise = agent.run("prompt", "/cwd"); + + emitLine(proc, { + type: "result", + subtype: "error_during_execution", + is_error: true, + usage: { + input_tokens: 0, + cache_read_input_tokens: 0, + cache_creation_input_tokens: 0, + output_tokens: 0, + }, + structured_output: null, + }); proc.emit("close", 0); - await expect(promise).rejects.toThrow( - "claude returned no structured_output", - ); + await expect(promise).rejects.toThrow("claude reported error"); + expect(mockSpawn).toHaveBeenCalledTimes(1); }); it("picks up structured_output from a later result event when the first had none", async () => { diff --git a/src/core/agents/claude.ts b/src/core/agents/claude.ts index 1d3efe86..92bdea80 100644 --- a/src/core/agents/claude.ts +++ b/src/core/agents/claude.ts @@ -1,5 +1,4 @@ import { execFileSync, spawn } from "node:child_process"; -import { createWriteStream } from "node:fs"; import { buildAgentOutputSchema, type Agent, @@ -7,11 +6,22 @@ import { type AgentOutputSchema, type AgentResult, type AgentRunOptions, + type OnMessage, + type OnUsage, type TokenUsage, PermanentAgentError, } from "./types.js"; +import { appendDebugLog } from "../debug-log.js"; +import { + EmptyAgentResponseError, + runTurnWithEmptyResponseRetry, +} from "./empty-response.js"; import { shutdownChildProcess } from "./managed-process.js"; -import { parseJSONLStream, setupAbortHandler } from "./stream-utils.js"; +import { + AgentLogFile, + parseJSONLStream, + setupAbortHandler, +} from "./stream-utils.js"; const DEFAULT_FINAL_RESULT_EXIT_GRACE_MS = 15_000; /** Upper bound on the stdout tail kept for non-zero-exit error reporting. */ @@ -50,7 +60,10 @@ interface ClaudeResultEvent { structured_output: AgentOutput | null; } -type ClaudeEvent = ClaudeAssistantEvent | ClaudeResultEvent | { type: string }; +type ClaudeEvent = + | ClaudeAssistantEvent + | ClaudeResultEvent + | { type: string; session_id?: string }; interface ClaudeAgentDeps { bin?: string; @@ -138,13 +151,27 @@ function isFinalStructuredResult(event: ClaudeResultEvent): boolean { ); } +function isSessionContinuationArg(arg: string): boolean { + return arg === "-c" || arg === "--continue"; +} + +// `--no-session-persistence` tells claude not to write the session to disk, so +// there is nothing for `--resume` to reopen. +function sessionPersistenceDisabled(userArgs: string[]): boolean { + return userArgs.some((arg) => arg === "--no-session-persistence"); +} + function buildClaudeArgs( prompt: string, schema: AgentOutputSchema, extraArgs?: string[], + resumeSessionId?: string | null, ): string[] { const userArgs = extraArgs ?? []; - const userSpecifiedPermissionMode = userArgs.some( + const turnArgs = resumeSessionId + ? userArgs.filter((arg) => !isSessionContinuationArg(arg)) + : userArgs; + const userSpecifiedPermissionMode = turnArgs.some( (arg) => arg === "--dangerously-skip-permissions" || arg === "--permission-mode" || @@ -154,7 +181,7 @@ function buildClaudeArgs( ); return [ - ...userArgs, + ...turnArgs, "-p", prompt, "--verbose", @@ -162,6 +189,7 @@ function buildClaudeArgs( "stream-json", "--json-schema", JSON.stringify(schema), + ...(resumeSessionId ? ["--resume", resumeSessionId] : []), ...(userSpecifiedPermissionMode ? [] : ["--dangerously-skip-permissions"]), ]; } @@ -321,19 +349,61 @@ export class ClaudeAgent implements Agent { deps.schema ?? buildAgentOutputSchema({ includeStopField: false }); } - run( + async run( prompt: string, cwd: string, options?: AgentRunOptions, ): Promise { const { onUsage, onMessage, signal, logPath } = options ?? {}; + const logFile = new AgentLogFile(logPath); + // Populated from the first turn's stream so a continuation resumes that + // exact conversation and still sees its own reasoning and tool calls. + let sessionId: string | null = null; - return new Promise((resolve, reject) => { - const logStream = logPath ? createWriteStream(logPath) : null; + try { + // `--json-schema` is a spawn flag, so the continuation turn keeps the + // output contract without any extra prompt scaffolding. + return await runTurnWithEmptyResponseRetry({ + logEvent: "claude:output:continuation", + onUsage, + signal, + initialText: prompt, + runTurn: (text, onTurnUsage) => + this.runTurn(text, cwd, { + onUsage: onTurnUsage, + onMessage, + signal, + logFile, + resumeSessionId: sessionId, + onSessionId: (id) => { + sessionId = id; + }, + }), + }); + } finally { + logFile.finish(); + } + } + private runTurn( + prompt: string, + cwd: string, + options: { + onUsage?: OnUsage; + onMessage?: OnMessage; + signal?: AbortSignal; + logFile: AgentLogFile; + resumeSessionId: string | null; + onSessionId: (sessionId: string) => void; + }, + ): Promise { + const { onUsage, onMessage, signal, logFile } = options; + const { resumeSessionId, onSessionId } = options; + + return new Promise((resolve, reject) => { const child = spawn( this.bin, - buildClaudeArgs(prompt, this.schema, this.extraArgs), + buildClaudeArgs(prompt, this.schema, this.extraArgs, resumeSessionId), { cwd, detached: this.platform !== "win32", @@ -342,6 +412,7 @@ export class ClaudeAgent implements Agent { env: process.env, }, ); + logFile.track(child); if ( setupAbortHandler(signal, child, reject, () => @@ -352,6 +423,7 @@ export class ClaudeAgent implements Agent { } let resultEvent: ClaudeResultEvent | null = null; + let turnSessionId: string | null = resumeSessionId; let finalStructuredResultEvent: ClaudeResultEvent | null = null; let latestResultUsage: ClaudeResultEvent["usage"] | null = null; let finalResultCleanupTimer: ReturnType | null = null; @@ -382,7 +454,13 @@ export class ClaudeAgent implements Agent { reject(new Error(`Failed to spawn claude: ${err.message}`)); }); - parseJSONLStream(child.stdout!, logStream, (event) => { + parseJSONLStream(child.stdout!, logFile, (event) => { + const eventSessionId = (event as { session_id?: unknown }).session_id; + if (typeof eventSessionId === "string" && eventSessionId) { + turnSessionId = eventSessionId; + onSessionId(eventSessionId); + } + if (event.type === "assistant") { const msg = (event as ClaudeAssistantEvent).message; const nextUsage = toTokenUsage(msg.usage); @@ -499,7 +577,6 @@ export class ClaudeAgent implements Agent { if (finalResultCleanupTimer) { clearTimeout(finalResultCleanupTimer); } - logStream?.end(); if (code !== 0 && !closedAfterFinalCleanup) { const failure = describeExitFailure(code, stdoutTail, stderr); reject( @@ -533,7 +610,33 @@ export class ClaudeAgent implements Agent { } if (!terminalResultEvent.structured_output) { - reject(new Error("claude returned no structured_output")); + const userArgs = this.extraArgs ?? []; + const resumeBlockedReason = sessionPersistenceDisabled(userArgs) + ? "--no-session-persistence disables session resume, so the turn cannot be continued" + : !turnSessionId + ? "claude reported no session id, so the turn cannot be resumed" + : null; + appendDebugLog("claude:output:missing", { + subtype: terminalResultEvent.subtype, + resumed: resumeSessionId !== null, + hasSessionId: turnSessionId !== null, + resumeBlockedReason, + }); + reject( + new EmptyAgentResponseError( + resumeBlockedReason + ? `claude returned no structured_output (${resumeBlockedReason})` + : "claude returned no structured_output", + { + // A non-error `result` event with subtype "success" is claude's + // own end-of-turn signal; it just carried no final answer. + turnCompleted: resumeBlockedReason === null, + usage: toTokenUsage( + latestResultUsage ?? terminalResultEvent.usage, + ), + }, + ), + ); return; } diff --git a/src/core/agents/codex.test.ts b/src/core/agents/codex.test.ts index 86369a8d..021daf0e 100644 --- a/src/core/agents/codex.test.ts +++ b/src/core/agents/codex.test.ts @@ -6,10 +6,18 @@ vi.mock("node:child_process", () => ({ spawn: vi.fn(), })); +vi.mock("../debug-log.js", () => ({ + appendDebugLog: vi.fn(), + initDebugLog: vi.fn(), + serializeError: vi.fn(), +})); + import { execFileSync, spawn } from "node:child_process"; +import { appendDebugLog } from "../debug-log.js"; import { CodexAgent } from "./codex.js"; const mockSpawn = vi.mocked(spawn); +const mockAppendDebugLog = vi.mocked(appendDebugLog); function createMockProcess() { const proc = Object.assign(new EventEmitter(), { @@ -21,6 +29,39 @@ function createMockProcess() { return proc as typeof proc & ReturnType; } +function emitJson(proc: ReturnType, event: unknown) { + proc.stdout.emit("data", Buffer.from(`${JSON.stringify(event)}\n`)); +} + +function threadStarted(threadId: string) { + return { type: "thread.started", thread_id: threadId }; +} + +function agentMessage(text: string) { + return { + type: "item.completed", + item: { type: "agent_message", text }, + }; +} + +function turnCompleted(inputTokens: number, outputTokens: number) { + return { + type: "turn.completed", + usage: { + input_tokens: inputTokens, + cached_input_tokens: 0, + output_tokens: outputTokens, + }, + }; +} + +const FINAL_OUTPUT = JSON.stringify({ + success: true, + summary: "recovered", + key_changes_made: [], + key_learnings: [], +}); + describe("CodexAgent", () => { beforeEach(() => { vi.clearAllMocks(); @@ -203,4 +244,233 @@ describe("CodexAgent", () => { ); expect(proc.kill).not.toHaveBeenCalled(); }); + + it("resumes the recorded thread once with the bare nudge when a completed turn had no agent message", async () => { + const first = createMockProcess(); + const second = createMockProcess(); + mockSpawn.mockReturnValueOnce(first).mockReturnValueOnce(second); + const agent = new CodexAgent("/tmp/schema.json"); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(first, threadStarted("thread-abc")); + emitJson(first, turnCompleted(10, 5)); + first.emit("close", 0); + + await vi.waitFor(() => expect(mockSpawn).toHaveBeenCalledTimes(2)); + emitJson(second, agentMessage(FINAL_OUTPUT)); + emitJson(second, turnCompleted(3, 2)); + second.emit("close", 0); + + await expect(promise).resolves.toMatchObject({ + output: { success: true, summary: "recovered" }, + usage: { inputTokens: 13, outputTokens: 7 }, + }); + + expect(mockSpawn.mock.calls[1]![1]).toEqual([ + "exec", + "resume", + "thread-abc", + "You did not produce a final answer. Continue and provide your final summary now.", + "--json", + "--output-schema", + "/tmp/schema.json", + "--dangerously-bypass-approvals-and-sandbox", + ]); + expect(mockAppendDebugLog).toHaveBeenCalledWith( + "codex:output:continuation", + expect.objectContaining({ attempt: 1 }), + ); + }); + + it("recovers a completed turn whose agent message is only whitespace", async () => { + const first = createMockProcess(); + const second = createMockProcess(); + mockSpawn.mockReturnValueOnce(first).mockReturnValueOnce(second); + const agent = new CodexAgent("/tmp/schema.json"); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(first, threadStarted("thread-abc")); + emitJson(first, agentMessage(" \n\t")); + emitJson(first, turnCompleted(10, 5)); + first.emit("close", 0); + + await vi.waitFor(() => expect(mockSpawn).toHaveBeenCalledTimes(2)); + emitJson(second, agentMessage(FINAL_OUTPUT)); + emitJson(second, turnCompleted(3, 2)); + second.emit("close", 0); + + await expect(promise).resolves.toMatchObject({ + output: { success: true, summary: "recovered" }, + }); + expect(mockAppendDebugLog).toHaveBeenCalledWith( + "codex:output:continuation", + expect.objectContaining({ attempt: 1 }), + ); + }); + + it("does not re-ask when the turn never completed", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + const agent = new CodexAgent("/tmp/schema.json"); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(proc, threadStarted("thread-abc")); + proc.emit("close", 0); + + await expect(promise).rejects.toThrow("codex returned no agent message"); + expect(mockSpawn).toHaveBeenCalledTimes(1); + expect(mockAppendDebugLog).not.toHaveBeenCalledWith( + "codex:output:continuation", + expect.anything(), + ); + }); + + it("does not re-ask when codex reported no thread id to resume", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + const agent = new CodexAgent("/tmp/schema.json"); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(proc, turnCompleted(10, 5)); + proc.emit("close", 0); + + await expect(promise).rejects.toThrow(/no thread id/); + expect(mockSpawn).toHaveBeenCalledTimes(1); + }); + + // Each of these is accepted by `codex exec` but rejected by + // `codex exec resume`, so forwarding it would replace the accurate + // empty-response diagnostic with a codex CLI usage error. + it.each([ + ["--add-dir", "/shared"], + ["-C", "/shared"], + ["--cd", "/shared"], + ["--sandbox", "workspace-write"], + ["--full-auto"], + ["--oss"], + ["--profile", "work"], + ])( + "does not re-ask when configured codex args include %s", + async (...extraArgs) => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + const agent = new CodexAgent("/tmp/schema.json", { extraArgs }); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(proc, threadStarted("thread-abc")); + emitJson(proc, turnCompleted(10, 5)); + proc.emit("close", 0); + + await expect(promise).rejects.toThrow( + new RegExp(`${extraArgs[0]!.replace("-", "\\-")}.*codex exec resume`), + ); + expect(mockSpawn).toHaveBeenCalledTimes(1); + }, + ); + + it("does not re-ask when --ephemeral leaves no rollout to resume", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + const agent = new CodexAgent("/tmp/schema.json", { + extraArgs: ["--ephemeral"], + }); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(proc, threadStarted("thread-abc")); + emitJson(proc, turnCompleted(10, 5)); + proc.emit("close", 0); + + await expect(promise).rejects.toThrow(/--ephemeral records no rollout/); + expect(mockSpawn).toHaveBeenCalledTimes(1); + expect(mockAppendDebugLog).not.toHaveBeenCalledWith( + "codex:output:continuation", + expect.anything(), + ); + }); + + it("keeps the empty-response diagnostic when the continuation spawn itself fails", async () => { + const first = createMockProcess(); + const second = createMockProcess(); + mockSpawn.mockReturnValueOnce(first).mockReturnValueOnce(second); + const agent = new CodexAgent("/tmp/schema.json"); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(first, threadStarted("thread-abc")); + emitJson(first, turnCompleted(10, 5)); + first.emit("close", 0); + + await vi.waitFor(() => expect(mockSpawn).toHaveBeenCalledTimes(2)); + second.stderr.emit( + "data", + Buffer.from("error: unexpected argument '--add-dir' found"), + ); + second.emit("close", 2); + + const error = await promise.then( + () => null, + (err: Error) => err, + ); + expect(error?.message).toBe("codex returned no agent message"); + expect((error?.cause as Error).message).toContain( + "codex exited with code 2", + ); + expect(mockAppendDebugLog).toHaveBeenCalledWith( + "codex:output:continuation", + expect.objectContaining({ continuationFailed: true }), + ); + }); + + it("forwards resume-compatible user args to the continuation", async () => { + const first = createMockProcess(); + const second = createMockProcess(); + mockSpawn.mockReturnValueOnce(first).mockReturnValueOnce(second); + const agent = new CodexAgent("/tmp/schema.json", { + extraArgs: ["--model", "gpt-5.5"], + }); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(first, threadStarted("thread-abc")); + emitJson(first, turnCompleted(1, 1)); + first.emit("close", 0); + + await vi.waitFor(() => expect(mockSpawn).toHaveBeenCalledTimes(2)); + emitJson(second, agentMessage(FINAL_OUTPUT)); + emitJson(second, turnCompleted(1, 1)); + second.emit("close", 0); + + await expect(promise).resolves.toMatchObject({ + output: { summary: "recovered" }, + }); + expect(mockSpawn.mock.calls[1]![1]).toEqual([ + "exec", + "resume", + "--model", + "gpt-5.5", + "thread-abc", + "You did not produce a final answer. Continue and provide your final summary now.", + "--json", + "--output-schema", + "/tmp/schema.json", + "--dangerously-bypass-approvals-and-sandbox", + ]); + }); + + it("fails after exactly one re-ask when the continuation is also empty", async () => { + const first = createMockProcess(); + const second = createMockProcess(); + mockSpawn.mockReturnValueOnce(first).mockReturnValueOnce(second); + const agent = new CodexAgent("/tmp/schema.json"); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(first, threadStarted("thread-abc")); + emitJson(first, turnCompleted(10, 5)); + first.emit("close", 0); + + await vi.waitFor(() => expect(mockSpawn).toHaveBeenCalledTimes(2)); + emitJson(second, turnCompleted(1, 1)); + second.emit("close", 0); + + await expect(promise).rejects.toThrow("codex returned no agent message"); + expect(mockSpawn).toHaveBeenCalledTimes(2); + }); }); diff --git a/src/core/agents/codex.ts b/src/core/agents/codex.ts index f1b25a19..06715268 100644 --- a/src/core/agents/codex.ts +++ b/src/core/agents/codex.ts @@ -1,13 +1,20 @@ import { execFileSync, spawn } from "node:child_process"; -import { createWriteStream } from "node:fs"; import type { Agent, AgentResult, AgentOutput, + OnMessage, + OnUsage, TokenUsage, AgentRunOptions, } from "./types.js"; +import { appendDebugLog } from "../debug-log.js"; import { + EmptyAgentResponseError, + runTurnWithEmptyResponseRetry, +} from "./empty-response.js"; +import { + AgentLogFile, parseJSONLStream, setupAbortHandler, setupChildProcessHandlers, @@ -27,7 +34,65 @@ interface CodexTurnCompleted { }; } -type CodexEvent = CodexItemCompleted | CodexTurnCompleted | { type: string }; +interface CodexThreadStarted { + type: "thread.started"; + thread_id?: string; + threadId?: string; + id?: string; +} + +type CodexEvent = + | CodexItemCompleted + | CodexTurnCompleted + | CodexThreadStarted + | { type: string }; + +function threadIdOf(event: CodexThreadStarted): string | null { + const candidate = event.thread_id ?? event.threadId ?? event.id; + return typeof candidate === "string" && candidate ? candidate : null; +} + +// `codex exec resume` is a narrower subcommand than `codex exec` and clap +// rejects anything it does not declare. Rather than silently downgrading a +// user's sandbox choice or shelling out a command codex will refuse - which +// would replace the accurate empty-response diagnostic with a CLI usage error +// - gnhf skips the empty-response continuation for these configurations. +// Verified against codex-cli 0.147.0 by diffing `codex exec --help` against +// `codex exec resume --help`; the approval flags are kept because older codex +// releases still accept them on `codex exec` and resume rejects them too. +const CODEX_RESUME_UNSUPPORTED_ARGS = [ + "--add-dir", + "-C", + "--cd", + "-s", + "--sandbox", + "--approve-for-me", + "--oss", + "--local-provider", + "-p", + "--profile", + "--full-auto", + "-a", + "--ask-for-approval", +]; + +// `--ephemeral` is accepted by both `codex exec` and `codex exec resume`, so +// the denylist above cannot catch it: the block is semantic. An ephemeral run +// records no rollout, so resuming its thread id fails with "no rollout found". +function codexRecordsNoRollout(extraArgs?: string[]): boolean { + return (extraArgs ?? []).includes("--ephemeral"); +} + +function codexResumeUnsupportedArg(extraArgs?: string[]): string | null { + return ( + (extraArgs ?? []).find((arg) => + CODEX_RESUME_UNSUPPORTED_ARGS.some( + (unsupported) => + arg === unsupported || arg.startsWith(`${unsupported}=`), + ), + ) ?? null + ); +} interface CodexAgentDeps { bin?: string; @@ -84,13 +149,8 @@ function terminateCodexProcess( child.kill("SIGTERM"); } -function buildCodexArgs( - prompt: string, - schemaPath: string, - extraArgs?: string[], -): string[] { - const userArgs = extraArgs ?? []; - const userSpecifiedExecutionMode = userArgs.some( +function userSpecifiedExecutionMode(userArgs: string[]): boolean { + return userArgs.some( (arg) => arg === "--full-auto" || arg === "--dangerously-bypass-approvals-and-sandbox" || @@ -101,6 +161,14 @@ function buildCodexArgs( arg.startsWith("--ask-for-approval=") || arg === "-a", ); +} + +function buildCodexArgs( + prompt: string, + schemaPath: string, + extraArgs?: string[], +): string[] { + const userArgs = extraArgs ?? []; return [ "exec", @@ -109,7 +177,7 @@ function buildCodexArgs( "--json", "--output-schema", schemaPath, - ...(userSpecifiedExecutionMode + ...(userSpecifiedExecutionMode(userArgs) ? [] : ["--dangerously-bypass-approvals-and-sandbox"]), "--color", @@ -117,6 +185,32 @@ function buildCodexArgs( ]; } +// `codex exec resume ` replays the recorded session, so the +// continuation turn keeps the first turn's reasoning and tool calls. It does +// not accept `--color`, so that flag is dropped here rather than forwarded. +function buildCodexResumeArgs( + prompt: string, + schemaPath: string, + threadId: string, + extraArgs?: string[], +): string[] { + const userArgs = extraArgs ?? []; + + return [ + "exec", + "resume", + ...userArgs, + threadId, + prompt, + "--json", + "--output-schema", + schemaPath, + ...(userSpecifiedExecutionMode(userArgs) + ? [] + : ["--dangerously-bypass-approvals-and-sandbox"]), + ]; +} + export class CodexAgent implements Agent { name = "codex"; @@ -133,19 +227,66 @@ export class CodexAgent implements Agent { this.schemaPath = schemaPath; } - run( + async run( prompt: string, cwd: string, options?: AgentRunOptions, ): Promise { const { onUsage, onMessage, signal, logPath } = options ?? {}; + const logFile = new AgentLogFile(logPath); + let threadId: string | null = null; - return new Promise((resolve, reject) => { - const logStream = logPath ? createWriteStream(logPath) : null; + try { + // `--output-schema` is a spawn flag, so the continuation turn carries the + // same output contract without any extra prompt scaffolding. + return await runTurnWithEmptyResponseRetry({ + logEvent: "codex:output:continuation", + onUsage, + signal, + initialText: prompt, + runTurn: (text, onTurnUsage) => + this.runTurn(text, cwd, { + onUsage: onTurnUsage, + onMessage, + signal, + logFile, + resumeThreadId: threadId, + onThreadId: (id) => { + threadId = id; + }, + }), + }); + } finally { + logFile.finish(); + } + } + + private runTurn( + prompt: string, + cwd: string, + options: { + onUsage?: OnUsage; + onMessage?: OnMessage; + signal?: AbortSignal; + logFile: AgentLogFile; + resumeThreadId: string | null; + onThreadId: (threadId: string) => void; + }, + ): Promise { + const { onUsage, onMessage, signal, logFile } = options; + const { resumeThreadId, onThreadId } = options; + return new Promise((resolve, reject) => { const child = spawn( this.bin, - buildCodexArgs(prompt, this.schemaPath, this.extraArgs), + resumeThreadId + ? buildCodexResumeArgs( + prompt, + this.schemaPath, + resumeThreadId, + this.extraArgs, + ) + : buildCodexArgs(prompt, this.schemaPath, this.extraArgs), { cwd, shell: shouldUseWindowsShell(this.bin, this.platform), @@ -153,6 +294,7 @@ export class CodexAgent implements Agent { env: process.env, }, ); + logFile.track(child); if ( setupAbortHandler(signal, child, reject, () => @@ -163,6 +305,11 @@ export class CodexAgent implements Agent { } let lastAgentMessage: string | null = null; + // `turn.completed` is codex's own end-of-turn signal, so it - not a + // clean process exit - is what separates a finished-but-silent turn + // from a turn that never got to answer. + let sawTurnCompleted = false; + let turnThreadId: string | null = resumeThreadId; const cumulative: TokenUsage = { inputTokens: 0, outputTokens: 0, @@ -170,7 +317,15 @@ export class CodexAgent implements Agent { cacheCreationTokens: 0, }; - parseJSONLStream(child.stdout!, logStream, (event) => { + parseJSONLStream(child.stdout!, logFile, (event) => { + if (event.type === "thread.started") { + const id = threadIdOf(event as CodexThreadStarted); + if (id) { + turnThreadId = id; + onThreadId(id); + } + } + if ( event.type === "item.completed" && "item" in event && @@ -180,23 +335,50 @@ export class CodexAgent implements Agent { onMessage?.(lastAgentMessage); } - if (event.type === "turn.completed" && "usage" in event) { - const u = (event as CodexTurnCompleted).usage; - cumulative.inputTokens += u.input_tokens ?? 0; - cumulative.outputTokens += u.output_tokens ?? 0; - cumulative.cacheReadTokens += u.cached_input_tokens ?? 0; - onUsage?.({ ...cumulative }); + if (event.type === "turn.completed") { + sawTurnCompleted = true; + if ("usage" in event) { + const u = (event as CodexTurnCompleted).usage; + cumulative.inputTokens += u.input_tokens ?? 0; + cumulative.outputTokens += u.output_tokens ?? 0; + cumulative.cacheReadTokens += u.cached_input_tokens ?? 0; + onUsage?.({ ...cumulative }); + } } }); - setupChildProcessHandlers(child, "codex", logStream, reject, () => { - if (!lastAgentMessage) { - reject(new Error("codex returned no agent message")); + setupChildProcessHandlers(child, "codex", reject, () => { + const finalAgentMessage = lastAgentMessage?.trim(); + if (!finalAgentMessage) { + const unsupportedArg = codexResumeUnsupportedArg(this.extraArgs); + const resumeBlockedReason = !turnThreadId + ? "codex reported no thread id, so the turn cannot be resumed" + : codexRecordsNoRollout(this.extraArgs) + ? "--ephemeral records no rollout, so the thread cannot be resumed" + : unsupportedArg + ? `configured codex arg "${unsupportedArg}" is not supported by \`codex exec resume\`, so the turn cannot be resumed` + : null; + appendDebugLog("codex:output:missing", { + sawTurnCompleted, + hasThreadId: turnThreadId !== null, + resumeBlockedReason, + }); + reject( + new EmptyAgentResponseError( + resumeBlockedReason + ? `codex returned no agent message (${resumeBlockedReason})` + : "codex returned no agent message", + { + turnCompleted: sawTurnCompleted && resumeBlockedReason === null, + usage: cumulative, + }, + ), + ); return; } try { - const output = JSON.parse(lastAgentMessage) as AgentOutput; + const output = JSON.parse(finalAgentMessage) as AgentOutput; resolve({ output, usage: cumulative }); } catch (err) { reject( diff --git a/src/core/agents/copilot.test.ts b/src/core/agents/copilot.test.ts index e5e326c0..487da894 100644 --- a/src/core/agents/copilot.test.ts +++ b/src/core/agents/copilot.test.ts @@ -6,11 +6,19 @@ vi.mock("node:child_process", () => ({ spawn: vi.fn(), })); +vi.mock("../debug-log.js", () => ({ + appendDebugLog: vi.fn(), + initDebugLog: vi.fn(), + serializeError: vi.fn(), +})); + import { execFileSync, spawn } from "node:child_process"; +import { appendDebugLog } from "../debug-log.js"; import { CopilotAgent } from "./copilot.js"; import { buildAgentOutputSchema } from "./types.js"; const mockSpawn = vi.mocked(spawn); +const mockAppendDebugLog = vi.mocked(appendDebugLog); function createMockProcess() { const proc = Object.assign(new EventEmitter(), { @@ -264,15 +272,40 @@ describe("CopilotAgent", () => { expect(args[1]).toContain("should_fully_stop"); }); - it("rejects when copilot returns no assistant message", async () => { + it("fails an empty response without retrying and names the missing resume contract", async () => { const proc = createMockProcess(); mockSpawn.mockReturnValue(proc); const agent = new CopilotAgent(); const promise = agent.run("test prompt", "/work/dir"); + emitJson(proc, { + type: "session.started", + data: { session_id: "session-abc" }, + }); + emitJson(proc, { type: "assistant.message", data: { outputTokens: 5 } }); proc.emit("close", 0); - await expect(promise).rejects.toThrow("copilot returned no agent message"); + await expect(promise).rejects.toThrow( + /copilot returned no agent message.*resume contract/, + ); + expect(mockSpawn).toHaveBeenCalledTimes(1); + expect(mockAppendDebugLog).not.toHaveBeenCalledWith( + "copilot:output:continuation", + expect.anything(), + ); + }); + + it("does not re-ask when copilot exits non-zero", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + const agent = new CopilotAgent(); + + const promise = agent.run("test prompt", "/work/dir"); + proc.stderr.emit("data", Buffer.from("boom")); + proc.emit("close", 1); + + await expect(promise).rejects.toThrow("copilot exited with code 1"); + expect(mockSpawn).toHaveBeenCalledTimes(1); }); it("rejects when the final assistant message is not valid JSON", async () => { diff --git a/src/core/agents/copilot.ts b/src/core/agents/copilot.ts index e7ae1506..597a9218 100644 --- a/src/core/agents/copilot.ts +++ b/src/core/agents/copilot.ts @@ -1,5 +1,4 @@ import { execFileSync, spawn } from "node:child_process"; -import { createWriteStream } from "node:fs"; import { buildAgentOutputSchema, parseAgentOutput, @@ -7,9 +6,13 @@ import { type AgentOutputSchema, type AgentResult, type AgentRunOptions, + type OnMessage, + type OnUsage, type TokenUsage, } from "./types.js"; +import { appendDebugLog } from "../debug-log.js"; import { + AgentLogFile, parseJSONLStream, setupAbortHandler, setupChildProcessHandlers, @@ -31,6 +34,16 @@ type CopilotEvent = | CopilotAssistantMessageEvent | (CopilotUsageEvent & { type: string }); +// Copilot has no verified exact-session resume contract: `--continue` selects +// by recency rather than identity, and nothing in its JSONL output is +// documented to carry a session id that `--resume` would accept. Continuing +// the wrong session would let the agent describe an earlier iteration's work +// as if it were this turn's, so copilot is excluded from empty-response +// recovery and reports why instead. See +// https://github.com/kunchenguid/gnhf/issues/193. +const COPILOT_EMPTY_RESPONSE_MESSAGE = + "copilot returned no agent message (copilot has no verified exact-session resume contract, so the empty response cannot be safely recovered)"; + interface CopilotAgentDeps { bin?: string; extraArgs?: string[]; @@ -207,16 +220,39 @@ export class CopilotAgent implements Agent { deps.schema ?? buildAgentOutputSchema({ includeStopField: false }); } - run( + async run( prompt: string, cwd: string, options?: AgentRunOptions, ): Promise { const { onUsage, onMessage, signal, logPath } = options ?? {}; + const logFile = new AgentLogFile(logPath); - return new Promise((resolve, reject) => { - const logStream = logPath ? createWriteStream(logPath) : null; + try { + return await this.runTurn(prompt, cwd, { + onUsage, + onMessage, + signal, + logFile, + }); + } finally { + logFile.finish(); + } + } + private runTurn( + prompt: string, + cwd: string, + options: { + onUsage?: OnUsage; + onMessage?: OnMessage; + signal?: AbortSignal; + logFile: AgentLogFile; + }, + ): Promise { + const { onUsage, onMessage, signal, logFile } = options; + + return new Promise((resolve, reject) => { const child = spawn( this.bin, buildCopilotArgs(prompt, this.schema, this.extraArgs), @@ -227,6 +263,7 @@ export class CopilotAgent implements Agent { env: process.env, }, ); + logFile.track(child); if ( setupAbortHandler(signal, child, reject, () => @@ -244,7 +281,7 @@ export class CopilotAgent implements Agent { cacheCreationTokens: 0, }; - parseJSONLStream(child.stdout!, logStream, (event) => { + parseJSONLStream(child.stdout!, logFile, (event) => { if (event.type === "assistant.message") { const data = (event as CopilotAssistantMessageEvent).data; if (typeof data.content === "string") { @@ -272,9 +309,13 @@ export class CopilotAgent implements Agent { } }); - setupChildProcessHandlers(child, "copilot", logStream, reject, () => { + setupChildProcessHandlers(child, "copilot", reject, () => { if (!lastAgentMessage) { - reject(new Error("copilot returned no agent message")); + appendDebugLog("copilot:output:missing", { + recoverable: false, + reason: "copilot has no verified exact-session resume contract", + }); + reject(new Error(COPILOT_EMPTY_RESPONSE_MESSAGE)); return; } diff --git a/src/core/agents/empty-response.test.ts b/src/core/agents/empty-response.test.ts new file mode 100644 index 00000000..70302ffd --- /dev/null +++ b/src/core/agents/empty-response.test.ts @@ -0,0 +1,166 @@ +import { describe, expect, it, vi } from "vitest"; + +vi.mock("../debug-log.js", () => ({ + appendDebugLog: vi.fn(), + serializeError: vi.fn(), +})); + +import { appendDebugLog } from "../debug-log.js"; +import { + EmptyAgentResponseError, + runTurnWithEmptyResponseRetry, +} from "./empty-response.js"; +import type { TokenUsage } from "./types.js"; + +function usage(inputTokens: number, outputTokens: number): TokenUsage { + return { + inputTokens, + outputTokens, + cacheReadTokens: 0, + cacheCreationTokens: 0, + }; +} + +describe("runTurnWithEmptyResponseRetry", () => { + it("does not start or log a continuation after usage aborts the run", async () => { + const controller = new AbortController(); + const runTurn = vi.fn().mockRejectedValue( + new EmptyAgentResponseError("empty", { + turnCompleted: true, + usage: usage(10, 5), + }), + ); + + await expect( + runTurnWithEmptyResponseRetry({ + logEvent: "agent:continuation", + onUsage: () => controller.abort(), + signal: controller.signal, + initialText: "prompt", + runTurn, + }), + ).rejects.toThrow("Agent was aborted"); + + expect(runTurn).toHaveBeenCalledTimes(1); + expect(appendDebugLog).not.toHaveBeenCalled(); + }); + + it("preserves cumulative usage when the continuation is also empty", async () => { + const firstUsage = usage(10, 5); + const continuationUsage = usage(3, 2); + const onUsage = vi.fn(); + const runTurn = vi + .fn() + .mockRejectedValueOnce( + new EmptyAgentResponseError("first empty", { + turnCompleted: true, + usage: firstUsage, + }), + ) + .mockRejectedValueOnce( + new EmptyAgentResponseError("continuation empty", { + turnCompleted: true, + usage: continuationUsage, + }), + ); + + const error = await runTurnWithEmptyResponseRetry({ + logEvent: "agent:continuation", + onUsage, + initialText: "prompt", + runTurn, + }).catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(EmptyAgentResponseError); + expect(error).toMatchObject({ + message: "continuation empty", + usage: usage(13, 7), + }); + expect(onUsage).toHaveBeenLastCalledWith(usage(13, 7)); + expect(runTurn).toHaveBeenCalledTimes(2); + }); + + it("surfaces a token abort after publishing cumulative empty usage", async () => { + const controller = new AbortController(); + const onUsage = vi.fn((reported: TokenUsage) => { + if (reported.inputTokens + reported.outputTokens >= 20) { + controller.abort(); + } + }); + const runTurn = vi + .fn() + .mockRejectedValueOnce( + new EmptyAgentResponseError("first empty", { + turnCompleted: true, + usage: usage(10, 5), + }), + ) + .mockRejectedValueOnce( + new EmptyAgentResponseError("continuation empty", { + turnCompleted: true, + usage: usage(3, 2), + }), + ); + + const error = await runTurnWithEmptyResponseRetry({ + logEvent: "agent:continuation", + onUsage, + signal: controller.signal, + initialText: "prompt", + runTurn, + }).catch((caught: unknown) => caught); + + expect(error).toBeInstanceOf(EmptyAgentResponseError); + expect(error).toMatchObject({ + message: "Agent was aborted", + usage: usage(13, 7), + }); + expect(onUsage).toHaveBeenLastCalledWith(usage(13, 7)); + expect(runTurn).toHaveBeenCalledTimes(2); + }); + + it("rejects a successful continuation after cumulative usage aborts", async () => { + const controller = new AbortController(); + const onUsage = vi.fn((reported: TokenUsage) => { + if (reported.inputTokens + reported.outputTokens >= 20) { + controller.abort(); + } + }); + const continuationUsage = usage(3, 2); + const runTurn = vi + .fn() + .mockRejectedValueOnce( + new EmptyAgentResponseError("first empty", { + turnCompleted: true, + usage: usage(10, 5), + }), + ) + .mockImplementationOnce( + async (_text: string, onTurnUsage: (usage: TokenUsage) => void) => { + onTurnUsage(continuationUsage); + return { + output: { + success: true, + summary: "recovered", + key_changes_made: [], + key_learnings: [], + }, + usage: continuationUsage, + }; + }, + ); + + await expect( + runTurnWithEmptyResponseRetry({ + logEvent: "agent:continuation", + onUsage, + signal: controller.signal, + initialText: "prompt", + runTurn, + }), + ).rejects.toThrow("Agent was aborted"); + + expect(onUsage).toHaveBeenLastCalledWith(usage(13, 7)); + expect(runTurn).toHaveBeenCalledTimes(2); + }); +}); diff --git a/src/core/agents/empty-response.ts b/src/core/agents/empty-response.ts new file mode 100644 index 00000000..67868c46 --- /dev/null +++ b/src/core/agents/empty-response.ts @@ -0,0 +1,168 @@ +import { appendDebugLog, serializeError } from "../debug-log.js"; +import { PermanentAgentError } from "./types.js"; +import type { AgentResult, OnUsage, TokenUsage } from "./types.js"; + +export const EMPTY_RESPONSE_CONTINUATION_PROMPT = + "You did not produce a final answer. Continue and provide your final summary now."; + +/** + * Thrown by an adapter when a turn produced no final message. `turnCompleted` + * separates "the agent finished its turn and simply said nothing" - which one + * continuation nudge can recover - from "the transport died before the turn + * ended", where nudging would post into a session that is still working (or + * gone) and would replace a clear diagnostic with a transport error. + * + * `usage` is what the empty turn actually cost. It is required so the tokens + * burned by a turn that never returned an `AgentResult` stay part of the + * iteration total instead of depending on whichever `onUsage` callback the + * adapter happened to fire last. + */ +export class EmptyAgentResponseError extends Error { + readonly turnCompleted: boolean; + readonly usage: TokenUsage; + + constructor( + message: string, + options: { turnCompleted: boolean; usage: TokenUsage; cause?: unknown }, + ) { + super(message, { cause: options.cause }); + this.name = "EmptyAgentResponseError"; + this.turnCompleted = options.turnCompleted; + this.usage = { ...options.usage }; + } +} + +function isAbortError(error: unknown): boolean { + return ( + error instanceof Error && + (error.name === "AbortError" || error.message === "Agent was aborted") + ); +} + +function createAbortError(): Error { + return new Error("Agent was aborted"); +} + +export function addTokenUsage(left: TokenUsage, right: TokenUsage): TokenUsage { + const total: TokenUsage = { + inputTokens: left.inputTokens + right.inputTokens, + outputTokens: left.outputTokens + right.outputTokens, + cacheReadTokens: left.cacheReadTokens + right.cacheReadTokens, + cacheCreationTokens: left.cacheCreationTokens + right.cacheCreationTokens, + }; + if (left.estimated || right.estimated) { + total.estimated = true; + } + return total; +} + +export interface EmptyResponseRetryOptions { + /** Debug-log event name recorded when the nudge is sent. */ + logEvent: string; + logFields?: Record; + onUsage?: OnUsage; + signal?: AbortSignal; + combineUsage?: (left: TokenUsage, right: TokenUsage) => TokenUsage; + /** + * Text for the first turn. The continuation turn is always the bare nudge - + * adapters that cannot parse a bare turn apply their own existing prompt + * scaffolding inside `runTurn` so both turns are wrapped identically. + */ + initialText: string; + runTurn: (text: string, onTurnUsage: OnUsage) => Promise; +} + +/** + * Runs one turn and, if it completed without a final message, runs exactly one + * bare continuation turn. Usage reported to `onUsage` stays cumulative across + * both turns; any other failure propagates untouched. + * + * If the continuation itself fails for a reason unrelated to the empty response + * - a CLI that refuses the resume command, say - the original empty-response + * diagnostic is what the user gets, with the continuation failure attached as + * `cause` and recorded in the run log. Recovery is best effort, so a broken + * continuation must never overwrite the accurate description of what went + * wrong. Aborts and permanent errors still propagate so `--max-tokens`, Ctrl+C, + * and abort-worthy provider failures behave exactly as before. + */ +export async function runTurnWithEmptyResponseRetry({ + logEvent, + logFields, + onUsage, + signal, + combineUsage = addTokenUsage, + initialText, + runTurn, +}: EmptyResponseRetryOptions): Promise { + try { + return await runTurn(initialText, (usage) => onUsage?.(usage)); + } catch (error) { + if (!(error instanceof EmptyAgentResponseError) || !error.turnCompleted) { + throw error; + } + + const firstTurnUsage = error.usage; + onUsage?.({ ...firstTurnUsage }); + if (signal?.aborted) { + throw createAbortError(); + } + + appendDebugLog(logEvent, { + ...logFields, + attempt: 1, + prompt: EMPTY_RESPONSE_CONTINUATION_PROMPT, + }); + + let retry: AgentResult; + try { + retry = await runTurn(EMPTY_RESPONSE_CONTINUATION_PROMPT, (usage) => { + onUsage?.(combineUsage(firstTurnUsage, usage)); + }); + } catch (continuationError) { + if (continuationError instanceof EmptyAgentResponseError) { + const cumulativeUsage = combineUsage( + firstTurnUsage, + continuationError.usage, + ); + onUsage?.({ ...cumulativeUsage }); + throw new EmptyAgentResponseError( + signal?.aborted ? "Agent was aborted" : continuationError.message, + { + turnCompleted: continuationError.turnCompleted, + usage: cumulativeUsage, + cause: continuationError, + }, + ); + } + + if ( + continuationError instanceof PermanentAgentError || + isAbortError(continuationError) + ) { + throw continuationError; + } + + appendDebugLog(logEvent, { + ...logFields, + attempt: 1, + continuationFailed: true, + error: serializeError(continuationError), + }); + + throw new EmptyAgentResponseError(error.message, { + turnCompleted: error.turnCompleted, + usage: firstTurnUsage, + cause: continuationError, + }); + } + + if (signal?.aborted) { + throw createAbortError(); + } + + return { + output: retry.output, + usage: combineUsage(firstTurnUsage, retry.usage), + }; + } +} diff --git a/src/core/agents/opencode.test.ts b/src/core/agents/opencode.test.ts index 6209ce4b..7c2b6d75 100644 --- a/src/core/agents/opencode.test.ts +++ b/src/core/agents/opencode.test.ts @@ -22,6 +22,7 @@ vi.mock("../debug-log.js", () => ({ })); import { execFileSync, spawn } from "node:child_process"; +import { appendDebugLog } from "../debug-log.js"; import { OpenCodeAgent } from "./opencode.js"; import { buildAgentOutputSchema } from "./types.js"; @@ -34,6 +35,7 @@ const STOP_AGENT_OUTPUT_SCHEMA = buildAgentOutputSchema({ }); const mockSpawn = vi.mocked(spawn); +const mockAppendDebugLog = vi.mocked(appendDebugLog); function createMockProcess() { const proc = Object.assign(new EventEmitter(), { @@ -884,7 +886,82 @@ describe("OpenCodeAgent", () => { }); }); - it("rejects with 'OpenCode produced no final answer' when the stream ends with no structured output and no final_answer text", async () => { + it("continues the same session once when the first turn has no final answer", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + + fetchMock + .mockResolvedValueOnce(jsonResponse({ healthy: true, version: "1.3.13" })) + .mockResolvedValueOnce(jsonResponse({ id: "session-123" })) + .mockResolvedValueOnce( + sseResponse([ + 'data: {"directory":"/repo","payload":{"type":"message.updated","properties":{"sessionID":"session-123","info":{"id":"msg-empty-1","role":"assistant","tokens":{"input":1,"output":1,"cache":{"read":0,"write":0}}}}}}\n\n', + 'data: {"directory":"/repo","payload":{"type":"session.idle","properties":{"sessionID":"session-123"}}}\n\n', + ]), + ) + .mockResolvedValueOnce(promptAsyncResponse()) + .mockResolvedValueOnce( + sseResponse( + finalAnswerEvents("recovered", { + input: 2, + output: 3, + read: 0, + write: 0, + }), + ), + ) + .mockResolvedValueOnce(promptAsyncResponse()) + .mockResolvedValueOnce(jsonResponse(true)); + + await expect(agent.run("test", "/repo")).resolves.toMatchObject({ + output: { success: true, summary: "recovered" }, + usage: { inputTokens: 3, outputTokens: 4 }, + }); + expect(mockAppendDebugLog).toHaveBeenCalledWith( + "opencode:output:continuation", + expect.objectContaining({ sessionId: "session-123", attempt: 1 }), + ); + + const promptRequests = fetchMock.mock.calls.filter( + ([url, init]) => + String(url).includes("/prompt_async") && init?.method === "POST", + ); + expect(promptRequests).toHaveLength(2); + expect(promptRequests[1]?.[1]?.body).toContain( + "You did not produce a final answer", + ); + }); + + it("does not continue when the event stream ends before session.idle", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + + fetchMock + .mockResolvedValueOnce(jsonResponse({ healthy: true, version: "1.3.13" })) + .mockResolvedValueOnce(jsonResponse({ id: "session-123" })) + .mockResolvedValueOnce( + sseResponse([ + 'data: {"directory":"/repo","payload":{"type":"message.updated","properties":{"sessionID":"session-123","info":{"id":"msg-empty-1","role":"assistant","tokens":{"input":1,"output":1,"cache":{"read":0,"write":0}}}}}}\n\n', + ]), + ) + .mockResolvedValueOnce(promptAsyncResponse()) + .mockResolvedValueOnce(jsonResponse(true)); + + await expect(agent.run("test", "/repo")).rejects.toThrow( + "OpenCode produced no final answer", + ); + expect( + fetchMock.mock.calls.filter(([url]) => + String(url).includes("/prompt_async"), + ), + ).toHaveLength(1); + expect(mockAppendDebugLog).not.toHaveBeenCalledWith( + "opencode:output:continuation", + expect.anything(), + ); + }); + + it("rejects with 'OpenCode produced no final answer' after one continuation is also empty", async () => { const proc = createMockProcess(); mockSpawn.mockReturnValue(proc); @@ -905,11 +982,23 @@ describe("OpenCodeAgent", () => { parts: [{ type: "step-start" }], }), ) + .mockResolvedValueOnce( + sseResponse([ + 'data: {"directory":"/repo","payload":{"type":"message.part.updated","properties":{"sessionID":"session-123","part":{"id":"finish-2","type":"step-finish","tokens":{"input":2,"output":1,"cache":{"read":0,"write":0}}}}}}\n\n', + 'data: {"directory":"/repo","payload":{"type":"session.idle","properties":{"sessionID":"session-123"}}}\n\n', + ]), + ) + .mockResolvedValueOnce(promptAsyncResponse()) .mockResolvedValueOnce(jsonResponse(true)); await expect(agent.run("test", "/repo")).rejects.toThrow( "OpenCode produced no final answer", ); + expect( + fetchMock.mock.calls.filter(([url]) => + String(url).includes("/prompt_async"), + ), + ).toHaveLength(2); }); it("does not fall back to reasoning-phase text when no final_answer text was emitted", async () => { @@ -926,6 +1015,12 @@ describe("OpenCodeAgent", () => { ]), ) .mockResolvedValueOnce(promptAsyncResponse()) + .mockResolvedValueOnce( + sseResponse( + 'data: {"directory":"/repo","payload":{"type":"session.idle","properties":{"sessionID":"session-123"}}}\n\n', + ), + ) + .mockResolvedValueOnce(promptAsyncResponse()) .mockResolvedValueOnce(jsonResponse(true)); await expect(agent.run("test", "/repo")).rejects.toThrow( @@ -950,6 +1045,12 @@ describe("OpenCodeAgent", () => { ]), ) .mockResolvedValueOnce(promptAsyncResponse()) + .mockResolvedValueOnce( + sseResponse( + 'data: {"directory":"/repo","payload":{"type":"session.idle","properties":{"sessionID":"session-123"}}}\n\n', + ), + ) + .mockResolvedValueOnce(promptAsyncResponse()) .mockResolvedValueOnce(jsonResponse(true)); await expect(agent.run("test", "/repo")).rejects.toThrow( diff --git a/src/core/agents/opencode.ts b/src/core/agents/opencode.ts index c797a640..20be2cf6 100644 --- a/src/core/agents/opencode.ts +++ b/src/core/agents/opencode.ts @@ -16,6 +16,10 @@ import { type TokenUsage, } from "./types.js"; import { appendDebugLog, serializeError } from "../debug-log.js"; +import { + EmptyAgentResponseError, + runTurnWithEmptyResponseRetry, +} from "./empty-response.js"; import { shutdownChildProcess } from "./managed-process.js"; interface OpenCodeMessagePart { @@ -371,15 +375,24 @@ export class OpenCodeAgent implements Agent { try { const server = await this.ensureServer(cwd, runController.signal); sessionId = await this.createSession(server, cwd, runController.signal); - const result = await this.streamMessage( - server, - sessionId, - buildPrompt(prompt, this.schema), - runController.signal, - logStream, + const activeSessionId = sessionId; + const result = await runTurnWithEmptyResponseRetry({ + logEvent: "opencode:output:continuation", + logFields: { sessionId: activeSessionId }, onUsage, - onMessage, - ); + signal: runController.signal, + initialText: buildPrompt(prompt, this.schema), + runTurn: (text, onTurnUsage) => + this.streamMessage( + server, + activeSessionId, + text, + runController.signal, + logStream, + onTurnUsage, + onMessage, + ), + }); appendDebugLog("opencode:run:end", { sessionId, elapsedMs: Date.now() - runStartedAt, @@ -847,6 +860,13 @@ export class OpenCodeAgent implements Agent { onMessage?.(trimmed); }; + // `sawSessionIdle` means the session really reported idle, so it is the + // only safe gate for an empty-response continuation. `streamTerminated` + // is the broader "stop reading" signal and is also set by provider + // errors, which must never qualify for a nudge. + let sawSessionIdle = false; + let streamTerminated = false; + const handleEvent = (event: OpenCodeStreamEvent) => { const errorInfo = extractStreamError(event, sessionId); if (errorInfo) { @@ -907,13 +927,17 @@ export class OpenCodeAgent implements Agent { return false; } - return payload?.type === "session.idle"; + if (payload?.type === "session.idle") { + sawSessionIdle = true; + return true; + } + + return false; }; const decoder = new TextDecoder(); const reader = eventResponse.body.getReader(); let buffer = ""; - let sawSessionIdle = false; const processRawEvent = (rawEvent: string) => { if (!rawEvent.trim()) return; @@ -928,7 +952,7 @@ export class OpenCodeAgent implements Agent { const event = JSON.parse(dataLines.join("\n")) as OpenCodeStreamEvent; noteEvent(event.payload?.type); if (handleEvent(event)) { - sawSessionIdle = true; + streamTerminated = true; } } catch { // Ignore malformed SSE events. @@ -956,7 +980,7 @@ export class OpenCodeAgent implements Agent { processRawEvent(buffer.slice(0, boundary)); buffer = buffer.slice(boundary + separatorLen); - if (sawSessionIdle) return; + if (streamTerminated) return; } if (flushRemainder && buffer.trim()) { @@ -967,7 +991,7 @@ export class OpenCodeAgent implements Agent { let bytesRead = 0; try { - while (!sawSessionIdle) { + while (!streamTerminated) { let readResult: ReadableStreamReadResult; try { readResult = await reader.read(); @@ -1033,6 +1057,7 @@ export class OpenCodeAgent implements Agent { elapsedMs: Date.now() - streamStartedAt, bytesRead, sawSessionIdle, + streamTerminated, telemetry: buildTelemetry(), }); @@ -1082,8 +1107,12 @@ export class OpenCodeAgent implements Agent { appendDebugLog("opencode:output:missing", { sessionId, hasStructuredOutput: structuredOutputFromSSE !== null, + sawSessionIdle, + }); + throw new EmptyAgentResponseError("OpenCode produced no final answer", { + turnCompleted: sawSessionIdle, + usage: { ...usage }, }); - throw new Error("OpenCode produced no final answer"); } try { diff --git a/src/core/agents/pi.test.ts b/src/core/agents/pi.test.ts index 3ce1a367..15e87560 100644 --- a/src/core/agents/pi.test.ts +++ b/src/core/agents/pi.test.ts @@ -6,11 +6,19 @@ vi.mock("node:child_process", () => ({ spawn: vi.fn(), })); +vi.mock("../debug-log.js", () => ({ + appendDebugLog: vi.fn(), + initDebugLog: vi.fn(), + serializeError: vi.fn(), +})); + import { execFileSync, spawn } from "node:child_process"; +import { appendDebugLog } from "../debug-log.js"; import { PiAgent } from "./pi.js"; import { buildAgentOutputSchema } from "./types.js"; const mockSpawn = vi.mocked(spawn); +const mockAppendDebugLog = vi.mocked(appendDebugLog); function createMockProcess() { const proc = Object.assign(new EventEmitter(), { @@ -359,7 +367,7 @@ describe("PiAgent", () => { await expect(promise).rejects.toThrow("Invalid pi output"); }); - it("rejects empty final text", async () => { + it("fails an empty response without retrying and names --no-session as the reason", async () => { const proc = createMockProcess(); mockSpawn.mockReturnValue(proc); const agent = new PiAgent(); @@ -371,7 +379,35 @@ describe("PiAgent", () => { }); proc.emit("close", 0); - await expect(promise).rejects.toThrow("pi returned no text output"); + await expect(promise).rejects.toThrow( + /pi returned no text output.*--no-session/, + ); + expect(mockSpawn).toHaveBeenCalledTimes(1); + expect(mockAppendDebugLog).not.toHaveBeenCalledWith( + "pi:output:continuation", + expect.anything(), + ); + }); + + it("does not re-ask when pi reports an error stop reason", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + const agent = new PiAgent(); + + const promise = agent.run("test prompt", "/work/dir"); + emitJson(proc, { + type: "message_end", + message: { + role: "assistant", + content: "", + stopReason: "error", + errorMessage: "provider exploded", + }, + }); + proc.emit("close", 0); + + await expect(promise).rejects.toThrow("pi reported error: provider explod"); + expect(mockSpawn).toHaveBeenCalledTimes(1); }); it("rejects malformed JSON", async () => { diff --git a/src/core/agents/pi.ts b/src/core/agents/pi.ts index 6581e713..6a2b27b7 100644 --- a/src/core/agents/pi.ts +++ b/src/core/agents/pi.ts @@ -1,5 +1,4 @@ import { execFileSync, spawn } from "node:child_process"; -import { createWriteStream } from "node:fs"; import { buildAgentOutputSchema, parseAgentOutput, @@ -8,14 +7,26 @@ import { type AgentOutputSchema, type AgentResult, type AgentRunOptions, + type OnMessage, + type OnUsage, type TokenUsage, } from "./types.js"; +import { appendDebugLog } from "../debug-log.js"; import { + AgentLogFile, parseJSONLStream, setupAbortHandler, setupChildProcessHandlers, } from "./stream-utils.js"; +// gnhf runs pi with `--no-session`, so a turn leaves nothing behind to +// continue. A second spawn would be a fresh agent with no knowledge of the +// task, the workspace edits, or the notes - it could only invent a final +// summary - so pi is deliberately excluded from empty-response recovery and +// reports the reason instead. +const PI_EMPTY_RESPONSE_MESSAGE = + "pi returned no text output (gnhf runs pi with --no-session, so there is no session to continue and the empty response cannot be recovered)"; + interface PiAgentDeps { bin?: string; extraArgs?: string[]; @@ -213,15 +224,39 @@ export class PiAgent implements Agent { deps.schema ?? buildAgentOutputSchema({ includeStopField: false }); } - run( + async run( prompt: string, cwd: string, options?: AgentRunOptions, ): Promise { const { onUsage, onMessage, signal, logPath } = options ?? {}; + const logFile = new AgentLogFile(logPath); + + try { + return await this.runTurn(prompt, cwd, { + onUsage, + onMessage, + signal, + logFile, + }); + } finally { + logFile.finish(); + } + } + + private runTurn( + prompt: string, + cwd: string, + options: { + onUsage?: OnUsage; + onMessage?: OnMessage; + signal?: AbortSignal; + logFile: AgentLogFile; + }, + ): Promise { + const { onUsage, onMessage, signal, logFile } = options; return new Promise((resolve, reject) => { - const logStream = logPath ? createWriteStream(logPath) : null; const child = spawn(this.bin, buildPiArgs(this.extraArgs), { cwd, detached: this.platform !== "win32", @@ -229,6 +264,7 @@ export class PiAgent implements Agent { stdio: ["pipe", "pipe", "pipe"], env: process.env, }); + logFile.track(child); child.stdin?.write(buildPiPrompt(prompt, this.schema)); child.stdin?.end(); @@ -299,7 +335,7 @@ export class PiAgent implements Agent { updateUsage(message, streaming); }; - parseJSONLStream(child.stdout!, logStream, (event) => { + parseJSONLStream(child.stdout!, logFile, (event) => { if (!isRecord(event)) return; if (event.type === "message_update") { @@ -358,7 +394,7 @@ export class PiAgent implements Agent { } }); - setupChildProcessHandlers(child, "pi", logStream, reject, () => { + setupChildProcessHandlers(child, "pi", reject, () => { if (latestAssistantMessage) { const stopReason = latestAssistantMessage.stopReason; if (stopReason === "error" || stopReason === "aborted") { @@ -379,7 +415,11 @@ export class PiAgent implements Agent { textByIndexToString(streamTextByIndex).trim(); if (!finalText) { - reject(new Error("pi returned no text output")); + appendDebugLog("pi:output:missing", { + recoverable: false, + reason: "pi runs with --no-session", + }); + reject(new Error(PI_EMPTY_RESPONSE_MESSAGE)); return; } diff --git a/src/core/agents/rovodev.test.ts b/src/core/agents/rovodev.test.ts index e2986d99..837fea87 100644 --- a/src/core/agents/rovodev.test.ts +++ b/src/core/agents/rovodev.test.ts @@ -23,8 +23,10 @@ vi.mock("../debug-log.js", () => ({ })); import { execFileSync, spawn } from "node:child_process"; +import { appendDebugLog } from "../debug-log.js"; const mockSpawn = vi.mocked(spawn); +const mockAppendDebugLog = vi.mocked(appendDebugLog); function createMockProcess() { const proc = Object.assign(new EventEmitter(), { @@ -411,6 +413,191 @@ describe("RovoDevAgent", () => { expect(getPort).toHaveBeenCalledTimes(1); }); + it("continues the same session once when a closed turn produced no text", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + + fetchMock + .mockResolvedValueOnce(jsonResponse({ status: "healthy" })) + .mockResolvedValueOnce( + jsonResponse({ session_id: "session-123", title: "gnhf" }), + ) + .mockResolvedValueOnce(jsonResponse({ message: "ok", prompt_set: true })) + .mockResolvedValueOnce(jsonResponse({ response: "Chat message set" })) + .mockResolvedValueOnce( + textResponse( + [ + "event: request-usage", + 'data: {"input_tokens":10,"cache_write_tokens":0,"cache_read_tokens":0,"output_tokens":4}', + "", + "event: close", + "data: ", + "", + ].join("\n"), + ), + ) + .mockResolvedValueOnce(jsonResponse({ response: "Chat message set" })) + .mockResolvedValueOnce( + textResponse( + [ + "event: part_start", + 'data: {"index":0,"part":{"content":"{\\"success\\":true,\\"summary\\":\\"recovered\\",\\"key_changes_made\\":[],\\"key_learnings\\":[]}","part_kind":"text"},"event_kind":"part_start"}', + "", + "event: request-usage", + 'data: {"input_tokens":5,"cache_write_tokens":0,"cache_read_tokens":0,"output_tokens":3}', + "", + "event: close", + "data: ", + "", + ].join("\n"), + ), + ) + .mockResolvedValueOnce(jsonResponse({ message: "deleted" })); + + const onUsage = vi.fn(); + const result = await agent.run("test", "/repo", { onUsage }); + + expect(result.output).toEqual({ + success: true, + summary: "recovered", + key_changes_made: [], + key_learnings: [], + }); + expect(result.usage).toMatchObject({ inputTokens: 15, outputTokens: 7 }); + expect(onUsage).toHaveBeenLastCalledWith(result.usage); + expect(mockAppendDebugLog).toHaveBeenCalledWith( + "rovodev:output:continuation", + expect.objectContaining({ sessionId: "session-123", attempt: 1 }), + ); + + const chatMessages = fetchMock.mock.calls.filter(([url]) => + String(url).includes("/v3/set_chat_message"), + ); + expect(chatMessages).toHaveLength(2); + expect(chatMessages[1]?.[1]?.body).toContain( + "You did not produce a final answer", + ); + }); + + it("continues when tool work leaves no final text segment", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + + fetchMock + .mockResolvedValueOnce(jsonResponse({ status: "healthy" })) + .mockResolvedValueOnce( + jsonResponse({ session_id: "session-123", title: "gnhf" }), + ) + .mockResolvedValueOnce(jsonResponse({ message: "ok", prompt_set: true })) + .mockResolvedValueOnce(jsonResponse({ response: "Chat message set" })) + .mockResolvedValueOnce( + textResponse( + [ + "event: part_start", + 'data: {"index":0,"part":{"content":"I will inspect the file.","part_kind":"text"},"event_kind":"part_start"}', + "", + "event: on_call_tools_start", + 'data: {"parts":[{"tool_name":"open_files","args":"{}","tool_call_id":"tool-1","part_kind":"tool-call"}]}', + "", + "event: tool-return", + 'data: {"tool_name":"open_files","content":"ok","tool_call_id":"tool-1","part_kind":"tool-return"}', + "", + "event: close", + "data: ", + "", + ].join("\n"), + ), + ) + .mockResolvedValueOnce(jsonResponse({ response: "Chat message set" })) + .mockResolvedValueOnce( + textResponse( + [ + "event: part_start", + 'data: {"index":0,"part":{"content":"{\\"success\\":true,\\"summary\\":\\"recovered\\",\\"key_changes_made\\":[],\\"key_learnings\\":[]}","part_kind":"text"},"event_kind":"part_start"}', + "", + "event: close", + "data: ", + "", + ].join("\n"), + ), + ) + .mockResolvedValueOnce(jsonResponse({ message: "deleted" })); + + const result = await agent.run("test", "/repo"); + + expect(result.output.summary).toBe("recovered"); + expect( + fetchMock.mock.calls.filter(([url]) => + String(url).includes("/v3/set_chat_message"), + ), + ).toHaveLength(2); + }); + + it("rejects after the rovodev continuation is also empty", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + + const emptyClosedTurn = () => + textResponse(["event: close", "data: ", ""].join("\n")); + + fetchMock + .mockResolvedValueOnce(jsonResponse({ status: "healthy" })) + .mockResolvedValueOnce( + jsonResponse({ session_id: "session-123", title: "gnhf" }), + ) + .mockResolvedValueOnce(jsonResponse({ message: "ok", prompt_set: true })) + .mockResolvedValueOnce(jsonResponse({ response: "Chat message set" })) + .mockResolvedValueOnce(emptyClosedTurn()) + .mockResolvedValueOnce(jsonResponse({ response: "Chat message set" })) + .mockResolvedValueOnce(emptyClosedTurn()) + .mockResolvedValueOnce(jsonResponse({ message: "deleted" })); + + await expect(agent.run("test", "/repo")).rejects.toThrow( + "rovodev returned no text output", + ); + expect( + fetchMock.mock.calls.filter(([url]) => + String(url).includes("/v3/set_chat_message"), + ), + ).toHaveLength(2); + }); + + it("does not continue when the stream ends without a close event", async () => { + const proc = createMockProcess(); + mockSpawn.mockReturnValue(proc); + + fetchMock + .mockResolvedValueOnce(jsonResponse({ status: "healthy" })) + .mockResolvedValueOnce( + jsonResponse({ session_id: "session-123", title: "gnhf" }), + ) + .mockResolvedValueOnce(jsonResponse({ message: "ok", prompt_set: true })) + .mockResolvedValueOnce(jsonResponse({ response: "Chat message set" })) + .mockResolvedValueOnce( + textResponse( + [ + "event: request-usage", + 'data: {"input_tokens":10,"cache_write_tokens":0,"cache_read_tokens":0,"output_tokens":4}', + "", + ].join("\n"), + ), + ) + .mockResolvedValueOnce(jsonResponse({ message: "deleted" })); + + await expect(agent.run("test", "/repo")).rejects.toThrow( + "rovodev returned no text output", + ); + expect( + fetchMock.mock.calls.filter(([url]) => + String(url).includes("/v3/set_chat_message"), + ), + ).toHaveLength(1); + expect(mockAppendDebugLog).not.toHaveBeenCalledWith( + "rovodev:output:continuation", + expect.anything(), + ); + }); + it("rejects when the final text is not valid JSON", async () => { const proc = createMockProcess(); mockSpawn.mockReturnValue(proc); diff --git a/src/core/agents/rovodev.ts b/src/core/agents/rovodev.ts index 5fc571ba..65b3db1b 100644 --- a/src/core/agents/rovodev.ts +++ b/src/core/agents/rovodev.ts @@ -14,6 +14,10 @@ import type { } from "./types.js"; import { validateAgentOutput } from "./types.js"; import { appendDebugLog, serializeError } from "../debug-log.js"; +import { + EmptyAgentResponseError, + runTurnWithEmptyResponseRetry, +} from "./empty-response.js"; import { parseAgentJson } from "./json-extract.js"; import { shutdownChildProcess } from "./managed-process.js"; @@ -227,22 +231,36 @@ export class RovoDevAgent implements Agent { try { const server = await this.ensureServer(cwd, runController.signal); sessionId = await this.createSession(server, runController.signal); - await this.setInlineSystemPrompt(server, sessionId, runController.signal); - await this.setChatMessage( + const activeSessionId = sessionId; + await this.setInlineSystemPrompt( server, - sessionId, - prompt, + activeSessionId, runController.signal, ); - const result = await this.streamChat( - server, - sessionId, - runController.signal, - logStream, + const result = await runTurnWithEmptyResponseRetry({ + logEvent: "rovodev:output:continuation", + logFields: { sessionId: activeSessionId }, onUsage, - onMessage, - ); + signal: runController.signal, + initialText: prompt, + runTurn: async (text, onTurnUsage) => { + await this.setChatMessage( + server, + activeSessionId, + text, + runController.signal, + ); + return this.streamChat( + server, + activeSessionId, + runController.signal, + logStream, + onTurnUsage, + onMessage, + ); + }, + }); appendDebugLog("rovodev:run:end", { sessionId, elapsedMs: Date.now() - runStartedAt, @@ -555,6 +573,10 @@ export class RovoDevAgent implements Agent { cacheCreationTokens: 0, }; let latestTextSegment = ""; + // Rovo Dev terminates a turn with an `event: close` frame. Seeing it is the + // only way to tell a finished-but-silent turn from a stream that was cut + // short, so it gates the empty-response continuation. + let sawClose = false; let currentTextParts: string[] = []; let currentTextIndexes = new Map(); const decoder = new TextDecoder(); @@ -570,6 +592,7 @@ export class RovoDevAgent implements Agent { }; const resetCurrentMessage = () => { + latestTextSegment = ""; currentTextParts = []; currentTextIndexes = new Map(); }; @@ -598,6 +621,11 @@ export class RovoDevAgent implements Agent { } } + if (eventName === "close") { + sawClose = true; + return; + } + const rawData = dataLines.join("\n"); if (rawData.length === 0) return; @@ -614,6 +642,11 @@ export class RovoDevAgent implements Agent { ? ((payload as Record).event_kind as string) : ""); + if (kind === "close") { + sawClose = true; + return; + } + if (kind === "request-usage") { handleUsage(payload as RovoDevRequestUsageEvent); return; @@ -735,12 +768,16 @@ export class RovoDevAgent implements Agent { sessionId, elapsedMs: Date.now() - streamStartedAt, bytesRead, + sawClose, }); const finalText = latestTextSegment.trim(); if (!finalText) { - appendDebugLog("rovodev:output:missing", { sessionId }); - throw new Error("rovodev returned no text output"); + appendDebugLog("rovodev:output:missing", { sessionId, sawClose }); + throw new EmptyAgentResponseError("rovodev returned no text output", { + turnCompleted: sawClose, + usage: { ...usage }, + }); } const schema = JSON.parse( diff --git a/src/core/agents/stream-utils.test.ts b/src/core/agents/stream-utils.test.ts index 81a00491..b1238e59 100644 --- a/src/core/agents/stream-utils.test.ts +++ b/src/core/agents/stream-utils.test.ts @@ -1,6 +1,10 @@ import { EventEmitter } from "node:events"; -import { describe, it, expect, vi } from "vitest"; +import { mkdtempSync, readFileSync, rmSync } from "node:fs"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import { afterEach, describe, it, expect, vi } from "vitest"; import { + AgentLogFile, parseJSONLStream, setupAbortHandler, setupChildProcessHandlers, @@ -20,6 +24,80 @@ function createMockReadable() { return new EventEmitter(); } +describe("AgentLogFile", () => { + const tempDirs: string[] = []; + + function tempLogPath() { + const dir = mkdtempSync(join(tmpdir(), "gnhf-log-")); + tempDirs.push(dir); + return join(dir, "iteration.jsonl"); + } + + afterEach(() => { + while (tempDirs.length > 0) { + rmSync(tempDirs.pop()!, { recursive: true, force: true }); + } + }); + + it("keeps recording stdout that arrives after an aborted run finished", async () => { + const logPath = tempLogPath(); + const logFile = new AgentLogFile(logPath); + const child = createMockChild(); + const stdout = createMockReadable(); + logFile.track(child as never); + parseJSONLStream(stdout as never, logFile, () => {}); + + stdout.emit("data", Buffer.from('{"type":"before-abort"}\n')); + // The abort path rejects the run while the child is still streaming. + const closed = logFile.finish(); + stdout.emit("data", Buffer.from('{"type":"after-abort"}\n')); + child.emit("close", 143); + await closed; + + const written = readFileSync(logPath, "utf8"); + expect(written).toContain("before-abort"); + expect(written).toContain("after-abort"); + }); + + it("stays open across a second spawn and closes once the last child exits", async () => { + const logPath = tempLogPath(); + const logFile = new AgentLogFile(logPath); + const first = createMockChild(); + logFile.track(first as never); + logFile.write('{"turn":1}\n'); + first.emit("close", 0); + + const second = createMockChild(); + logFile.track(second as never); + logFile.write('{"turn":2}\n'); + const closed = logFile.finish(); + second.emit("close", 0); + await closed; + + const written = readFileSync(logPath, "utf8"); + expect(written).toContain('{"turn":1}'); + expect(written).toContain('{"turn":2}'); + + // A late stdout chunk after the file closed must not raise + // ERR_STREAM_WRITE_AFTER_END. + expect(() => logFile.write('{"turn":"late"}\n')).not.toThrow(); + expect(readFileSync(logPath, "utf8")).not.toContain("late"); + }); + + it("closes a run whose child failed to spawn", async () => { + const logPath = tempLogPath(); + const logFile = new AgentLogFile(logPath); + const child = createMockChild(); + logFile.track(child as never); + logFile.write('{"spawn":"failed"}\n'); + + child.emit("error", new Error("ENOENT")); + await logFile.finish(); + + expect(readFileSync(logPath, "utf8")).toContain('{"spawn":"failed"}'); + }); +}); + describe("parseJSONLStream", () => { it("parses complete JSONL events across chunk boundaries and writes chunks to the log", () => { const stream = createMockReadable(); @@ -73,20 +151,12 @@ describe("setupChildProcessHandlers", () => { const child = createMockChild(); const reject = vi.fn(); const onSuccess = vi.fn(); - const logStream = { end: vi.fn() }; - setupChildProcessHandlers( - child as never, - "codex", - logStream as never, - reject, - onSuccess, - ); + setupChildProcessHandlers(child as never, "codex", reject, onSuccess); child.stderr.emit("data", Buffer.from("boom")); child.emit("close", 2); - expect(logStream.end).toHaveBeenCalledTimes(1); expect(onSuccess).not.toHaveBeenCalled(); expect(reject).toHaveBeenCalledWith( new Error("codex exited with code 2: boom"), @@ -97,15 +167,8 @@ describe("setupChildProcessHandlers", () => { const child = createMockChild(); const reject = vi.fn(); const onSuccess = vi.fn(); - const logStream = { end: vi.fn() }; - setupChildProcessHandlers( - child as never, - "rovodev", - logStream as never, - reject, - onSuccess, - ); + setupChildProcessHandlers(child as never, "rovodev", reject, onSuccess); child.emit("error", new Error("ENOENT")); expect(reject).toHaveBeenCalledWith( @@ -113,7 +176,6 @@ describe("setupChildProcessHandlers", () => { ); child.emit("close", 0); - expect(logStream.end).toHaveBeenCalledTimes(1); expect(onSuccess).toHaveBeenCalledTimes(1); }); }); diff --git a/src/core/agents/stream-utils.ts b/src/core/agents/stream-utils.ts index 22899469..6ea7f5f3 100644 --- a/src/core/agents/stream-utils.ts +++ b/src/core/agents/stream-utils.ts @@ -1,16 +1,86 @@ import type { ChildProcess } from "node:child_process"; import type { Readable } from "node:stream"; -import type { WriteStream } from "node:fs"; +import { createWriteStream } from "node:fs"; +import { appendDebugLog, serializeError } from "../debug-log.js"; + +/** Sink for raw agent stdout. */ +export interface AgentLogSink { + write(chunk: string | Buffer): void; +} + +/** + * Per-run log file for adapters that spawn a CLI once per turn. + * + * An agent run can span more than one spawn (an empty-response continuation + * reuses the same log file), and a turn can be rejected - by an abort, say - + * while its child is still streaming stdout. Ending the file on either event + * alone would truncate a later turn or write after end, so the file is closed + * only once the run has finished *and* every child it spawned has exited. + */ +export class AgentLogFile implements AgentLogSink { + private readonly stream: ReturnType | null; + private readonly closePromise: Promise; + private openChildren = 0; + private runFinished = false; + private ended = false; + + constructor(logPath?: string) { + const stream = logPath ? createWriteStream(logPath) : null; + this.stream = stream; + this.closePromise = stream + ? new Promise((resolve) => stream.once("close", () => resolve())) + : Promise.resolve(); + // Log writes are best effort; a failed write must not take down the run. + stream?.on("error", (error) => { + appendDebugLog("agent:log:write-failed", { + error: serializeError(error), + }); + }); + } + + write(chunk: string | Buffer): void { + if (this.ended) return; + this.stream?.write(chunk); + } + + /** Keep the file open until `child` has exited. */ + track(child: ChildProcess): void { + this.openChildren += 1; + let settled = false; + const onChildGone = () => { + if (settled) return; + settled = true; + this.openChildren -= 1; + this.endIfIdle(); + }; + child.on("close", onChildGone); + child.on("error", onChildGone); + } + + /** The run is done; close as soon as no tracked child is still running. */ + finish(): Promise { + this.runFinished = true; + this.endIfIdle(); + return this.closePromise; + } + + private endIfIdle(): void { + if (this.ended || !this.runFinished || this.openChildren > 0) return; + this.ended = true; + this.stream?.end(); + } +} /** - * Wire stderr collection, spawn-error handling, and the common close-handler - * prefix (logStream.end + non-zero exit code rejection) for a child process. - * Calls `onSuccess` only when the process exits with code 0. + * Wire stderr collection, spawn-error handling, and non-zero exit rejection + * for a child process. Calls `onSuccess` only when the process exits with + * code 0. + * + * The log file is deliberately not owned here - see `AgentLogFile`. */ export function setupChildProcessHandlers( child: ChildProcess, agentName: string, - logStream: WriteStream | null, reject: (err: Error) => void, onSuccess: () => void, ): void { @@ -25,7 +95,6 @@ export function setupChildProcessHandlers( }); child.on("close", (code) => { - logStream?.end(); if (code !== 0) { reject(new Error(`${agentName} exited with code ${code}: ${stderr}`)); return; @@ -40,7 +109,7 @@ export function setupChildProcessHandlers( */ export function parseJSONLStream( stream: Readable, - logStream: WriteStream | null, + logStream: AgentLogSink | null, callback: (event: T) => void, ): void { let buffer = ""; diff --git a/src/core/config.test.ts b/src/core/config.test.ts index fe188bb9..e1828ab8 100644 --- a/src/core/config.test.ts +++ b/src/core/config.test.ts @@ -430,6 +430,19 @@ describe("loadConfig", () => { ); }); + it.each(["-r", "--resume", "--resume=abc123"])( + "throws when agentArgsOverride.claude contains reserved flag %s", + (flag) => { + mockReadFileSync.mockReturnValue( + `agentArgsOverride:\n claude:\n - ${flag}\n`, + ); + + expect(() => loadConfig()).toThrow( + /agentArgsOverride\.claude\[0\].*managed by gnhf/, + ); + }, + ); + it("reads acpRegistryOverrides from config", () => { mockReadFileSync.mockReturnValue( [ diff --git a/src/core/config.ts b/src/core/config.ts index 5613639f..b287cd4e 100644 --- a/src/core/config.ts +++ b/src/core/config.ts @@ -118,11 +118,15 @@ function isReservedAgentArg(agent: AgentName, arg: string): boolean { arg === "--output-format" || arg.startsWith("--output-format=") || arg === "--json-schema" || - arg.startsWith("--json-schema=") + arg.startsWith("--json-schema=") || + arg === "-r" || + arg === "--resume" || + arg.startsWith("--resume=") ); case "codex": return ( arg === "exec" || + arg === "resume" || arg === "--json" || arg === "--output-schema" || arg.startsWith("--output-schema=") || diff --git a/src/no-mistakes-required.test.ts b/src/no-mistakes-required.test.ts new file mode 100644 index 00000000..6704228c --- /dev/null +++ b/src/no-mistakes-required.test.ts @@ -0,0 +1,118 @@ +import { spawnSync } from "node:child_process"; +import { + chmodSync, + mkdtempSync, + mkdirSync, + readFileSync, + rmSync, + writeFileSync, +} from "node:fs"; +import { tmpdir } from "node:os"; +import { delimiter, join } from "node:path"; +import { fileURLToPath } from "node:url"; +import { afterEach, describe, expect, it } from "vitest"; +import yaml from "js-yaml"; + +const root = fileURLToPath(new URL("..", import.meta.url)); +const temporaryDirectories: string[] = []; + +function verificationScript(): string { + const workflow = yaml.load( + readFileSync( + join(root, ".github", "workflows", "no-mistakes-required.yml"), + "utf8", + ), + ) as { + jobs: { check: { steps: Array<{ name?: string; run?: string }> } }; + }; + const step = workflow.jobs.check.steps.find( + ({ name }) => name === "Verify no-mistakes signature in PR body", + ); + if (!step?.run) throw new Error("no-mistakes verification step is missing"); + return step.run; +} + +function runVerification(signedAfter: number) { + const temp = mkdtempSync(join(tmpdir(), "gnhf-no-mistakes-check-")); + temporaryDirectories.push(temp); + const bin = join(temp, "bin"); + const countFile = join(temp, "gh-count"); + mkdirSync(bin); + + const gh = join(bin, "gh"); + writeFileSync( + gh, + `#!/bin/sh +set -eu +count=0 +if [ -f "$MOCK_GH_COUNT_FILE" ]; then + count="$(cat "$MOCK_GH_COUNT_FILE")" +fi +count=$((count + 1)) +printf '%s' "$count" > "$MOCK_GH_COUNT_FILE" +if [ "$count" -ge "$MOCK_SIGNED_AFTER" ]; then + printf '%s\n' "$MOCK_SIGNED_BODY" +else + printf '%s\n' "$MOCK_UNSIGNED_BODY" +fi +`, + ); + chmodSync(gh, 0o755); + + const sleep = join(bin, "sleep"); + writeFileSync(sleep, "#!/bin/sh\nexit 0\n"); + chmodSync(sleep, 0o755); + + const result = spawnSync("bash", ["-c", verificationScript()], { + encoding: "utf8", + env: { + ...process.env, + PATH: `${bin}${delimiter}${process.env.PATH ?? ""}`, + GH_TOKEN: "test-token", + GITHUB_REPOSITORY: "kunchenguid/gnhf", + PR_AUTHOR: "contributor", + PR_NUMBER: "198", + MOCK_GH_COUNT_FILE: countFile, + MOCK_SIGNED_AFTER: String(signedAfter), + MOCK_UNSIGNED_BODY: "## What Changed\n\nPending pipeline summary.", + MOCK_SIGNED_BODY: + "## Pipeline\n\nUpdates from [git push no-mistakes](https://github.com/kunchenguid/no-mistakes)", + }, + }); + + return { + ...result, + calls: Number(readFileSync(countFile, "utf8")), + }; +} + +afterEach(() => { + for (const directory of temporaryDirectories.splice(0)) { + rmSync(directory, { recursive: true, force: true }); + } +}); + +describe.skipIf(process.platform === "win32")( + "no-mistakes PR signature workflow", + () => { + it("accepts a signature written shortly after the synchronize event", () => { + const result = runVerification(2); + + expect(result.status).toBe(0); + expect(result.calls).toBe(2); + expect(result.stdout).toContain( + "Found no-mistakes signature in PR #198 body.", + ); + }); + + it("still rejects a PR whose live body remains unsigned", () => { + const result = runVerification(6); + + expect(result.status).toBe(1); + expect(result.calls).toBe(5); + expect(result.stderr).toContain( + "This PR was not raised through no-mistakes.", + ); + }); + }, +);