Compare commits

...
16 changed files with 1108 additions and 128 deletions

No files matched your search

+3 -2
View File
@@ -662,8 +662,9 @@ credentials through OpenCode.
- **Anthropic / Claude (the default)** — use `anthropic/...` model ids and every
anthropic pass runs through the **Claude Code CLI** (`claude -p
--output-format json`), inferred from the model. The credential is (in order) a
`tokenEnv` you name, an ambient `CLAUDE_CODE_OAUTH_TOKEN`, or your local
--output-format stream-json --verbose`), inferred from the model. The
credential is (in order) a `tokenEnv` you name, an ambient
`CLAUDE_CODE_OAUTH_TOKEN`, or your local
`claude` login — an `auth` entry is entirely optional. Run `claude setup-token`
for a Max/Team subscription token or point `tokenEnv` at an Anthropic Console
API key (`sk-ant-api…`, forwarded as `ANTHROPIC_API_KEY`); the CLI reads
@@ -99,9 +99,38 @@ oversight, and must not be silently reintroduced as a "fix" ([observed]
## Claude Code CLI Containment
The Claude Code engine runs each review pass as a single stateless
`claude -p --output-format json` subprocess fed the task on stdin. The argv and child
environment are built to keep an untrusted PR (the review input) from turning that
subprocess into code execution or credential exfiltration.
`claude -p --output-format stream-json --verbose` subprocess fed the task on stdin.
The JSONL stream makes tool activity observable while the pass runs instead of leaving
only a periodic heartbeat until the final result. The incremental decoder emits only
bounded lifecycle metadata and read-tool names with confined repo-relative targets;
raw assistant text, tool results, grep patterns, and attempted host paths never reach
the progress or error logs because model output is untrusted and may contain source
secrets or terminal/log-injection payloads. If a stream ends without a final result,
or its error result has no explicit message, parsing returns a fixed diagnostic rather
than falling back to the raw JSONL transcript. The fixed diagnostic carries the child
exit code and may derive an actionable, fixed category from stderr only when no stream
was emitted (rejected flags, authentication, or launch failure); arbitrary stderr is
never copied into logs, and only an explicit final-result error message reaches
provider-error classification. Every task-specific progress line is tagged with the
stable agent bucket by the review pipeline, so concurrent passes remain attributable
[observed] `claude-code.ts` `claudeExitDiagnostic`/`runClaudePrompt`, `review.ts`
`formatAgentActivity`/`taskProgress`; tests in
`claude-code.test.ts` and `review-internals.test.ts`. The argv and child environment
are built to keep an untrusted PR (the review input) from turning that subprocess into
code execution or credential exfiltration.
Every production structured-output parser carries a draft-07 JSON Schema generated
from the same Zod contract used at the local trust boundary. The Claude runtime passes
that schema through `--json-schema`; Claude Code validates the final `StructuredOutput`
tool call and re-prompts inside the original session when a required field or type is
wrong. The validated `structured_output` object is then serialized through the local
Zod parser again — provider validation improves reliability but never replaces local
validation. If Claude exhausts its in-session structured-output retries (or claims
success without returning `structured_output`), the runtime makes one clean-process
attempt and accounts for both attempts; it never accepts unvalidated fallback text
[observed] `schema.ts` `structuredParser`; `opencode.ts` `promptAndParse`;
`claude-code.ts` `buildClaudeArgs`, `parseClaudeResult`, `runClaudePrompt`,
`claudeCodePromptAndParse`.
**Flags.** `--safe-mode` disables CLAUDE.md/hooks/MCP/plugins while keeping OAuth;
`--bare` is deliberately **not** used because bare mode ignores
@@ -247,6 +276,14 @@ child resolves with `timedOut: true` in the result regardless of the `check` fla
callers get one consistent shape to distinguish "our own deadline fired" from a real
command failure ([observed] `exec.ts:71-87, 181-189, 230-250`).
**Structured stdout observation stays inside the capture bound.** Spawn/input callers
may observe stdout chunks as they arrive, but only the bytes admitted by `maxBuffer`;
the observer is best-effort and its exceptions cannot affect the child. Claude uses
this seam to decode `stream-json` without piping raw subprocess output to the terminal,
while the complete bounded stdout remains available for final-result parsing
([observed] `exec.ts` `RunOptions.onStdout` / `runWithInput`; `claude-code.ts`
`createClaudeActivityStream`).
## Retry Taxonomy
The runtime tells apart five failure classes and handles each differently. Conflating
@@ -271,12 +308,21 @@ retry, because re-sending the whole context into a limited account only makes it
always capped at half the pass's own `maxWaitMs` so it can never outlast the deadline it
protects ([observed] `opencode.ts:552-555`).
**JSON-parse failure — same-session corrective first.** A reply that will not parse is
**JSON-parse failure — same-session corrective first.** Claude Code receives the local
parser's draft-07 JSON Schema and performs validation-aware retries inside the original
session before returning `structured_output`; this retains the investigation context
and names the exact missing field/type to the model. Local Zod validation still runs on
the provider-validated object. If those in-session repairs are exhausted, Claude gets
one fresh-process corrective attempt; both attempts remain in cost/token accounting.
OpenCode has no equivalent structured-output seam, so a reply that will not parse is
retried in the **same** OpenCode session first: the model still holds all the file
context it read, so the corrective is a cheap cache-read re-emit with better recall than
re-investigating; only if that also fails does it fall back to a fresh session. Parse
failure and timeout are handled by entirely separate mechanisms and must not be
conflated ([observed] `opencode.ts:994-1003`).
context it read, making the corrective a cheap cache-read
re-emit with better recall than re-investigating; only if that also fails does it fall
back to a fresh session. Claude's legacy/no-schema path retains one fresh-process
corrective as defense in depth. Parse failure and timeout are handled by entirely
separate mechanisms and must not be conflated ([observed] `schema.ts`
`structuredParser`; `claude-code.ts` `claudeCodePromptAndParse`; `opencode.ts`
`promptAndParse`).
**Finalize and corrective prompts run with every tool disabled.** The "stop and return
what you have" finalize and the "re-emit valid JSON" corrective are sent with `NO_TOOLS`
+503 -10
View File
@@ -9,11 +9,13 @@ import {
assertClaudeModels,
buildClaudeArgs,
buildEngineMap,
claudeActivities,
claudeCodePromptAndParse,
claudeSubscriptionActive,
claudeTemperatureNote,
claudeTokenCredential,
classifyClaudeError,
createClaudeActivityStream,
claudeModelId,
claudeModelMatches,
engineForModel,
@@ -25,7 +27,14 @@ import {
usageLimitResetMs,
} from "../core/claude-code.js";
import type { ClaudeCodeHandle } from "../core/claude-code.js";
import { AgentTimeoutError, CLAUDE_CODE_ENGINE, isTransientApiError } from "../core/opencode.js";
import {
AgentTimeoutError,
CLAUDE_CODE_ENGINE,
isTransientApiError,
promptAndParse,
} from "../core/opencode.js";
import type { OpencodeHandle } from "../core/opencode.js";
import { parseReviewerOutput } from "../core/schema.js";
import { RateLimitWatch } from "../core/throttle.js";
import type { AuthConfigEntry, LoadedConfig } from "../config/schema.js";
@@ -50,7 +59,7 @@ async function fakeClaude(script: string): Promise<{ bin: string; cleanup: () =>
}
/**
* A fake `claude` that returns a different canned `--output-format json` body on
* A fake `claude` that returns a different canned final-result body on
* each successive invocation (tracked via a counter file, since childEnv is fixed
* for the whole test) and logs each call's stdin (the prompt actually sent) to
* `<dir>/stdin.<n>` via `stdinLog(n)` — so a corrective-retry test can assert what
@@ -60,6 +69,7 @@ async function fakeClaude(script: string): Promise<{ bin: string; cleanup: () =>
async function sequencedFakeClaude(outputs: string[]): Promise<{
bin: string;
stdinLog: (n: number) => Promise<string>;
argsLog: (n: number) => Promise<string>;
cleanup: () => Promise<void>;
}> {
const dir = await mkdtemp(path.join(tmpdir(), "ecr-claude-code-test-"));
@@ -76,6 +86,7 @@ async function sequencedFakeClaude(outputs: string[]): Promise<{
"COUNT=$((COUNT + 1))",
'echo "$COUNT" > "$COUNT_FILE"',
`cat > "${dir}/stdin.$COUNT"`,
`printf '%s\n' "$@" > "${dir}/args.$COUNT"`,
`if [ "$COUNT" -gt ${outputs.length} ]; then COUNT=${outputs.length}; fi`,
`cat "${dir}/out.$COUNT"`,
"",
@@ -85,6 +96,7 @@ async function sequencedFakeClaude(outputs: string[]): Promise<{
return {
bin,
stdinLog: (n: number) => readFile(path.join(dir, `stdin.${n}`), "utf8"),
argsLog: (n: number) => readFile(path.join(dir, `args.${n}`), "utf8"),
cleanup: () => rm(dir, { recursive: true, force: true }),
};
}
@@ -206,16 +218,25 @@ test("claudeModelMatches: dated-suffix family match; cross-family mismatch", ()
});
test("buildClaudeArgs: read-only, trust-isolated, subscription argv; never --bare", () => {
const jsonSchema = {
type: "object",
properties: { title: { type: "string" } },
required: ["title"],
};
const args = buildClaudeArgs({
model: "claude-opus-5",
system: "SYS",
cwd: "/work/repo",
maxTurns: 60,
jsonSchema,
});
const joined = args.join(" ");
expect(joined).toContain("--output-format json");
expect(joined).toContain("--output-format stream-json --verbose");
expect(joined).toContain("--model claude-opus-5");
expect(joined).toContain("--append-system-prompt SYS");
const schemaIndex = args.indexOf("--json-schema");
expect(schemaIndex).toBeGreaterThan(-1);
expect(JSON.parse(args[schemaIndex + 1]!)).toEqual(jsonSchema);
// Read, Grep, AND Glob are all path-scoped to the review tree — a bare Grep
// would read arbitrary files (credentials, /proc/self/environ) despite the
// scoped Read, defeating the whole exfiltration defense.
@@ -260,6 +281,144 @@ test("parseClaudeResult: success maps text/cost/tokens/model", () => {
expect(r.cost).toBe(0.0123);
expect(r.tokens).toEqual({ input: 100, output: 200, cache: { write: 50, read: 30 } });
expect(r.modelOutputTokens).toEqual({ "claude-opus-5-20250101": 200 });
expect(r.hasStructuredOutput).toBe(false);
expect(r.structuredOutputFailure).toBe(false);
});
test("parseClaudeResult: reads the final result from stream-json JSONL", () => {
const stdout = [
JSON.stringify({ type: "system", subtype: "init", model: "claude-opus-5" }),
JSON.stringify({
type: "assistant",
message: { content: [{ type: "tool_use", id: "tool-1", name: "Read", input: {} }] },
}),
JSON.stringify({
type: "result",
subtype: "success",
is_error: false,
result: "the findings",
total_cost_usd: 0.02,
usage: { input_tokens: 12, output_tokens: 34 },
}),
].join("\n");
const result = parseClaudeResult(stdout);
expect(result.isError).toBe(false);
expect(result.text).toBe("the findings");
expect(result.cost).toBe(0.02);
expect(result.tokens).toEqual({
input: 12,
output: 34,
cache: { write: undefined, read: undefined },
});
});
test("parseClaudeResult: prefers provider-validated structured_output", () => {
const stdout = JSON.stringify({
type: "result",
subtype: "success",
is_error: false,
// Keep a conflicting text result to prove the structured object wins.
result: '{"title":"wrong"}',
structured_output: { title: "validated" },
});
const result = parseClaudeResult(stdout);
expect(result.isError).toBe(false);
expect(JSON.parse(result.text)).toEqual({ title: "validated" });
expect(result.hasStructuredOutput).toBe(true);
});
test("parseClaudeResult: an error ignores stale structured_output", () => {
const stdout = JSON.stringify({
type: "result",
subtype: "error_max_structured_output_retries",
is_error: true,
result: "Could not produce valid structured output",
structured_output: { title: "PR_SECRET rate limit" },
});
const result = parseClaudeResult(stdout);
expect(result.isError).toBe(true);
expect(result.errorText).toBe("Could not produce valid structured output");
expect(result.errorText).not.toContain("PR_SECRET");
expect(result.structuredOutputFailure).toBe(true);
});
test("claudeActivities: reports lifecycle and safe tool targets, never raw text/results/patterns", () => {
expect(
claudeActivities({ type: "system", subtype: "init", model: "claude-opus-5" }, "/repo"),
).toEqual([{ line: "started claude-opus-5" }]);
expect(
claudeActivities(
{
type: "assistant",
message: {
content: [
{ type: "text", text: "secret reasoning" },
{
type: "tool_use",
id: "read-1",
name: "Read",
input: { file_path: "/repo/src/auth.ts" },
},
{
type: "tool_use",
id: "grep-1",
name: "Grep",
input: { path: "/repo/src", pattern: "DO_NOT_LOG_THIS" },
},
{
type: "tool_use",
id: "outside-1",
name: "Read",
input: { file_path: "/etc/passwd\nforged log line" },
},
{
type: "tool_use",
id: "glob-1",
name: "Glob",
input: { pattern: "SECRET_PATTERN" },
},
],
},
},
"/repo",
),
).toEqual([
{ key: "read-1", line: "Read src/auth.ts" },
{ key: "grep-1", line: "Grep src" },
{ key: "outside-1", line: "Read" },
{ key: "glob-1", line: "Glob" },
]);
expect(
claudeActivities(
{ type: "result", is_error: false, duration_ms: 12_400, num_turns: 3 },
"/repo",
),
).toEqual([{ line: "completed (12s, 3 turn(s))" }]);
expect(
claudeActivities(
{ type: "user", message: { content: [{ type: "tool_result", content: "SECRET" }] } },
"/repo",
),
).toEqual([]);
});
test("createClaudeActivityStream: handles chunk boundaries and dedupes repeated tool ids", () => {
const lines: string[] = [];
const stream = createClaudeActivityStream("/repo", (line) => lines.push(line));
const init = `${JSON.stringify({ type: "system", subtype: "init", model: "claude-opus-5" })}\n`;
const tool = JSON.stringify({
type: "assistant",
message: {
content: [
{ type: "tool_use", id: "tool-1", name: "Read", input: { file_path: "/repo/a.ts" } },
],
},
});
stream.push(init.slice(0, 15));
stream.push(`${init.slice(15)}${tool}\n${tool.slice(0, 20)}`);
stream.push(tool.slice(20));
stream.finish();
expect(lines).toEqual(["started claude-opus-5", "Read a.ts"]);
});
test("pickAnsweringModel: prefers the requested family over key order; else max output", () => {
@@ -298,7 +457,37 @@ test("parseClaudeResult: is_error true even when subtype stays 'success'", () =>
test("parseClaudeResult: unparseable stdout is an error, not a throw", () => {
const r = parseClaudeResult("not json at all");
expect(r.isError).toBe(true);
expect(r.errorText).toBe("not json at all");
expect(r.errorText).toBe("Claude Code stream ended without a final result event");
});
test("parseClaudeResult: a stream without a final result does not expose its transcript", () => {
const secret = "PR_SECRET rate limit";
const stdout = [
JSON.stringify({ type: "system", subtype: "init", model: "claude-opus-5" }),
JSON.stringify({
type: "user",
message: { content: [{ type: "tool_result", content: secret }] },
}),
].join("\n");
const result = parseClaudeResult(stdout);
expect(result.isError).toBe(true);
expect(result.errorText).toBe("Claude Code stream ended without a final result event");
expect(result.errorText).not.toContain(secret);
});
test("parseClaudeResult: an error result without a message does not expose its transcript", () => {
const secret = "PR_SECRET quota";
const stdout = [
JSON.stringify({
type: "user",
message: { content: [{ type: "tool_result", content: secret }] },
}),
JSON.stringify({ type: "result", subtype: "error", is_error: true, result: "" }),
].join("\n");
const result = parseClaudeResult(stdout);
expect(result.isError).toBe(true);
expect(result.errorText).toBe("Claude Code returned an error without a message");
expect(result.errorText).not.toContain(secret);
});
test("classifyClaudeError: rate-limit / auth / usage-limit / other", () => {
@@ -663,17 +852,25 @@ test("runClaudePrompt: our own deadline kills the child and throws AgentTimeoutE
}
});
test("runClaudePrompt: a non-timeout signal throws a crash error naming the signal", async () => {
const { bin, cleanup } = await fakeClaude("#!/bin/sh\ncat > /dev/null\nkill -9 $$\n");
test("runClaudePrompt: a non-timeout signal names the signal without exposing stderr", async () => {
const secret = "PR_SECRET from stderr";
const { bin, cleanup } = await fakeClaude(
`#!/bin/sh\ncat > /dev/null\nprintf '%s' '${secret}' >&2\nkill -9 $$\n`,
);
try {
await expect(
runClaudePrompt(testHandle(bin), {
let message = "";
try {
await runClaudePrompt(testHandle(bin), {
agent: "reviewer",
system: "SYS",
text: "hi",
title: "t",
}),
).rejects.toThrow(/killed by signal SIGKILL/);
});
} catch (error) {
message = error instanceof Error ? error.message : String(error);
}
expect(message).toContain("killed by signal SIGKILL");
expect(message).not.toContain(secret);
} finally {
await cleanup();
}
@@ -720,6 +917,115 @@ test("runClaudePrompt: a 429 result dispatches to a rate-limit throw and records
}
});
test("runClaudePrompt: an incomplete stream cannot forge rate-limit evidence or leak tool results", async () => {
const secret = "PR_SECRET rate limit quota usage limit reached|1728000000";
const body = [
JSON.stringify({ type: "system", subtype: "init", model: "claude-opus-5" }),
JSON.stringify({
type: "user",
message: { content: [{ type: "tool_result", content: secret }] },
}),
].join("\n");
const { bin, cleanup } = await fakeClaude(`#!/bin/sh\ncat > /dev/null\nprintf '%s' '${body}'\n`);
try {
const handle = testHandle(bin);
let message = "";
try {
await runClaudePrompt(handle, {
agent: "reviewer",
system: "SYS",
text: "hi",
title: "t",
});
} catch (error) {
message = error instanceof Error ? error.message : String(error);
}
expect(message).toContain("stream ended without a final result event");
expect(message).toContain("Claude Code exited with code 0");
expect(message).not.toContain(secret);
expect(handle.rateLimit.events).toBe(0);
} finally {
await cleanup();
}
});
test("runClaudePrompt: an early bad-flag failure reports a safe fixed diagnostic", async () => {
const secret = "PR_SECRET from stderr";
const { bin, cleanup } = await fakeClaude(
`#!/bin/sh\ncat > /dev/null\nprintf '%s' "error: unknown option '--json-schema' ${secret}" >&2\nexit 2\n`,
);
try {
let message = "";
try {
await runClaudePrompt(testHandle(bin), {
agent: "reviewer",
system: "SYS",
text: "hi",
title: "t",
});
} catch (error) {
message = error instanceof Error ? error.message : String(error);
}
expect(message).toContain("Claude Code exited with code 2");
expect(message).toContain("CLI rejected its arguments");
expect(message).toContain("verify the pinned @anthropic-ai/claude-code version");
expect(message).not.toContain("--json-schema");
expect(message).not.toContain(secret);
} finally {
await cleanup();
}
});
test("runClaudePrompt: an early auth failure names the fix without copying stderr", async () => {
const secret = "PR_SECRET from auth stderr";
const { bin, cleanup } = await fakeClaude(
`#!/bin/sh\ncat > /dev/null\nprintf '%s' 'OAuth authentication failed: ${secret}' >&2\nexit 1\n`,
);
try {
let message = "";
try {
await runClaudePrompt(testHandle(bin), {
agent: "reviewer",
system: "SYS",
text: "hi",
title: "t",
});
} catch (error) {
message = error instanceof Error ? error.message : String(error);
}
expect(message).toContain("Claude Code exited with code 1");
expect(message).toContain("authentication failed before a result was emitted");
expect(message).toContain("`claude auth status` and `ecr doctor`");
expect(message).not.toContain(secret);
} finally {
await cleanup();
}
});
test("runClaudePrompt: unknown early stderr is never copied into the failure", async () => {
const secret = "PR_SECRET arbitrary diagnostic";
const { bin, cleanup } = await fakeClaude(
`#!/bin/sh\ncat > /dev/null\nprintf '%s' '${secret}' >&2\nexit 7\n`,
);
try {
let message = "";
try {
await runClaudePrompt(testHandle(bin), {
agent: "reviewer",
system: "SYS",
text: "hi",
title: "t",
});
} catch (error) {
message = error instanceof Error ? error.message : String(error);
}
expect(message).toContain("Claude Code exited with code 7");
expect(message).not.toContain(secret);
} finally {
await cleanup();
}
});
test("runClaudePrompt: a usage-limit result throws the non-retryable, reset-time-carrying message", async () => {
const body = JSON.stringify({
type: "result",
@@ -743,6 +1049,53 @@ test("runClaudePrompt: a usage-limit result throws the non-retryable, reset-time
}
});
test("runClaudePrompt: streams safe tool activity while preserving the final result", async () => {
const body = [
JSON.stringify({ type: "system", subtype: "init", model: "claude-opus-5" }),
JSON.stringify({
type: "assistant",
message: {
content: [
{
type: "tool_use",
id: "read-1",
name: "Read",
input: { file_path: process.cwd() + "/src/core/auth.ts" },
},
],
},
}),
JSON.stringify({
type: "result",
subtype: "success",
is_error: false,
result: "the findings",
duration_ms: 2500,
num_turns: 2,
total_cost_usd: 0.01,
}),
].join("\n");
const { bin, cleanup } = await fakeClaude(`#!/bin/sh\ncat > /dev/null\nprintf '%s' '${body}'\n`);
try {
const activity: string[] = [];
const result = await runClaudePrompt(testHandle(bin), {
agent: "reviewer",
system: "SYS",
text: "hi",
title: "t",
onActivity: (line) => activity.push(line),
});
expect(result.text).toBe("the findings");
expect(activity).toEqual([
"started claude-opus-5",
"Read src/core/auth.ts",
"completed (3s, 2 turn(s))",
]);
} finally {
await cleanup();
}
});
test("claudeCodePromptAndParse: a parseable first reply returns immediately with truncated:false", async () => {
const body = JSON.stringify({
type: "result",
@@ -773,6 +1126,146 @@ test("claudeCodePromptAndParse: a parseable first reply returns immediately with
}
});
test("promptAndParse: a Claude-routed production parser forwards its JSON Schema", async () => {
const body = JSON.stringify({
type: "result",
subtype: "success",
is_error: false,
result: '{"findings":[]}',
structured_output: { findings: [] },
});
const { bin, argsLog, cleanup } = await sequencedFakeClaude([body]);
try {
const output = await promptAndParse(
testHandle(bin) as unknown as OpencodeHandle,
{ agent: "reviewer", system: "SYS", text: "hi", title: "t" },
parseReviewerOutput,
);
expect(output.value).toEqual({ findings: [] });
const args = (await argsLog(1)).split("\n");
const schemaIndex = args.indexOf("--json-schema");
expect(schemaIndex).toBeGreaterThan(-1);
const schema = JSON.parse(args[schemaIndex + 1]!) as {
properties: { findings: { items: { required: string[] } } };
};
expect(schema.properties.findings.items.required).toContain("title");
} finally {
await cleanup();
}
});
test("promptAndParse: Claude structured-output exhaustion retries once and preserves usage", async () => {
const exhausted = JSON.stringify({
type: "result",
subtype: "error_max_structured_output_retries",
is_error: true,
result: "Could not produce valid structured output",
total_cost_usd: 0.01,
usage: { input_tokens: 5, output_tokens: 7 },
});
const recovered = JSON.stringify({
type: "result",
subtype: "success",
is_error: false,
result: "ignored in structured mode",
structured_output: { findings: [] },
total_cost_usd: 0.03,
usage: { input_tokens: 11, output_tokens: 13 },
});
const { bin, stdinLog, cleanup } = await sequencedFakeClaude([exhausted, recovered]);
try {
const activity: string[] = [];
const output = await promptAndParse(
testHandle(bin) as unknown as OpencodeHandle,
{
agent: "security",
system: "SYS",
text: "review",
title: "security review",
onActivity: (line) => activity.push(line),
},
parseReviewerOutput,
);
expect(output.value).toEqual({ findings: [] });
expect(output.cost).toBeCloseTo(0.04);
expect(output.tokens).toEqual({
input: 16,
output: 20,
reasoning: 0,
cache: { read: 0, write: 0 },
});
expect(activity).toContain("structured output validation failed — retrying once");
expect(await stdinLog(2)).toContain("IMPORTANT: reply with ONLY the single JSON object");
} finally {
await cleanup();
}
});
test("claudeCodePromptAndParse: schema mode never accepts result text without structured_output", async () => {
const unvalidated = JSON.stringify({
type: "result",
subtype: "success",
is_error: false,
result: JSON.stringify({ ok: "unvalidated" }),
});
const validated = JSON.stringify({
type: "result",
subtype: "success",
is_error: false,
result: JSON.stringify({ ok: "wrong-channel" }),
structured_output: { ok: "validated" },
});
const { bin, cleanup } = await sequencedFakeClaude([unvalidated, validated]);
try {
const output = await claudeCodePromptAndParse(
testHandle(bin),
{
agent: "reviewer",
system: "SYS",
text: "hi",
title: "t",
jsonSchema: {
type: "object",
properties: { ok: { type: "string" } },
required: ["ok"],
},
},
(text) => JSON.parse(text) as { ok: string },
);
expect(output.value).toEqual({ ok: "validated" });
} finally {
await cleanup();
}
});
test("claudeCodePromptAndParse: repeated structured-output exhaustion fails explicitly", async () => {
const exhausted = JSON.stringify({
type: "result",
subtype: "error_max_structured_output_retries",
is_error: true,
result: "Could not produce valid structured output",
});
const { bin, cleanup } = await sequencedFakeClaude([exhausted, exhausted]);
try {
await expect(
claudeCodePromptAndParse(
testHandle(bin),
{
agent: "security",
system: "SYS",
text: "review",
title: "security review",
jsonSchema: { type: "object", required: ["findings"] },
},
parseReviewerOutput,
),
).rejects.toThrow(/could not satisfy its required JSON Schema after retries/);
} finally {
await cleanup();
}
});
test("claudeCodePromptAndParse: an unparseable first reply retries with CLAUDE_CORRECTIVE appended, accumulating cost/tokens across both calls", async () => {
const bad = JSON.stringify({
type: "result",
+26
View File
@@ -21,6 +21,27 @@ test("run with input: feeds stdin through the spawn path", async () => {
expect(result.overflowed).toBeFalsy();
});
test("run with input: observes admitted stdout incrementally without changing capture", async () => {
const observed: string[] = [];
const result = await run("sh", ["-c", "printf first; sleep 0.05; printf second"], {
input: "",
onStdout: (chunk) => observed.push(chunk),
});
expect(result.stdout).toBe("firstsecond");
expect(observed.join("")).toBe(result.stdout);
});
test("run with input: a stdout observer failure never breaks the command", async () => {
const result = await run("printf", ["safe"], {
input: "",
onStdout: () => {
throw new Error("observer failed");
},
});
expect(result.stdout).toBe("safe");
expect(result.code).toBe(0);
});
test("run with input: timeout kills the child and resolves timedOut (no throw)", async () => {
const started = Date.now();
const result = await run("sleep", ["30"], { input: "", timeout: 250, check: true });
@@ -70,12 +91,17 @@ test("run without input: timeout resolves timedOut too (execFile path, same cont
});
test("run with input: maxBuffer overflow is flagged, not silently truncated", async () => {
let observed = "";
const result = await run("sh", ["-c", "head -c 100000 /dev/zero | tr '\\0' 'x'"], {
input: "",
maxBuffer: 1024,
check: false,
onStdout: (chunk) => {
observed += chunk;
},
});
expect(result.overflowed).toBe(true);
expect(observed.length).toBe(1024);
});
test("run with input: check=false returns a non-zero exit instead of throwing", async () => {
+23
View File
@@ -11,6 +11,7 @@ import {
formatUsageSummary,
renderUsageMarkdown,
effectiveConcurrency,
formatAgentActivity,
} from "../core/review.js";
import type { LoadedConfig } from "../config/schema.js";
import type { PatchWorkspaceFile } from "../core/noise.js";
@@ -51,6 +52,28 @@ test("chunkByLines: a single over-budget file is its own chunk", () => {
expect(chunks.map((c) => c.map((f) => f.path))).toEqual([["big"], ["small"]]);
});
test("formatAgentActivity: stable agent tag leads every line and pass detail stays visible", () => {
expect(formatAgentActivity("security", "security", "Read src/auth.ts")).toBe(
" [security] Read src/auth.ts",
);
expect(formatAgentActivity("correctness", "correctness [2/3]", "Grep src")).toBe(
" [correctness] [2/3]: Grep src",
);
expect(formatAgentActivity("cross-cutting", "cross-file", "still working…\n45s")).toBe(
" [cross-cutting] cross-file: still working… 45s",
);
expect(formatAgentActivity("correctness", "correctness [1/2]", "FAILED (bad output)")).toBe(
" [correctness] [1/2]: FAILED (bad output)",
);
expect(
formatAgentActivity(
"security",
"security [2/2]",
"exceeded 15m — retrying 3 files with a fast no-tools pass",
),
).toBe(" [security] [2/2]: exceeded 15m — retrying 3 files with a fast no-tools pass");
});
test("applyReviewPolicy: drops suggestions, sorts by severity", () => {
const out: CoordinatorOutput = {
decision: "request_changes",
+26
View File
@@ -5,6 +5,7 @@ import {
fingerprintFinding,
parseCoordinatorOutput,
parseReviewerOutput,
parseVerdict,
} from "../core/schema.js";
import type { Finding } from "../core/schema.js";
@@ -80,6 +81,31 @@ test("a forged `agent` in model output is dropped at the parse boundary", () =>
expect(attributed[0]!.agent).toBe("correctness");
});
test("structured parsers expose draft-07 contracts for provider-side validation", () => {
const reviewer = parseReviewerOutput.jsonSchema as {
$schema: string;
properties: {
findings: { items: { properties: Record<string, unknown>; required: string[] } };
};
};
expect(reviewer.$schema).toBe("http://json-schema.org/draft-07/schema#");
expect(reviewer.properties.findings.items.required).toContain("title");
expect(reviewer.properties.findings.items.properties.agent).toBeUndefined();
const coordinator = parseCoordinatorOutput.jsonSchema as {
properties: Record<string, unknown>;
required: string[];
};
expect(coordinator.required).toEqual(["decision", "findings", "summary"]);
expect(coordinator.properties.incomplete).toBeUndefined();
expect(coordinator.properties.couldNotComplete).toBeUndefined();
expect(parseVerdict.jsonSchema).toMatchObject({
type: "object",
required: ["verified", "reason"],
});
});
test("extractJsonObject: an empty response names the cause instead of 'undefined'", () => {
expect(() => extractJsonObject("")).toThrow(/an EMPTY response/);
expect(() => extractJsonObject("prose with no json at all")).toThrow(/no \{\.\.\.\}/);
+1
View File
@@ -244,6 +244,7 @@ export async function adjudicateFeedback(
title: `adjudicate-${index}`,
maxWaitMs: ADJUDICATE_TIMEOUT_MS,
finalizeOnTimeout: true,
onActivity: (line) => debug(` [adjudicator] #${index + 1}: ${line}`),
},
parseAdjudication,
);
+356 -54
View File
@@ -20,7 +20,7 @@ import { RateLimitWatch } from "./throttle.js";
/**
* The Claude Code CLI review engine. Runs each pass as a single `claude -p
* --output-format json` subprocess on the user's Claude Max/Team subscription,
* --output-format stream-json --verbose` subprocess on the user's Claude Max/Team subscription,
* bypassing OpenCode entirely (which cannot use Anthropic subscription OAuth).
* The seam functions in opencode.ts dispatch here on `handle.engine`.
*
@@ -59,6 +59,8 @@ export interface ClaudePromptArgs {
maxWaitMs?: number;
maxToolCalls?: number;
finalizeOnTimeout?: boolean;
/** Provider-enforced output contract, generated from the local Zod parser. */
jsonSchema?: Record<string, unknown>;
}
/** Coarse per-pass wander bound; the review's own maxWaitMs is the real ceiling. */
@@ -66,9 +68,9 @@ const CLAUDE_MAX_TURNS = 60;
/** Fallback per-pass ceiling when a caller passes no maxWaitMs. */
const DEFAULT_MAX_WAIT_MS = 8 * 60 * 1000;
/**
* A stateless `claude -p` pass emits no incremental tool lines to the caller, so a
* long pass would look hung. Emit a "still working" heartbeat this often (matching
* opencode.ts's HEARTBEAT_MS) to keep the progress signal alive.
* Emit a "still working" heartbeat after this long with no structured stream
* activity (matching opencode.ts's HEARTBEAT_MS), so a thinking stretch still has
* a progress signal between tool calls.
*/
const CLAUDE_HEARTBEAT_MS = 45_000;
@@ -289,6 +291,7 @@ export function buildClaudeArgs(opts: {
*/
tools?: readonly string[];
maxTurns?: number;
jsonSchema?: Record<string, unknown>;
}): string[] {
// Permission rules are gitignore-style with forward slashes; a Windows cwd
// (`C:\Users\dev\repo`) must be normalized or every rule silently matches
@@ -305,11 +308,13 @@ export function buildClaudeArgs(opts: {
return [
"-p",
"--output-format",
"json",
"stream-json",
"--verbose",
"--model",
opts.model,
"--append-system-prompt",
opts.system,
...(opts.jsonSchema ? ["--json-schema", JSON.stringify(opts.jsonSchema)] : []),
...(enabled.length > 0 ? ["--allowedTools", ...enabled.map(scope)] : []),
"--disallowedTools",
...deniedReadTools,
@@ -331,6 +336,184 @@ export interface ClaudeResult {
modelOutputTokens: Record<string, number>;
isError: boolean;
errorText: string;
/** Whether the final result carried the object validated by `--json-schema`. */
hasStructuredOutput: boolean;
/** Claude Code exhausted its own in-session structured-output repair attempts. */
structuredOutputFailure: boolean;
}
const CLAUDE_STREAM_MISSING_RESULT = "Claude Code stream ended without a final result event";
const CLAUDE_RESULT_MISSING_ERROR = "Claude Code returned an error without a message";
type JsonRecord = Record<string, unknown>;
function jsonRecord(value: unknown): JsonRecord | null {
return value !== null && typeof value === "object" && !Array.isArray(value)
? (value as JsonRecord)
: null;
}
/** The final result object from either legacy single JSON or JSONL stream output. */
function finalClaudeResult(stdout: string): JsonRecord | null {
try {
const parsed = jsonRecord(JSON.parse(stdout));
return parsed?.type === "result" ? parsed : null;
} catch {
let result: JsonRecord | null = null;
for (const line of stdout.split(/\r?\n/)) {
if (!line.trim()) {
continue;
}
try {
const event = jsonRecord(JSON.parse(line));
if (event?.type === "result") {
result = event;
}
} catch {
// A malformed/non-JSON line cannot be the structured final result.
}
}
return result;
}
}
const MAX_ACTIVITY_DETAIL = 180;
/** Collapse control/newline injection and bound provider/model-originated log text. */
function safeActivityDetail(value: unknown): string | undefined {
if (typeof value !== "string") {
return undefined;
}
const withoutControls = [...value]
.map((char) => {
const code = char.charCodeAt(0);
return code <= 0x1f || (code >= 0x7f && code <= 0x9f) ? " " : char;
})
.join("");
const clean = withoutControls.replace(/\s+/g, " ").trim();
if (!clean) {
return undefined;
}
return clean.length > MAX_ACTIVITY_DETAIL ? `${clean.slice(0, MAX_ACTIVITY_DETAIL - 1)}…` : clean;
}
/** Render an in-tree tool target without exposing attempted host paths. */
function activityPath(value: unknown, cwd: string): string | undefined {
const candidate = safeActivityDetail(value);
if (!candidate) {
return undefined;
}
const root = path.resolve(cwd);
const absolute = path.resolve(root, candidate);
if (absolute !== root && !pathInside(absolute, root)) {
return undefined;
}
const relative = path.relative(root, absolute).replace(/\\/g, "/");
return relative || ".";
}
interface ClaudeActivity {
/** Stable tool-use id; repeated assistant events with the same id log once. */
key?: string;
line: string;
}
/**
* Convert one Claude stream event into safe progress metadata. Raw assistant text,
* tool results, and grep patterns are deliberately never logged: PR/model content is
* untrusted and may contain secrets or terminal-control/log-injection payloads.
*/
// @ref LLP 0003#claude-code-cli-containment [implements] — stream only bounded lifecycle/tool metadata; raw model text and tool results never become progress logs
export function claudeActivities(eventValue: unknown, cwd: string): ClaudeActivity[] {
const event = jsonRecord(eventValue);
if (!event) {
return [];
}
if (event.type === "system" && event.subtype === "init") {
const model = safeActivityDetail(event.model);
return [{ line: model ? `started ${model}` : "started" }];
}
if (event.type === "result") {
if (event.is_error === true) {
return [];
}
const duration =
typeof event.duration_ms === "number"
? `${Math.max(0, Math.round(event.duration_ms / 1000))}s`
: null;
const turns = typeof event.num_turns === "number" ? `${event.num_turns} turn(s)` : null;
const detail = [duration, turns].filter(Boolean).join(", ");
return [{ line: detail ? `completed (${detail})` : "completed" }];
}
if (event.type !== "assistant") {
return [];
}
const message = jsonRecord(event.message);
const content = Array.isArray(message?.content) ? message.content : [];
const activities: ClaudeActivity[] = [];
for (const value of content) {
const block = jsonRecord(value);
if (block?.type !== "tool_use" || typeof block.name !== "string") {
continue;
}
const input = jsonRecord(block.input) ?? {};
const key = typeof block.id === "string" ? block.id : undefined;
if (block.name === "Read") {
const target = activityPath(input.file_path, cwd);
activities.push({ key, line: target ? `Read ${target}` : "Read" });
} else if (block.name === "Grep") {
const target = activityPath(input.path, cwd);
activities.push({ key, line: target ? `Grep ${target}` : "Grep" });
} else if (block.name === "Glob") {
const target = activityPath(input.path, cwd);
activities.push({ key, line: target && target !== "." ? `Glob ${target}` : "Glob" });
}
}
return activities;
}
/** Incremental JSONL decoder for Claude's stream-json stdout. */
export function createClaudeActivityStream(
cwd: string,
onActivity: (line: string) => void,
): { push: (chunk: string) => void; finish: () => void } {
let buffered = "";
const reported = new Set<string>();
const consume = (line: string): void => {
if (!line.trim()) {
return;
}
try {
for (const activity of claudeActivities(JSON.parse(line), cwd)) {
if (activity.key && reported.has(activity.key)) {
continue;
}
if (activity.key) {
reported.add(activity.key);
}
onActivity(activity.line);
}
} catch {
// Ignore malformed progress events; final result parsing still reports errors.
}
};
return {
push(chunk: string): void {
buffered += chunk;
for (;;) {
const newline = buffered.indexOf("\n");
if (newline < 0) {
break;
}
consume(buffered.slice(0, newline).replace(/\r$/, ""));
buffered = buffered.slice(newline + 1);
}
},
finish(): void {
consume(buffered);
buffered = "";
},
};
}
/**
@@ -353,21 +536,26 @@ export function pickAnsweringModel(
}
/**
* Parse the `--output-format json` result object. Keys off `is_error` / a parse
* Parse the final result from `--output-format stream-json` JSONL (also accepts the
* former single JSON object for compatibility/tests). Keys off `is_error` / a parse
* failure, NOT `subtype` — `subtype` stays `"success"` on some API errors.
*/
export function parseClaudeResult(stdout: string): ClaudeResult {
let parsed: Record<string, unknown>;
try {
parsed = JSON.parse(stdout) as Record<string, unknown>;
} catch {
const parsed = finalClaudeResult(stdout);
if (!parsed) {
// The stream may contain assistant text and tool results sourced from the
// untrusted review tree. Never turn that transcript into an error message or
// feed it to provider-error classification.
// @ref LLP 0003#claude-code-cli-containment [constrained-by] — raw JSONL transcript content must never reach logs or error classifiers
return {
text: "",
cost: 0,
tokens: {},
modelOutputTokens: {},
isError: true,
errorText: stdout.trim(),
errorText: CLAUDE_STREAM_MISSING_RESULT,
hasStructuredOutput: false,
structuredOutputFailure: false,
};
}
const usage = (parsed.usage ?? {}) as Record<string, unknown>;
@@ -386,18 +574,42 @@ export function parseClaudeResult(stdout: string): ClaudeResult {
for (const [key, value] of Object.entries(modelUsage)) {
modelOutputTokens[key] = num((value as Record<string, unknown> | null)?.outputTokens) ?? 0;
}
const result = typeof parsed.result === "string" ? parsed.result : "";
const isError = parsed.is_error === true;
const hasStructuredOutput = parsed.structured_output !== undefined;
// `--json-schema` returns a provider-validated object in `structured_output`.
// Serialize it back through the existing local parser so Zod remains the final
// trust boundary. Ignore any stale/retracted structured value on an error result;
// error classification must use only the CLI's explicit final error message.
const structured =
isError || parsed.structured_output === undefined
? undefined
: JSON.stringify(parsed.structured_output);
const result = structured ?? (typeof parsed.result === "string" ? parsed.result : "");
return {
text: result,
cost: num(parsed.total_cost_usd) ?? 0,
tokens,
modelOutputTokens,
isError,
errorText: isError ? result || stdout.trim() : "",
// Only the final result event's explicit error text is safe to classify and
// surface. Falling back to stdout would expose the full JSONL transcript.
errorText: isError ? result || CLAUDE_RESULT_MISSING_ERROR : "",
hasStructuredOutput,
structuredOutputFailure: parsed.subtype === "error_max_structured_output_retries",
};
}
/**
* A provider-side schema failure that spent tokens and may be retried once from a
* clean process. Carrying the attempt lets the caller retain honest run metrics.
*/
class ClaudeStructuredOutputError extends Error {
constructor(readonly result: PromptResult) {
super("Claude Code could not produce output matching the required JSON Schema");
this.name = "ClaudeStructuredOutputError";
}
}
/** Classify a Claude Code failure so the caller can pick backoff vs. hard fail. */
export function classifyClaudeError(
errorText: string,
@@ -488,6 +700,40 @@ export function claudeTemperatureNote(
: null;
}
/**
* Preserve actionable process diagnostics without copying arbitrary stderr into a
* public Actions log. A non-empty stream may already contain model/tool content, so
* stderr is not even classified in that case. With no stream, recognize only fixed
* CLI/setup categories and never interpolate the matched text.
*/
// @ref LLP 0003#claude-code-cli-containment [constrained-by] — stderr may reflect untrusted tree content; expose only exit metadata and fixed allowlisted categories
function claudeExitDiagnostic(result: { stdout: string; stderr: string; code: number }): string {
const exit = `Claude Code exited with code ${result.code}`;
if (result.stdout.trim() !== "") {
return exit;
}
if (/unknown (?:option|argument)|unrecognized option|unexpected argument/i.test(result.stderr)) {
return (
`${exit}; the CLI rejected its arguments — verify the pinned ` +
"@anthropic-ai/claude-code version supports the configured flags"
);
}
if (/authentication|oauth|api.?key|unauthorized|\b401\b|\b403\b/i.test(result.stderr)) {
return (
`${exit}; authentication failed before a result was emitted — check ` +
"`claude auth status` and `ecr doctor`"
);
}
if (/\bENOENT\b|command not found|no such file or directory/i.test(result.stderr)) {
return `${exit}; the Claude Code executable or one of its required files was not found`;
}
if (/\bEACCES\b/i.test(result.stderr)) {
return `${exit}; the Claude Code executable could not be launched due to permissions`;
}
return exit;
}
/**
* One prompt → text/cost/tokens/model, as a single `claude -p` subprocess (the
* Claude analogue of OpenCode's promptAgent; no sessions/polling).
@@ -507,14 +753,25 @@ export async function runClaudePrompt(
// stateless analogue of OpenCode's mid-run tool-call cap).
const maxTurns =
args.maxToolCalls != null && args.maxToolCalls > 0 ? args.maxToolCalls : undefined;
// A stateless `claude -p` pass streams nothing back, so emit a heartbeat while it
// runs or a long pass looks hung (cleared in finally, whatever the outcome).
// Stream safe structured activity. The heartbeat fires only after a quiet window,
// rather than on a fixed cadence that can land immediately after a tool line.
const heartbeatStart = Date.now();
let lastStreamActivityAt = heartbeatStart;
const emitActivity = (line: string): void => {
lastStreamActivityAt = Date.now();
args.onActivity?.(line);
};
const activityStream = args.onActivity
? createClaudeActivityStream(process.cwd(), emitActivity)
: undefined;
const heartbeat = args.onActivity
? setInterval(() => {
args.onActivity?.(
`still working… ${Math.round((Date.now() - heartbeatStart) / 1000)}s elapsed`,
);
if (Date.now() - lastStreamActivityAt >= CLAUDE_HEARTBEAT_MS) {
args.onActivity?.(
`still working… ${Math.round((Date.now() - heartbeatStart) / 1000)}s elapsed ` +
`(no new activity for ${Math.round((Date.now() - lastStreamActivityAt) / 1000)}s)`,
);
}
}, CLAUDE_HEARTBEAT_MS)
: undefined;
heartbeat?.unref?.();
@@ -528,6 +785,7 @@ export async function runClaudePrompt(
cwd: process.cwd(),
tools,
maxTurns,
jsonSchema: args.jsonSchema,
}),
{
input: args.text,
@@ -535,9 +793,11 @@ export async function runClaudePrompt(
cwd: process.cwd(),
timeout: maxWaitMs,
check: false,
onStdout: activityStream?.push,
},
);
} finally {
activityStream?.finish();
if (heartbeat) {
clearInterval(heartbeat);
}
@@ -551,9 +811,7 @@ export async function runClaudePrompt(
// A non-timeout signal is a crash (SIGSEGV, OOM SIGKILL, external kill), not a
// timeout — surface it as a hard error rather than the subdivide/retry path.
if (result.signal) {
throw new Error(
`Claude Code was killed by signal ${result.signal}: ${result.stderr.trim() || "(no output)"}`,
);
throw new Error(`Claude Code was killed by signal ${result.signal}`);
}
// Truncated output can't be parsed as JSON; report the cause plainly instead of
// letting it fall through as a generic parse failure.
@@ -562,6 +820,31 @@ export async function runClaudePrompt(
}
const parsed = parseClaudeResult(result.stdout);
const answered = pickAnsweringModel(configuredModel, parsed.modelOutputTokens);
const model = answered
? claudeModelMatches(configuredModel, answered)
? configuredModel
: `anthropic/${answered}`
: configuredModel;
const promptResult: PromptResult = {
text: parsed.text,
cost: parsed.cost,
sessionID: "",
tokens: parsed.tokens,
model,
};
// A schema-requesting caller must receive the provider-validated object, never a
// parseable-looking fallback from `result`. Claude repairs schema mismatches in
// session first; its documented exhaustion result gets one clean-process retry in
// claudeCodePromptAndParse. A success that omits structured_output is treated the
// same way because it did not honor the requested provider contract.
if (
args.jsonSchema &&
(parsed.structuredOutputFailure || (!parsed.isError && !parsed.hasStructuredOutput))
) {
throw new ClaudeStructuredOutputError(promptResult);
}
if (parsed.isError) {
const kind = classifyClaudeError(parsed.errorText);
if (kind === "rate-limit" || kind === "usage-limit") {
@@ -581,37 +864,33 @@ export async function runClaudePrompt(
"`ecr doctor`.",
);
}
const needsProcessDiagnostic =
parsed.errorText === CLAUDE_STREAM_MISSING_RESULT ||
parsed.errorText === CLAUDE_RESULT_MISSING_ERROR;
throw new Error(
parsed.errorText ||
`claude exited with code ${result.code}: ${
result.stderr.trim() || result.stdout.trim() || "(no output)"
}`,
needsProcessDiagnostic
? `${parsed.errorText} (${claudeExitDiagnostic(result)})`
: parsed.errorText,
);
}
const answered = pickAnsweringModel(configuredModel, parsed.modelOutputTokens);
const model = answered
? claudeModelMatches(configuredModel, answered)
? configuredModel
: `anthropic/${answered}`
: configuredModel;
return { text: parsed.text, cost: parsed.cost, sessionID: "", tokens: parsed.tokens, model };
return promptResult;
}
/**
* The claude analogue of opencode.ts's CORRECTIVE. NOT shared: that one says
* "your previous reply could not be parsed", which is true in OpenCode's
* same-session follow-up but false here — each `claude -p` invocation is a fresh
* stateless process with no previous reply to reference.
* Last-resort correction for a no-schema caller or disagreement between the
* provider's JSON Schema validator and our local parser. Production parsers pass
* `--json-schema`, so Claude Code already performs validation-aware repair inside
* the original session before this fresh-process fallback is needed.
*/
const CLAUDE_CORRECTIVE =
"\n\nIMPORTANT: reply with ONLY the single JSON object described above — no prose, " +
"no code fences, no partial output.";
/**
* Prompt via the Claude Code CLI and parse the reply, mirroring OpenCode's
* promptAndParse: transient retry on the first call, then one corrective re-run
* (a fresh process — the diff is inlined, so re-read is a cache hit).
* Prompt via the Claude Code CLI and parse the provider-validated structured
* result locally. Transient failures retry first; a remaining local parse failure
* gets one fresh-process corrective as defense in depth.
*/
export async function claudeCodePromptAndParse<T>(
handle: ClaudeCodeHandle,
@@ -627,24 +906,47 @@ export async function claudeCodePromptAndParse<T>(
model = result.model ?? model;
};
const first = await withTransientRetry(`Agent "${args.agent}"`, args.onActivity, () =>
runClaudePrompt(handle, args),
);
record(first);
let first: PromptResult | undefined;
try {
return { value: parse(first.text), cost, truncated: false, tokens, model };
} catch {
const retry = await runClaudePrompt(handle, { ...args, text: args.text + CLAUDE_CORRECTIVE });
record(retry);
first = await withTransientRetry(`Agent "${args.agent}"`, args.onActivity, () =>
runClaudePrompt(handle, args),
);
record(first);
try {
return { value: parse(retry.text), cost, truncated: false, tokens, model };
} catch (finalError) {
throw new Error(
`Agent "${args.agent}" did not return parseable JSON after retries: ${
finalError instanceof Error ? finalError.message : String(finalError)
}`,
);
return { value: parse(first.text), cost, truncated: false, tokens, model };
} catch {
// Provider validation and local Zod validation disagreed. Retry once from a
// clean process rather than accepting an object the trust boundary rejected.
}
} catch (error) {
if (!(error instanceof ClaudeStructuredOutputError)) {
throw error;
}
record(error.result);
args.onActivity?.("structured output validation failed — retrying once");
}
let retry: PromptResult;
try {
retry = await runClaudePrompt(handle, { ...args, text: args.text + CLAUDE_CORRECTIVE });
} catch (error) {
if (!(error instanceof ClaudeStructuredOutputError)) {
throw error;
}
record(error.result);
throw new Error(
`Agent "${args.agent}" could not satisfy its required JSON Schema after retries`,
);
}
record(retry);
try {
return { value: parse(retry.text), cost, truncated: false, tokens, model };
} catch (finalError) {
throw new Error(
`Agent "${args.agent}" did not return parseable JSON after retries: ${
finalError instanceof Error ? finalError.message : String(finalError)
}`,
);
}
}
+2
View File
@@ -35,6 +35,7 @@ export async function coordinate(
agentFindings: Record<string, Finding[]>,
coverageNotes: string[] = [],
stackManifest?: StackManifest | null,
onActivity?: (line: string) => void,
): Promise<CoordinationResult> {
const system = buildCoordinatorSystem(config);
const text = buildCoordinatorTask(metadata, agentFindings, coverageNotes, stackManifest);
@@ -47,6 +48,7 @@ export async function coordinate(
title: "review-coordinator",
maxWaitMs: COORDINATOR_TIMEOUT_MS,
finalizeOnTimeout: true,
onActivity,
},
parseCoordinatorOutput,
);
+15
View File
@@ -31,6 +31,12 @@ export interface RunOptions {
killSignal?: NodeJS.Signals;
/** When set, write this to the child's stdin (routes through spawn, not execFile). */
input?: string;
/**
* Best-effort observer for captured stdout chunks on the spawn/input path. Only
* bytes admitted by maxBuffer are observed; callback failures never affect the
* child process. Intended for structured incremental output, not raw log piping.
*/
onStdout?: (chunk: string) => void;
}
/**
@@ -205,7 +211,16 @@ function runWithInput(
child.stdout.setEncoding("utf8");
child.stderr.setEncoding("utf8");
child.stdout.on("data", (chunk: string) => {
const before = stdout.length;
stdout = cap(stdout, chunk);
const admitted = stdout.length - before;
if (admitted > 0 && options.onStdout) {
try {
options.onStdout(chunk.slice(0, admitted));
} catch {
// Observability must never break the command whose output it observes.
}
}
});
child.stderr.on("data", (chunk: string) => {
stderr = cap(stderr, chunk);
+13 -10
View File
@@ -1010,14 +1010,13 @@ export async function withTransientRetry<T>(
}
/**
* Prompt an agent and parse its reply. On a JSON-parse failure, first retry in
* the SAME session: the model still holds all the file context it read, so the
* corrective is a cache read and a cheap re-emit — and better for recall than
* re-investigating from scratch (the usual failure is a truncated/malformed reply
* after a sound investigation). Only if that also fails do we fall back to a fresh
* session as a clean-slate last resort. A timeout is NOT a parse failure:
* promptAgent throws AgentTimeoutError, which propagates so the caller abandons
* the task instead of retrying a non-convergent run.
* Prompt an agent and parse its reply. Claude-routed production parsers hand their
* JSON Schema to the CLI, which repairs validation failures in-session before
* returning. OpenCode retries a JSON-parse failure in the SAME session: the model
* still holds all the file context it read, so the corrective is a cheap cache-read
* re-emit with better recall than re-investigating. Only if that also fails does it
* use a fresh session. A timeout is NOT a parse failure: AgentTimeoutError propagates
* so the caller abandons the non-convergent task.
*/
export async function promptAndParse<T>(
handle: OpencodeHandle,
@@ -1031,12 +1030,16 @@ export async function promptAndParse<T>(
maxToolCalls?: number;
finalizeOnTimeout?: boolean;
},
parse: (text: string) => T,
parse: ((text: string) => T) & { jsonSchema?: Record<string, unknown> },
): Promise<{ value: T; cost: number; truncated: boolean; tokens: TokenUsage; model?: string }> {
const dispatch = resolveEngineDispatch(handle, args.agent);
if (dispatch.engine === CLAUDE_CODE_ENGINE) {
const { claudeCodePromptAndParse } = await import("./claude-code.js");
return claudeCodePromptAndParse(dispatch.claudeHandle, args, parse);
return claudeCodePromptAndParse(
dispatch.claudeHandle,
{ ...args, jsonSchema: parse.jsonSchema },
parse,
);
}
let cost = 0;
let truncated = false;
+43 -16
View File
@@ -181,6 +181,14 @@ export function effectiveConcurrency(
return 6;
}
/** Prefix live model activity with its stable agent bucket and optional pass label. */
export function formatAgentActivity(agent: string, label: string, line: string): string {
const pass =
label === agent ? "" : label.startsWith(`${agent} `) ? label.slice(agent.length).trim() : label;
const activity = line.replace(/[\r\n]+/g, " ").trim();
return ` [${agent}] ${pass ? `${pass}: ` : ""}${activity}`;
}
export async function runReview(
source: ReviewSource,
options: ReviewRunOptions,
@@ -463,7 +471,9 @@ export async function runReview(
let selectedAgents = explicitAgents ?? config.agents;
if (!explicitAgents && options.route) {
progress("Routing: selecting relevant agents…");
const routed = await routeAgents(handle, config, workspace.files);
const routed = await routeAgents(handle, config, workspace.files, (line) =>
progress(formatAgentActivity("router", "router", line)),
);
selectedAgents = routed.agents;
progress(
routed.routed
@@ -673,6 +683,8 @@ export async function runReview(
let completedPasses = 0;
let failedPasses = 0;
const taskProgress = (task: ReviewTask, line: string): void =>
progress(formatAgentActivity(task.bucket, task.label, line));
// promptAndParse already retries internally (same-session corrective, then a
// bounded fresh session). We do NOT wrap it in another retry loop. On a genuine
// TIMEOUT, instead of dropping the work we break it into units that converge:
@@ -688,7 +700,7 @@ export async function runReview(
system: task.system,
text: buildTaskText(task),
title: task.title,
onActivity: (line) => progress(` ${task.label}: ${line}`),
onActivity: (line) => taskProgress(task, line),
maxWaitMs: task.maxWaitMs,
maxToolCalls: task.maxToolCalls,
finalizeOnTimeout: true,
@@ -707,7 +719,7 @@ export async function runReview(
}
completedPasses++;
if (truncated) {
progress(` ${task.label}: hit its budget — returned partial findings`);
taskProgress(task, "hit its budget — returned partial findings");
incomplete.push(
`${capitalize(task.coverageLabel)} ran out of time; its findings may be incomplete.`,
);
@@ -717,7 +729,7 @@ export async function runReview(
// Non-timeout errors are genuine failures — record and move on.
if (!(error instanceof AgentTimeoutError)) {
failedPasses++;
progress(` ${task.label}: FAILED (${errorMessage(error)})`);
taskProgress(task, `FAILED (${errorMessage(error)})`);
// An auth/permission failure hits every pass identically; push one shared,
// actionable note (deduped into a single coverage line) instead of N generic
// per-pass failures that bury the real, fixable cause.
@@ -745,8 +757,9 @@ export async function runReview(
const mid = Math.ceil(task.files.length / 2);
const left = task.files.slice(0, mid);
const right = task.files.slice(mid);
progress(
` ${task.label}: exceeded ${minutes}m — splitting into 2 smaller passes (${left.length} + ${right.length} files)`,
taskProgress(
task,
`exceeded ${minutes}m — splitting into 2 smaller passes (${left.length} + ${right.length} files)`,
);
const over: Partial<ReviewTask> = { depth: task.depth + 1, maxWaitMs: childCap };
enqueue(childTask(task, left, `↳${left.length}f`, over));
@@ -759,8 +772,9 @@ export async function runReview(
// without tools; it just can't open a caller outside the diff. A lighter
// cross-file review beats the coverage gap it used to report.
if (!task.fallback && remaining > FALLBACK_TIMEOUT_MS) {
progress(
` ${task.label}: exceeded ${minutes}m — retrying ${filesLabel(task.files)} with a fast no-tools pass`,
taskProgress(
task,
`exceeded ${minutes}m — retrying ${filesLabel(task.files)} with a fast no-tools pass`,
);
enqueue(
childTask(task, task.files, "(no-tools fallback)", {
@@ -782,8 +796,9 @@ export async function runReview(
const couldStillReduce =
(canSubdivide && task.depth < MAX_SUBDIVIDE_DEPTH) || !task.fallback;
if (error.reason === "stall") {
progress(
` ${task.label}: its model requests went silent (stalled) and did not recover — ` +
taskProgress(
task,
`its model requests went silent (stalled) and did not recover — ` +
`most likely provider rate limiting; reporting a coverage gap`,
);
// Name the likely cause. OpenCode retries a 429 internally without surfacing
@@ -797,15 +812,17 @@ export async function runReview(
`those changes were not fully reviewed.`,
);
} else if (couldStillReduce) {
progress(
` ${task.label}: exceeded ${minutes}m and the run's time budget is spent — reporting a coverage gap`,
taskProgress(
task,
`exceeded ${minutes}m and the run's time budget is spent — reporting a coverage gap`,
);
incomplete.push(
`${capitalize(task.coverageLabel)} timed out and the overall review budget was exhausted before it could be broken down further; those changes were not fully reviewed.`,
);
} else {
progress(
` ${task.label}: exceeded ${minutes}m even at its smallest reviewable unit — reporting a coverage gap`,
taskProgress(
task,
`exceeded ${minutes}m even at its smallest reviewable unit — reporting a coverage gap`,
);
incomplete.push(
`${capitalize(task.coverageLabel)} exceeded its time budget even after being reduced to its smallest reviewable unit; those changes were not fully reviewed.`,
@@ -883,7 +900,15 @@ export async function runReview(
tokens: coordinatorTokens,
truncated: coordinatorTruncated,
model: coordinatorModel,
} = await coordinate(handle, config, metadata, agentFindings, coverageNotes, stackManifest);
} = await coordinate(
handle,
config,
metadata,
agentFindings,
coverageNotes,
stackManifest,
(line) => progress(formatAgentActivity("coordinator", "coordinator", line)),
);
agentCosts["coordinator"] = cost;
trackTokens("coordinator", coordinatorTokens);
trackModel("coordinator", config.coordinator.model, coordinatorModel);
@@ -978,7 +1003,9 @@ export async function runReview(
const confirmation = await confirmStackRequalifications(
grounded,
options.stackConfirm.maxConfirmations,
patchConfirmer(handle, source),
patchConfirmer(handle, source, (line) =>
progress(formatAgentActivity(STACK_VERIFIER_AGENT, STACK_VERIFIER_AGENT, line)),
),
progress,
);
grounded = confirmation.findings;
+2
View File
@@ -21,6 +21,7 @@ export async function routeAgents(
handle: OpencodeHandle,
config: LoadedConfig,
files: PatchWorkspaceFile[],
onActivity?: (line: string) => void,
): Promise<RouteResult> {
const always = config.agents.filter((agent) => agent.alwaysRun);
try {
@@ -31,6 +32,7 @@ export async function routeAgents(
system: buildRouterSystem(),
text: buildRouterTask(config.agents, files),
title: "route",
onActivity,
},
parseRouteOutput,
);
+34 -27
View File
@@ -112,10 +112,6 @@ export const VerdictSchema = z.object({
});
export type Verdict = z.infer<typeof VerdictSchema>;
export function parseVerdict(text: string): Verdict {
return VerdictSchema.parse(extractJsonObject(text));
}
/**
* A stack verifier's verdict on whether a later stacked PR's patch actually
* addresses an absence-style finding (v2 patch confirmation). Fails toward
@@ -128,21 +124,16 @@ export const StackVerdictSchema = z.object({
});
export type StackVerdict = z.infer<typeof StackVerdictSchema>;
export function parseStackVerdict(text: string): StackVerdict {
return StackVerdictSchema.parse(extractJsonObject(text));
}
// @ref LLP 0011#attribution-and-identity [implements] — `agent` is engine-populated, so the model-facing schema drops it at the parse boundary instead of trusting call sites to strip it
/**
* The finding shape a MODEL may emit: `FindingSchema` minus the engine-only `agent`.
* Both model outputs parse through this, so an `agent` a reviewer pass or the
* coordinator invented is dropped where model JSON becomes typed data — the engine's own
* fingerprint lookup is then the only thing that can set it. The transform re-widens the
* result to `Finding` so downstream code (which does set `agent`) needs no change.
* fingerprint lookup is then the only thing that can set it. This schema deliberately
* has no Zod transform: the Claude runtime converts it to JSON Schema so the provider
* can enforce the same contract before the local parse boundary checks it again.
*/
const ModelFindingSchema = FindingSchema.omit({ agent: true }).transform(
(finding): Finding => finding,
);
const ModelFindingSchema = FindingSchema.omit({ agent: true });
/** Shape each sub-reviewer must emit. */
export const ReviewerOutputSchema = z.object({
@@ -151,10 +142,13 @@ export const ReviewerOutputSchema = z.object({
export type ReviewerOutput = z.infer<typeof ReviewerOutputSchema>;
/** Mode-agnostic coordinator result; each Reporter decides how to render it. */
export const CoordinatorOutputSchema = z.object({
const CoordinatorModelOutputSchema = z.object({
decision: z.enum(DECISIONS),
findings: z.array(ModelFindingSchema).default([]),
summary: z.string(),
});
export const CoordinatorOutputSchema = CoordinatorModelOutputSchema.extend({
/**
* Human-readable notes about reduced coverage (e.g. a review pass that hit its
* time limit and returned partial findings, or was skipped). Populated by the
@@ -358,10 +352,6 @@ export const AdjudicationSchema = z.object({
});
export type Adjudication = z.infer<typeof AdjudicationSchema>;
export function parseAdjudication(text: string): Adjudication {
return AdjudicationSchema.parse(extractJsonObject(text));
}
/** Minimum normalized evidence length to key a fingerprint on the code (below
* this we fall back to the title). */
const MIN_FP_EVIDENCE_LEN = 12;
@@ -442,14 +432,31 @@ export const RouteOutputSchema = z.object({
});
export type RouteOutput = z.infer<typeof RouteOutputSchema>;
export function parseRouteOutput(text: string): RouteOutput {
return RouteOutputSchema.parse(extractJsonObject(text));
type JsonSchema = Record<string, unknown>;
type StructuredParser<T> = ((text: string) => T) & { jsonSchema: JsonSchema };
/**
* Bind local Zod validation to the draft-07 JSON Schema Claude Code consumes.
* Local parsing remains authoritative; provider-side validation is a reliability
* layer that repairs malformed output before it reaches this trust boundary.
*/
// @ref LLP 0003#retry-taxonomy [implements] — Claude receives the same contract as the local parser and repairs mismatches in-session
function structuredParser<T>(
parseSchema: z.ZodType<T>,
outputSchema: z.ZodType = parseSchema,
): StructuredParser<T> {
const parser = ((text: string): T =>
parseSchema.parse(extractJsonObject(text))) as StructuredParser<T>;
parser.jsonSchema = z.toJSONSchema(outputSchema, { target: "draft-7" }) as JsonSchema;
return parser;
}
export function parseReviewerOutput(text: string): ReviewerOutput {
return ReviewerOutputSchema.parse(extractJsonObject(text));
}
export function parseCoordinatorOutput(text: string): CoordinatorOutput {
return CoordinatorOutputSchema.parse(extractJsonObject(text));
}
export const parseVerdict = structuredParser(VerdictSchema);
export const parseStackVerdict = structuredParser(StackVerdictSchema);
export const parseAdjudication = structuredParser(AdjudicationSchema);
export const parseRouteOutput = structuredParser(RouteOutputSchema);
export const parseReviewerOutput = structuredParser(ReviewerOutputSchema);
export const parseCoordinatorOutput = structuredParser(
CoordinatorOutputSchema,
CoordinatorModelOutputSchema,
);
+6 -1
View File
@@ -156,7 +156,11 @@ export async function confirmStackRequalifications(
* it into the no-tools stack verifier, and require `addressed: true`. The patch is
* inlined, never materialized — there is no disk read and no tool use at all.
*/
export function patchConfirmer(handle: OpencodeHandle, source: ReviewSource): ConfirmOne {
export function patchConfirmer(
handle: OpencodeHandle,
source: ReviewSource,
onActivity?: (line: string) => void,
): ConfirmOne {
// Per-run patch memo: distinct findings citing the same (prNumber, file) each get
// their own verdict but share one gh fetch. The source fails open to null (never
// rejects), so memoizing the promise is safe.
@@ -186,6 +190,7 @@ export function patchConfirmer(handle: OpencodeHandle, source: ReviewSource): Co
// A timeout here throws AgentTimeoutError, which the caller catches and STRIPS
// (fail toward blocking) — so no finalize salvage, unlike the main verifier.
finalizeOnTimeout: false,
onActivity: (line) => onActivity?.(`PR #${prNumber}: ${line}`),
},
parseStackVerdict,
);
+1
View File
@@ -159,6 +159,7 @@ export async function verifyFindings(
title: `verify-${index}`,
maxWaitMs: VERIFY_TIMEOUT_MS,
finalizeOnTimeout: true,
onActivity: (line) => onProgress?.(` [verifier] #${index + 1}: ${line}`),
},
parseVerdict,
);