diff --git a/apps/ai/scripts/pr-review-local.ts b/apps/ai/scripts/pr-review-local.ts index 15642a8ab2..198df9cf43 100644 --- a/apps/ai/scripts/pr-review-local.ts +++ b/apps/ai/scripts/pr-review-local.ts @@ -51,8 +51,10 @@ import type { ChatTurnEvent } from "@/chat/events" import { withToolTranscript } from "@/chat/close-out" import { PR_REVIEW_TOOLS } from "@/chat/permissions" import { PR_REVIEW_CLOSE_OUT_PROMPT } from "@/chat/prompts" +import { makeReviewCoverage } from "@/chat/review-coverage" +import { makeReviewLedger } from "@/chat/review-ledger" import { runChatTurn } from "@/chat/run" -import { makeRunUsage } from "@/chat/tools" +import { makeRunUsage, savedFindingsRequest } from "@/chat/tools" import type { McpToolExecutorApi } from "@/mcp/dispatcher" import { annotatePatch, @@ -897,6 +899,8 @@ export const reviewLocally = async ( } } + // Shared by the pass and its close-out, as in production. + const reviewState = { coverage: makeReviewCoverage(kickoff), ledger: makeReviewLedger() } const pass = (turn: { readonly text: string readonly history: ReadonlyArray @@ -916,6 +920,7 @@ export const reviewLocally = async ( submitted = request }), ...(turn.closeOut === true ? { closeOut: true } : undefined), + review: reviewState, ...(promptOverride === undefined ? undefined : { agent: { ...AGENTS["pr-review"], prompt: promptOverride } }), @@ -988,6 +993,16 @@ export const reviewLocally = async ( ) } + // As production does: findings the pass saved are filed as a partial when no report landed. + if (submitted === undefined && reviewState.ledger.findings().length > 0) { + submitted = savedFindingsRequest({ + findings: reviewState.ledger.findings(), + unreviewed: reviewState.coverage.unread(), + modelName: model.name, + usage, + }) + } + removeWorktrees() // Output diff --git a/apps/ai/src/chat/agents.ts b/apps/ai/src/chat/agents.ts index 2d9ce16d7e..447baddf32 100644 --- a/apps/ai/src/chat/agents.ts +++ b/apps/ai/src/chat/agents.ts @@ -152,6 +152,7 @@ export const agentPolicyFor = (agent: AgentDefinition, contextTokens?: number): // more without tools, to answer from what it found rather than stopping on a wall of tool // rows. This is also what makes `tokenBudget` a deadline rather than a way to lose a run. onExhaustion: "final-answer", + ...(budget.runStatus === true ? { runStatus: "appended" as const } : undefined), ...(contextTokens === undefined ? undefined : { contextTokenLimit: liveContextLimit(contextTokens) }), }) } diff --git a/apps/ai/src/chat/budgets.test.ts b/apps/ai/src/chat/budgets.test.ts index 1a1ba32d23..007746d05e 100644 --- a/apps/ai/src/chat/budgets.test.ts +++ b/apps/ai/src/chat/budgets.test.ts @@ -114,6 +114,12 @@ describe("agentPolicyFor", () => { ) }) + /** The review sees its clock and a warning at 80%; a chat turn keeps its cached prompt untouched. */ + it("shows the review its run status, and no other agent", () => { + expect(agentPolicyFor(AGENTS["pr-review"]!, GLM_CONTEXT).runStatus).toBe("appended") + expect(agentPolicyFor(AGENTS.default!, GLM_CONTEXT).runStatus).toBe("off") + }) + /** Crossing a budget must hand the run its closing call, never fail it outright. */ it("resolves an exhausted budget into a final answer", () => { expect(agentPolicyFor(AGENTS.investigate!, GLM_CONTEXT).onExhaustion).toBe("final-answer") diff --git a/apps/ai/src/chat/budgets.ts b/apps/ai/src/chat/budgets.ts index 0a817da888..b22cea7876 100644 --- a/apps/ai/src/chat/budgets.ts +++ b/apps/ai/src/chat/budgets.ts @@ -63,6 +63,12 @@ export interface AgentBudget { * which is smaller than any prompt this agent sends and would let research run to the last token. */ readonly completionReserveTokens: number + /** + * Append the engine's `` line (elapsed time, calls used, a warning at 80% of any + * limit) to each model call, so an unattended pass submits before its deadline rather than after. + * Only on the outgoing prompt, after the cached prefix. + */ + readonly runStatus?: boolean } /** @@ -91,6 +97,7 @@ export const PR_REVIEW_BUDGET: AgentBudget = { maxDuration: "10 minutes", tokenBudget: 64_000_000, completionReserveTokens: 48_000, + runStatus: true, } /** An answer on a pull request: narrower than a review, with room to read and to stage a fix. */ diff --git a/apps/ai/src/chat/close-out.test.ts b/apps/ai/src/chat/close-out.test.ts new file mode 100644 index 0000000000..596b046de4 --- /dev/null +++ b/apps/ai/src/chat/close-out.test.ts @@ -0,0 +1,44 @@ +/** + * The close-out's view of a failed pass: the evidence it gathered, bounded so the close-out fits. + */ +import { ChatMessage, ChatToolCall } from "@maple/domain/chat-session" +import { assert, describe, it } from "vitest" +import { CLOSE_OUT_EVIDENCE_CHARS, withToolTranscript } from "./close-out" + +const pass = (calls: number, outputChars: number) => + new ChatMessage({ + id: "pass", + role: "assistant", + text: "", + toolCalls: Array.from( + { length: calls }, + (_, i) => + new ChatToolCall({ + id: `c${i}`, + name: "pr_file_diff", + input: { paths: [`src/f${i}.ts`] }, + output: `${i}:${"x".repeat(outputChars)}`, + }), + ), + createdAt: 0, + startSeq: 1, + }) + +describe("withToolTranscript", () => { + it("renders every call of a small pass", () => { + const [message] = withToolTranscript([pass(3, 100)]) + assert.include(message!.text, "src/f0.ts") + assert.include(message!.text, "src/f2.ts") + assert.notInclude(message!.text, "left out for length") + }) + + /** A 149-call pass once rendered past the context limit, and the close-out died before its first call. */ + it("keeps a long pass under the budget, newest calls first, and counts what it left out", () => { + const [message] = withToolTranscript([pass(149, 19_000)]) + const text = message!.text + assert.isAtMost(text.length, CLOSE_OUT_EVIDENCE_CHARS + 1_000) + assert.include(text, "src/f148.ts") + assert.notInclude(text, "src/f0.ts") + assert.match(text, /\(\d+ earlier tool calls left out for length\.\)/) + }) +}) diff --git a/apps/ai/src/chat/close-out.ts b/apps/ai/src/chat/close-out.ts index 911125b476..0124275cdd 100644 --- a/apps/ai/src/chat/close-out.ts +++ b/apps/ai/src/chat/close-out.ts @@ -1,27 +1,60 @@ import { ChatMessage } from "@maple/domain/chat-session" /** How much of one tool's output the close-out turn is shown. */ -const CLOSE_OUT_TOOL_OUTPUT_CHARS = 60_000 +const CLOSE_OUT_TOOL_OUTPUT_CHARS = 20_000 + +/** + * All the evidence the close-out is shown, about 60k tokens. The newest message is replayed whole + * and compaction cannot split it, so an unbounded transcript fails the close-out before its first call. + */ +export const CLOSE_OUT_EVIDENCE_CHARS = 200_000 /** * The transcript as the close-out sees it: the same messages, with each assistant message's tool * calls and results rendered into its text. `promptFromHistory` replays prose only, and a pass * that gathered evidence through tools and wrote nothing would otherwise close out blind. + * + * Past the budget the oldest calls are dropped and counted, newest kept: they are where the pass + * was when it stopped. */ -export const withToolTranscript = (history: ReadonlyArray): ReadonlyArray => - history.map((message) => { - if (message.role !== "assistant" || message.toolCalls.length === 0) return message - const calls = message.toolCalls.map((call) => { - const output = call.output === undefined ? "(no result)" : renderToolValue(call.output) - return `[${call.name} ${renderToolValue(call.input)}]\n${output}` - }) +export const withToolTranscript = (history: ReadonlyArray): ReadonlyArray => { + let budget = CLOSE_OUT_EVIDENCE_CHARS + let full = false + const rendered = new Map; readonly omitted: number }>() + for (let index = history.length - 1; index >= 0; index--) { + const message = history[index]! + if (message.role !== "assistant" || message.toolCalls.length === 0) continue + const kept: Array = [] + for (let call = message.toolCalls.length - 1; call >= 0; call--) { + const { name, input, output } = message.toolCalls[call]! + const text = `[${name} ${renderToolValue(input)}]\n${output === undefined ? "(no result)" : renderToolValue(output)}` + if (full || text.length > budget) { + full = true + break + } + budget -= text.length + kept.unshift(text) + } + rendered.set(index, { calls: kept, omitted: message.toolCalls.length - kept.length }) + } + return history.map((message, index) => { + const evidence = rendered.get(index) + if (evidence === undefined) return message return new ChatMessage({ ...message, - text: [message.text, "Evidence gathered so far:", ...calls] + text: [ + message.text, + "Evidence gathered so far:", + ...(evidence.omitted > 0 + ? [`(${evidence.omitted} earlier tool calls left out for length.)`] + : []), + ...evidence.calls, + ] .filter((part) => part !== "") .join("\n\n"), }) }) +} const renderToolValue = (value: unknown): string => { const text = typeof value === "string" ? value : JSON.stringify(value) diff --git a/apps/ai/src/chat/prompts.ts b/apps/ai/src/chat/prompts.ts index e9b8c1a1d3..dedce5692e 100644 --- a/apps/ai/src/chat/prompts.ts +++ b/apps/ai/src/chat/prompts.ts @@ -259,7 +259,7 @@ A change is observable when the work it adds shows up in Maple with enough conte 1. Call pr_changed_files. Source, infra, config and test files are reviewed (tests for the tests category only); generated files, docs, tooling and lockfiles are not. A pull request with nothing left is verdict not_applicable: submit it straight away. Otherwise call pr_context once: never repeat what a comment already raised or what a failing check already reports. 2. The repository's rules (its CLAUDE.md, AGENTS.md and .maple/review.md, read at the base) are at the top of the kickoff when they could be read, or the kickoff says it has none: use them and never read those files again. Only when the kickoff says neither, read them yourself, once, before any hunk, at the BASE SHA (the base branch when the kickoff has no base SHA), never at the head: a pull request's edits to its own rules are part of the change under review, not rules for reviewing it. Use sandbox_read_file with that ref when the sandbox is available and read_source_file otherwise. When the diff adds production work, one sandbox_grep for the span helper and the SDK bootstrap (for example \`withSpan|startActiveSpan|Effect\\.fn|tracer|@opentelemetry|#\\[instrument\\]\`), narrowed to the part of the repository the diff touches. 3. Read the diffs with pr_file_diff, several files per call through \`paths\` (a small pull request fits in one or two calls). Line numbers in the output are on the NEW side of the diff; those are the only lines a finding may cite. -4. For each hunk, ask what can go wrong with it in production. When a suspected defect depends on code outside the diff (a caller, the type of a value, what a helper returns), read exactly that code with sandbox_grep or a narrow sandbox_read_file before you file it. A suspicion you could not confirm is not filed. +4. For each hunk, ask what can go wrong with it in production. When a suspected defect depends on code outside the diff (a caller, the type of a value, what a helper returns), read exactly that code with sandbox_grep or a narrow sandbox_read_file before you file it. A suspicion you could not confirm is not filed. A finding that survives that check is saved with record_finding right away, one call per finding, so it is posted even if the pass is stopped before step 6. 5. Try to break what the change claims. The description and commit messages state guarantees ("the result is unchanged", "non-ASCII words are skipped"); each is a claim to disprove, never a fact to repeat. Pick the inputs most likely to break it and follow them through the new code: - text: empty, one character, leading, trailing or repeated separators, and Unicode that case folding changes (the Kelvin sign \`\u212A\` lowercases to ASCII \`k\`; \`\u0130\`, \`\u00DF\`); - where a value lands: that language's metacharacters (\`%\` \`_\` \`\\\` in LIKE, regex specials, SQL quotes, shell words, \`..\` in a path), unescaped; @@ -267,6 +267,9 @@ A change is observable when the work it adds shows up in Maple with enough conte When the logic is self-contained (a parser, a regex, a split, arithmetic, a pure helper) and sandbox_exec is available, run it: one \`node -e\` or \`bun -e\` call on the head checkout with those inputs settles what reading can argue itself into or out of. 6. Call submit_review exactly once. +## Time +Every model call ends with a \`\` line giving the time used against your limit. When it carries a WARNING, stop reading: call submit_review with what you have established, and name what you did not reach in confidenceReason. + ## Large pull requests When pr_changed_files lists more than 12 files to review, do not read every diff yourself. After step 2, split the files into groups of related files (4 to 10 each, one area of the codebase per group) and call review_files once per group, all in the same message so they run in parallel. Pass the repository's rules that matter in \`focus\`. Each answers its group's findings one per line. File the ones you can stand behind: read the hunk behind any that looks doubtful, drop duplicates, and keep the discipline below. A group that ran out of budget (\`budgetExhausted\`) was reviewed in part; say so in the summary. @@ -288,6 +291,7 @@ When pr_changed_files lists more than 12 files to review, do not read every diff - A claim about how a library, database or runtime behaves (what a tokenizer splits on, what a function returns, what an operator matches) is evidence only when you ran it or read it in that project's own source or docs. Code on a branch the base does not contain is not the repository's history. - An observability finding carries the check id whose description above matches it. If none matches, it is not an observability finding. - A review of good code has no findings at all, and that is the result the author hopes for. +- record_finding cannot edit or remove a saved finding. Save each issue once, after checking it; never save it again in other words, and never repeat a saved finding in submit_review. - When the kickoff says the pull request was reviewed before, review the files it says changed since then, then judge every open finding it lists at this head, one by one. Read the code the finding describes, following it if it moved; lines being modified is not enough. Put a handle in \`resolved\` only when the code you read no longer has the defect; when unsure, leave it open. - A finding listed as open is never filed again, even where its code moved to other lines or you would word it differently: it keeps its handle. A different defect on nearby lines is not the same finding; file it. @@ -311,7 +315,7 @@ The review leads with one number, 1 to 5, how safe the change is to merge. It is - \`confidenceReason\` is one sentence, under 25 words, naming what decides the number or the one place that needs a careful look: "The new branch in \`listKeys\` changes tenant scoping and has no test". Never "Some risk remains". For a clean, contained change, say what makes it safe. ## Producing the review -Call \`submit_review\` once with: resolved (the handles of earlier findings this head fixes, when the kickoff listed any; a fix you did not read is not resolved), verdict (clean | issues | not_applicable), tests, risk, confidence (only to lower it), confidenceReason, summary, keyChanges, checked, coverage (observability only: one row per unit of production work the diff adds, with unit, kind, instrumented and evidence; empty when it adds none; build tooling, tests and scripts are not units), findings (path, line, endLine, category, checkId for observability, severity, title, body, suggestion, replacement). The review IS the submit_review call; prose instead of it is discarded. +Call \`submit_review\` once with: resolved (the handles of earlier findings this head fixes, when the kickoff listed any; a fix you did not read is not resolved), verdict (clean | issues | not_applicable), tests, risk, confidence (only to lower it), confidenceReason, summary, keyChanges, checked, coverage (observability only: one row per unit of production work the diff adds, with unit, kind, instrumented and evidence; empty when it adds none; build tooling, tests and scripts are not units), findings (path, line, endLine, category, checkId for observability, severity, title, body, suggestion, replacement; only the ones you did not already save, since saved findings are added for you). The review IS the submit_review call; prose instead of it is discarded. ## After the review If someone asks a follow-up in this session, answer with the same tools and the evidence you already gathered. @@ -347,4 +351,4 @@ Finish with submit_reply exactly once. Prose instead of it is discarded. export const PR_REPLY_CLOSE_OUT_PROMPT = `Your answer was not posted. Do not read more. Call \`submit_reply\` now with the answer you can support from what you already read, or a short note that you could not finish and what is missing. This is your only remaining action; prose is discarded.` /** The close-out for a review pass that stopped without filing: one call, from what it has. */ -export const PR_REVIEW_CLOSE_OUT_PROMPT = `Your review pass has ended without a submitted review. Do not read more. Call \`submit_review\` now with what you established: the verdict you can support, tests and risk as far as you read, a confidenceReason naming what you did not reach if any reviewed file's diff went unread, a summary of the change itself (the early end is shown for you, so do not repeat it) and keyChanges from what you read, the coverage rows you completed, and only the findings you anchored to a line. If you read nothing, submit verdict not_applicable with a summary saying the review could not be completed. This is your only remaining action; prose is discarded.` +export const PR_REVIEW_CLOSE_OUT_PROMPT = `Your review pass has ended without a submitted review. Do not read more. Call \`submit_review\` now with what you established: the verdict you can support, tests and risk as far as you read, a confidenceReason naming what you did not reach if any reviewed file's diff went unread, a summary of the change itself (the early end is shown for you, so do not repeat it) and keyChanges from what you read, the coverage rows you completed, and only the findings you anchored to a line that you have not already saved with record_finding (saved ones are included for you). If you read nothing, submit verdict not_applicable with a summary saying the review could not be completed. This is your only remaining action; prose is discarded.` diff --git a/apps/ai/src/chat/review-coverage.test.ts b/apps/ai/src/chat/review-coverage.test.ts index fbc5eb775a..d09ec63b35 100644 --- a/apps/ai/src/chat/review-coverage.test.ts +++ b/apps/ai/src/chat/review-coverage.test.ts @@ -2,7 +2,7 @@ * What a review pass has read. Fed the real renderers' answers, since the tracker reads them back. */ import type { PullRequestFile } from "@maple/domain/http" -import { renderFollowUp } from "@maple/backend/services/pr-review/findings" +import { renderFollowUp, renderIgnoredPaths } from "@maple/backend/services/pr-review/findings" import { assert, describe, it } from "vitest" import { renderChangedFiles, renderFileDiffs } from "../mcp/tools/pull-request" import { makeReviewCoverage, unreadRefusal } from "./review-coverage" @@ -35,6 +35,14 @@ describe("makeReviewCoverage", () => { assert.deepEqual(coverage.unread(), []) }) + it("never owes a file the repository's settings ignore", () => { + const coverage = makeReviewCoverage( + `Review pull request #7.\n${renderIgnoredPaths(["src/b.ts", "*.test.ts"])}`, + ) + coverage.observe("pr_changed_files", listing) + assert.deepEqual(coverage.unread(), ["src/a.ts"]) + }) + it("counts only pr_file_diff answers as reads", () => { const coverage = makeReviewCoverage("Review pull request #7.") coverage.observe("pr_changed_files", listing) diff --git a/apps/ai/src/chat/review-coverage.ts b/apps/ai/src/chat/review-coverage.ts index bb0fb89008..3aaeef2a2e 100644 --- a/apps/ai/src/chat/review-coverage.ts +++ b/apps/ai/src/chat/review-coverage.ts @@ -5,7 +5,7 @@ * tool answer the run dispatches, including the ones its `review_files` children dispatch through * the same handlers. `submit_review` consults it once before recording. */ -import { followUpScope } from "@maple/backend/services/pr-review/findings" +import { followUpScope, ignoredPathsInKickoff, pathIgnored } from "@maple/backend/services/pr-review/findings" import { pathsInDiffAnswer, reviewablePathsInListing } from "../mcp/tools/pull-request" export interface ReviewCoverage { @@ -15,10 +15,15 @@ export interface ReviewCoverage { readonly unread: () => ReadonlyArray } -/** `kickoff` is the pass's first message: a later push's kickoff narrows which files are due. */ +/** + * `kickoff` is the pass's first message: a later push's kickoff narrows which files are due, and a + * path the repository's settings ignore is never due, since the kickoff tells the reviewer to skip it. + */ export const makeReviewCoverage = (kickoff: string): ReviewCoverage => { const scope = followUpScope(kickoff) - const inScope = scope === undefined ? () => true : (path: string) => scope.includes(path) + const ignored = ignoredPathsInKickoff(kickoff) + const inScope = (path: string) => + (scope === undefined || scope.includes(path)) && !pathIgnored(path, ignored) const due = new Set() const read = new Set() return { diff --git a/apps/ai/src/chat/review-failure.test.ts b/apps/ai/src/chat/review-failure.test.ts new file mode 100644 index 0000000000..ff4ba1dbc6 --- /dev/null +++ b/apps/ai/src/chat/review-failure.test.ts @@ -0,0 +1,53 @@ +/** + * The words a failed review's pull request is shown come from the run's typed cause, never its text. + */ +import { AgentPolicyError, ContextBudgetError, ModelProtocolError } from "@effect-agent/core/AgentError" +import { prReviewFailureReason } from "@maple/domain/http" +import { Cause } from "effect" +import { assert, describe, it } from "vitest" +import { reviewFailureError, reviewFailureReason } from "./review-failure" + +const policy = (limit: AgentPolicyError["limit"]) => + Cause.fail(AgentPolicyError.make({ limit, message: `${limit} exceeded` })) + +describe("reviewFailureReason", () => { + it("names the limit that stopped the run", () => { + assert.equal(reviewFailureReason(policy("duration")), "time_limit") + assert.equal(reviewFailureReason(policy("tool-calls")), "step_limit") + assert.equal(reviewFailureReason(policy("tokens")), "step_limit") + assert.equal(reviewFailureReason(policy("repeated-failures")), "stuck") + }) + + it("tells a context overflow from anything else", () => { + const overflow = Cause.fail( + ContextBudgetError.make({ + message: "too big", + estimatedTokens: 200_000, + targetTokens: 128_000, + completionReserveTokens: 0, + }), + ) + assert.equal(reviewFailureReason(overflow), "context_limit") + assert.equal(reviewFailureReason(Cause.die("boom")), "agent_error") + }) +}) + +describe("reviewFailureReason over a combined cause", () => { + it("reports the typed limit even when a generic failure comes first", () => { + const combined = Cause.combine(Cause.die("boom"), policy("duration")) + assert.equal(reviewFailureReason(combined), "time_limit") + }) + + it("reads the engine's event ceiling as a step limit", () => { + const events = Cause.fail( + ModelProtocolError.make({ message: "Run exceeded the 65536-event buffer limit" }), + ) + assert.equal(reviewFailureReason(events), "step_limit") + }) +}) + +describe("reviewFailureError", () => { + it("reads back as its reason", () => { + assert.equal(prReviewFailureReason(reviewFailureError("time_limit")), "time_limit") + }) +}) diff --git a/apps/ai/src/chat/review-failure.ts b/apps/ai/src/chat/review-failure.ts new file mode 100644 index 0000000000..cf267d27fe --- /dev/null +++ b/apps/ai/src/chat/review-failure.ts @@ -0,0 +1,55 @@ +/** + * Why a review pass stopped without a report, in the fixed vocabulary the pull request is shown. + * + * The raw cause stays in logs and spans; only the reason reaches GitHub, as the prefix of the error + * the row stores (`time_limit: ...`), which `PrReviewService` turns back into one sentence. + */ +import { + AgentPolicyError, + ContextBudgetError, + ContextOverflowError, + ModelProtocolError, +} from "@effect-agent/core/AgentError" +import { PR_REVIEW_FAILURE_COPY, type PrReviewFailureReason } from "@maple/domain/http" +import { Cause } from "effect" +import { isAiError } from "effect/unstable/ai/AiError" + +/** The reason one error names, or `undefined` for one that says nothing specific. */ +const reasonOf = (error: unknown): PrReviewFailureReason | undefined => { + if (error instanceof AgentPolicyError) { + switch (error.limit) { + case "duration": + return "time_limit" + case "repeated-failures": + return "stuck" + default: + return "step_limit" + } + } + if (error instanceof ContextBudgetError || error instanceof ContextOverflowError) return "context_limit" + // The engine's run-event and response-part ceilings: the run outgrew its room, like a step limit. + if (error instanceof ModelProtocolError && /exceeded the \d+-(event|part)/.test(error.message)) + return "step_limit" + if (isAiError(error)) return "model_error" + return undefined +} + +/** + * The reason a failed run's cause names. Every failure is read, not only the first, so a typed + * limit is reported even when a generic failure sits ahead of it; otherwise it is an `agent_error`. + */ +export const reviewFailureReason = (cause: Cause.Cause): PrReviewFailureReason => { + const errors = cause.reasons.flatMap((reason) => + Cause.isFailReason(reason) ? [reason.error] : Cause.isDieReason(reason) ? [reason.defect] : [], + ) + const reasons = errors.map(reasonOf) + return ( + reasons.find((reason) => reason !== undefined && reason !== "model_error") ?? + reasons.find((reason) => reason !== undefined) ?? + "agent_error" + ) +} + +/** The row's `error` for a reason: its code first, so it can be read back. */ +export const reviewFailureError = (reason: PrReviewFailureReason): string => + `${reason}: ${PR_REVIEW_FAILURE_COPY[reason]} Retry with @maple review.` diff --git a/apps/ai/src/chat/review-fanout.ts b/apps/ai/src/chat/review-fanout.ts index 8fe3220327..bf592f702a 100644 --- a/apps/ai/src/chat/review-fanout.ts +++ b/apps/ai/src/chat/review-fanout.ts @@ -62,6 +62,8 @@ const workerPolicy = AgentPolicy.make({ completionReserveTokens: 16_000, toolConcurrency: 4, onExhaustion: "final-answer", + // A worker sees its own clock too, so it answers inside its four minutes. + runStatus: "appended", }) /** diff --git a/apps/ai/src/chat/review-ledger.ts b/apps/ai/src/chat/review-ledger.ts new file mode 100644 index 0000000000..11889f2236 --- /dev/null +++ b/apps/ai/src/chat/review-ledger.ts @@ -0,0 +1,30 @@ +/** + * The findings a review pass saved with `record_finding` as it established them. + * + * One per turn, shared by the pass and its close-out, so a finding saved before a deadline, a + * model error or a compaction is still posted: `submit_review` adds these to its own, and when no + * call to it lands the runner files them as a partial review itself. + */ +import { type PrReviewFinding, PR_REVIEW_MAX_FINDINGS } from "@maple/domain/http" +import { mergeSavedFindings } from "@maple/backend/services/pr-review/findings" + +export type RecordOutcome = "recorded" | "repeat" | "full" + +export interface ReviewLedger { + readonly record: (finding: PrReviewFinding) => RecordOutcome + readonly findings: () => ReadonlyArray +} + +export const makeReviewLedger = (): ReviewLedger => { + const saved: Array = [] + return { + record: (finding) => { + if (saved.length >= PR_REVIEW_MAX_FINDINGS) return "full" + // A restatement of a saved finding is the model re-recording after a compaction. + if (mergeSavedFindings(saved, [finding]).length === saved.length) return "repeat" + saved.push(finding) + return "recorded" + }, + findings: () => [...saved], + } +} diff --git a/apps/ai/src/chat/run.ts b/apps/ai/src/chat/run.ts index b2efbae434..7204f71668 100644 --- a/apps/ai/src/chat/run.ts +++ b/apps/ai/src/chat/run.ts @@ -21,7 +21,8 @@ import type { TenantContext } from "@maple/backend/services/auth/tenant-context" import { type AgentDefinition, agentForSession, chatAgent } from "./agents" import { profileForTurn } from "./profiles" import { makeTextSanitizer, toChatEvents, type ChatTurnEvent } from "./events" -import { makeReviewCoverage } from "./review-coverage" +import { makeReviewCoverage, type ReviewCoverage } from "./review-coverage" +import type { ReviewLedger } from "./review-ledger" import { buildReviewFanout } from "./review-fanout" import { accumulateUsage, @@ -99,6 +100,11 @@ export interface ChatRunInput { readonly stageEdit?: StageEdit /** This run is an autonomous pass's close-out: a report it files is a partial. */ readonly closeOut?: boolean + /** + * A review turn's own state, shared by its pass and its close-out: what was read, and the + * findings saved so far. Absent, a review run makes its own coverage and offers no `record_finding`. + */ + readonly review?: ReviewTurnState /** The agent to run as; defaults to the session's. The local review runner passes a variant. */ readonly agent?: AgentDefinition /** The message the user just sent, which is this run's input. */ @@ -111,6 +117,11 @@ export interface ChatRunInput { readonly append: (event: ChatTurnEvent) => void } +export interface ReviewTurnState { + readonly coverage: ReviewCoverage + readonly ledger: ReviewLedger +} + export interface ChatRunOutcome { /** This run was a machine-started pass: an investigation's or a review's. */ readonly autonomous: boolean @@ -139,7 +150,7 @@ export const runChatTurn = (input: ChatRunInput) => { // Per run, and fed by the Maple handlers the review_files children share, so a group a child // read counts as read. const coverage = isAutonomousReviewTurn(input.sessionId, input.origin) - ? makeReviewCoverage(input.text) + ? (input.review?.coverage ?? makeReviewCoverage(input.text)) : undefined // A call's UI payload, held until its result event is written. Never part of what the model reads. const uiByCall = new Map() @@ -177,6 +188,7 @@ export const runChatTurn = (input: ChatRunInput) => { input.closeOut === true, agentSessionSpanAttributes(input.model.tags), coverage, + input.review?.ledger, )) ?? (input.submitReply === undefined || input.stageEdit === undefined ? undefined diff --git a/apps/ai/src/chat/tools.test.ts b/apps/ai/src/chat/tools.test.ts index 47223c43c8..6bcf90071d 100644 --- a/apps/ai/src/chat/tools.test.ts +++ b/apps/ai/src/chat/tools.test.ts @@ -13,6 +13,8 @@ import { buildDiagnosisCompletion, buildReviewCompletion, makeRunUsage, + RECORD_FINDING, + savedFindingsRequest, SUBMIT_DIAGNOSIS, SUBMIT_REVIEW, type SubmitDiagnosis, @@ -20,8 +22,14 @@ import { } from "./tools" import type { TenantContext } from "@maple/backend/services/auth/tenant-context" import { makeRecordingTracer } from "@maple/backend/testing/recording-tracer" -import type { PrReviewSubmission } from "@maple/domain/http" +import { + PrReviewFinding, + type PrReviewFindingSubmission, + type PrReviewSubmission, + type SubmitPrReviewRequest, +} from "@maple/domain/http" import type { ReviewCoverage } from "./review-coverage" +import { makeReviewLedger } from "./review-ledger" const orgId = Schema.decodeSync(OrgId)("org_test") const human = Schema.decodeSync(UserId)("user_test") @@ -227,3 +235,110 @@ describe("buildReviewCompletion", () => { }) }) }) + +describe("record_finding", () => { + const REVIEW_SESSION = prReviewSessionId(orgId, "7f1d3c2e-9a4b-4c8d-8e2f-1a2b3c4d5e6f") + const RETRY: PrReviewFindingSubmission = { + path: "src/orders.ts", + line: 42, + category: "correctness", + severity: "warn", + title: "`retryFetch` re-sends POSTs that are not idempotent", + body: "A timeout after the server committed charges the order twice.", + } + const setup = (unread: ReadonlyArray = [], partial = false) => { + const filed: Array = [] + const ledger = makeReviewLedger() + const completion = buildReviewCompletion( + REVIEW_SESSION, + tenantFor(human), + { kind: "autonomous" }, + (_org, _id, request) => Effect.sync(() => filed.push(request)), + makeRunUsage(), + MODEL_NAME, + partial, + undefined, + { observe: () => {}, unread: () => unread }, + ledger, + ) + assert.isDefined(completion) + const handlers = completion!.handlers + return { + filed, + ledger, + save: (finding: PrReviewFindingSubmission) => + Effect.runPromise(Effect.result(handlers[RECORD_FINDING](finding, {} as never))), + submit: (submission: PrReviewSubmission) => + Effect.runPromise(Effect.result(handlers[SUBMIT_REVIEW](submission, {} as never))), + } + } + + it("saves a finding and adds it to the submitted report, dropping a restatement", async () => { + const { filed, ledger, save, submit } = setup() + assert.isTrue(Result.isSuccess(await save(RETRY))) + assert.lengthOf(ledger.findings(), 1) + + const result = await submit({ + verdict: "clean", + summary: "Adds a retry to the order client.", + findings: [ + { ...RETRY, line: 43, title: "`retryFetch` re-sends POSTs which are not idempotent" }, + { path: "src/orders.ts", line: 90, severity: "info", title: "Unused `backoff` import" }, + ], + }) + assert.isTrue(Result.isSuccess(result)) + const report = filed[0]!.report + assert.deepStrictEqual( + report.findings.map((finding) => finding.line), + [42, 90], + ) + // A saved warn makes the review one with issues, whatever the submission called it. + assert.equal(report.verdict, "issues") + }) + + it("answers a repeat without saving it twice", async () => { + const { ledger, save } = setup() + await save(RETRY) + const again = await save({ ...RETRY, line: 44 }) + assert.isTrue(Result.isSuccess(again)) + assert.include(Result.isSuccess(again) ? String(again.success) : "", "restates") + assert.lengthOf(ledger.findings(), 1) + }) + + it("refuses a finding it could not post", async () => { + const { ledger, save } = setup() + const result = await save({ ...RETRY, line: null }) + assert.isTrue(Result.isFailure(result)) + assert.lengthOf(ledger.findings(), 0) + }) + + it("names the reviewed files a close-out never read on its report", async () => { + const { filed, submit } = setup(["src/b.ts", "src/c.ts"], true) + assert.isTrue(Result.isSuccess(await submit({ verdict: "clean", summary: "Partial." }))) + assert.deepStrictEqual(filed[0]!.report.unreviewed, ["src/b.ts", "src/c.ts"]) + }) +}) + +describe("savedFindingsRequest", () => { + it("files saved findings as a partial with the unread files", () => { + const request = savedFindingsRequest({ + findings: [ + new PrReviewFinding({ + path: "src/a.ts", + line: 3, + category: "security", + severity: "critical", + title: "Tenant filter missing from `listKeys`", + body: "", + }), + ], + unreviewed: ["src/b.ts"], + modelName: MODEL_NAME, + usage: makeRunUsage(), + }) + assert.isTrue(request.partial) + assert.equal(request.report.verdict, "issues") + assert.deepStrictEqual(request.report.unreviewed, ["src/b.ts"]) + assert.lengthOf(request.report.findings, 1) + }) +}) diff --git a/apps/ai/src/chat/tools.ts b/apps/ai/src/chat/tools.ts index e8ee07dcfa..8c4bd05760 100644 --- a/apps/ai/src/chat/tools.ts +++ b/apps/ai/src/chat/tools.ts @@ -17,8 +17,12 @@ import { InvestigationDataCorruptionError, InvestigationNotFoundError, InvestigationPersistenceError, + normalizePrReviewFinding, normalizePrReviewSubmission, normalizeTriageSubmission, + PR_REVIEW_MAX_FINDINGS, + type PrReviewFinding, + PrReviewFindingSubmission, PrReviewId, PrReviewNotFoundError, PrReviewEditSubmission, @@ -26,6 +30,7 @@ import { PrReviewReplyId, type PrReviewReplyNotFoundError, PrReviewReplySubmission, + PrReviewReport, PrReviewSubmission, type PrReviewVerdict, SubmitDiagnosisRequest, @@ -45,7 +50,9 @@ import { } from "../mcp/tools/llm-tools" import { toolHandlersWithContent } from "../platform/genai-spans" import type { TenantContext } from "@maple/backend/services/auth/tenant-context" +import { mergeSavedFindings } from "@maple/backend/services/pr-review/findings" import { type ReviewCoverage, unreadRefusal } from "./review-coverage" +import { makeReviewLedger, type ReviewLedger } from "./review-ledger" const decodeInvestigationIdOption = Schema.decodeUnknownOption(InvestigationId) const decodePrReviewIdOption = Schema.decodeUnknownOption(PrReviewId) @@ -62,6 +69,7 @@ export type SubmitDiagnosis = ( export const SUBMIT_DIAGNOSIS = "submit_diagnosis" export const SUBMIT_REVIEW = "submit_review" +export const RECORD_FINDING = "record_finding" export const SUBMIT_REPLY = "submit_reply" export const PROPOSE_EDIT = "propose_edit" @@ -273,6 +281,55 @@ export const reviewTool = Tool.make(SUBMIT_REVIEW, { failure: MapleToolFailure, }) +/** Same lenient finding shape `submit_review` takes, one at a time. */ +export const recordFindingTool = Tool.make(RECORD_FINDING, { + description: + "Save one finding the moment you have established it: path, new-side line, category, " + + "severity, title, body, and suggestion or replacement. Saved findings are posted with your " + + "review even if the pass stops before submit_review, and submit_review adds them for you, so " + + "never repeat one there or save the same issue twice. There is no way to edit or remove a " + + "saved finding: check it against the code before you save it.", + parameters: PrReviewFindingSubmission, + success: Schema.String, + failure: MapleToolFailure, +}) + +/** What the model reads back after a `record_finding` call. */ +const recordAnswer = (outcome: ReturnType, count: number): string => { + switch (outcome) { + case "recorded": + return `Saved (${count} ${count === 1 ? "finding" : "findings"} so far). Do not repeat it in submit_review.` + case "repeat": + return `Not saved: it restates a finding already saved (${count} so far). Keep going.` + case "full": + return `Not saved: the review holds at most ${PR_REVIEW_MAX_FINDINGS} findings. Submit the review now.` + } +} + +/** + * The review a turn files itself when its findings were saved but no `submit_review` landed: a + * partial with exactly those findings, so the deadline that stopped the pass does not lose them. + */ +export const savedFindingsRequest = (input: { + readonly findings: ReadonlyArray + readonly unreviewed: ReadonlyArray + readonly modelName: string + readonly usage: RunUsage +}): SubmitPrReviewRequest => + new SubmitPrReviewRequest({ + report: new PrReviewReport({ + verdict: input.findings.some((finding) => finding.severity !== "info") ? "issues" : "clean", + summary: "", + coverage: [], + findings: input.findings, + ...(input.unreviewed.length > 0 ? { unreviewed: input.unreviewed } : undefined), + }), + model: input.modelName, + inputTokens: input.usage.input, + outputTokens: input.usage.output, + partial: true, + }) + /** * The `submit_review` tool for a review session (`":pr-"`), built like * {@link buildDiagnosisCompletion}: the review id rides on the session, so the agent never chooses @@ -290,13 +347,15 @@ export const buildReviewCompletion = ( sessionAttributes?: Readonly>, /** What this pass has read. The first submission with files unread is refused once. */ coverage?: ReviewCoverage, + /** Findings saved with `record_finding`; the turn's own, or this run's when none is passed. */ + ledger: ReviewLedger = makeReviewLedger(), ) => { const reviewId = prReviewForSession(sessionId) if (reviewId === undefined) return undefined // Only the unattended pass files a review. A follow-up in the session answers in prose: the // row is already settled, and a second submission would be dropped while reporting success. if (origin.kind !== "autonomous") return undefined - const toolkit = Toolkit.make(reviewTool) + const toolkit = Toolkit.make(reviewTool, recordFindingTool) let submitted = false let refusedUnread = false // Once only: a second call with files still unread is the agent submitting anyway. A close-out @@ -309,13 +368,18 @@ export const buildReviewCompletion = ( } const record = (submission: PrReviewSubmission) => Effect.suspend(() => { - const { report, filled, droppedFindings, resolved } = normalizePrReviewSubmission(submission) - const unread = unreadToRefuse(report.verdict) + const normalized = normalizePrReviewSubmission(submission) + const { filled, droppedFindings, resolved } = normalized + const unread = unreadToRefuse(normalized.report.verdict) if (unread.length > 0) { return Effect.annotateCurrentSpan("maple.pr_review.unread_files", unread.length).pipe( Effect.andThen(Effect.fail(new MapleToolFailure({ message: unreadRefusal(unread) }))), ) } + const saved = ledger.findings() + const unreviewed = + normalized.report.verdict === "not_applicable" ? [] : (coverage?.unread() ?? []) + const report = withSavedFindings(normalized.report, saved, unreviewed) return submitReview( tenant.orgId, reviewId, @@ -334,6 +398,8 @@ export const buildReviewCompletion = ( "maple.pr_review.filled_count": filled.length, "maple.pr_review.dropped_findings": droppedFindings, "maple.pr_review.findings": report.findings.length, + "maple.pr_review.saved_findings": saved.length, + "maple.pr_review.unreviewed_files": unreviewed.length, "maple.pr_review.verdict": report.verdict, "maple.pr_review.resolved": resolved.length, }), @@ -349,15 +415,62 @@ export const buildReviewCompletion = ( ), ) }) + const save = (raw: PrReviewFindingSubmission) => + Effect.suspend(() => { + const finding = normalizePrReviewFinding(raw) + if (finding === undefined) { + return Effect.fail( + new MapleToolFailure({ + message: + "Not saved: a finding needs a path and a positive new-side line from a hunk you read, and an observability finding needs its audit check id.", + }), + ) + } + const outcome = ledger.record(finding) + return Effect.annotateCurrentSpan("maple.pr_review.record_outcome", outcome).pipe( + Effect.as(recordAnswer(outcome, ledger.findings().length)), + ) + }) return { tool: SUBMIT_REVIEW, toolkit, - ...toolHandlersWithContent(toolkit, { [SUBMIT_REVIEW]: record }, sessionAttributes), + ...toolHandlersWithContent( + toolkit, + { [SUBMIT_REVIEW]: record, [RECORD_FINDING]: save }, + sessionAttributes, + ), autonomous: isAutonomousReviewTurn(sessionId, origin), submitted: () => submitted, } satisfies RunCompletion } +const SEVERITY_RANK = { critical: 2, warn: 1, info: 0 } as const satisfies Record< + PrReviewFinding["severity"], + number +> + +/** + * A submitted report with the pass's saved findings folded in, most important kept when over the + * cap, and the reviewed files it never read. The verdict follows the merged findings. + */ +const withSavedFindings = ( + report: PrReviewReport, + saved: ReadonlyArray, + unreviewed: ReadonlyArray, +): PrReviewReport => { + if (saved.length === 0 && unreviewed.length === 0) return report + // Stable, so saved findings stay ahead of submitted ones of the same severity. + const findings = [...mergeSavedFindings(saved, report.findings)] + .sort((a, b) => SEVERITY_RANK[b.severity] - SEVERITY_RANK[a.severity]) + .slice(0, PR_REVIEW_MAX_FINDINGS) + return new PrReviewReport({ + ...report, + findings, + verdict: findings.some((finding) => finding.severity !== "info") ? "issues" : report.verdict, + ...(unreviewed.length > 0 ? { unreviewed } : undefined), + }) +} + /** * The `submit_diagnosis` tool for an investigate-mode session (`":inv-"`). * diff --git a/apps/ai/src/chat/turn-runner.ts b/apps/ai/src/chat/turn-runner.ts index 307821b1ab..5538e29dec 100644 --- a/apps/ai/src/chat/turn-runner.ts +++ b/apps/ai/src/chat/turn-runner.ts @@ -21,7 +21,7 @@ import * as MapleCloudflareSDK from "@maple-dev/effect-sdk/cloudflare" import { MCP_ANTICIPATED_ERROR_IDENTIFIERS } from "../mcp/expected-failures" import { ChatMessage, type ChatTurnOrigin, type ChatTurnTenantEncoded } from "@maple/domain/chat-session" -import type { InvestigationProgress } from "@maple/domain/http" +import type { InvestigationProgress, PrReviewFailureReason } from "@maple/domain/http" import { workerEnvLayer } from "@maple/infra/worker-runtime" import { workerTelemetryConfig } from "@maple/infra/worker-telemetry" import { Cause, Effect, Layer, ManagedRuntime, Option } from "effect" @@ -30,12 +30,16 @@ import type { ChatTurnEvent } from "./events" import { withToolTranscript } from "./close-out" import { CLOSE_OUT_PROMPT, PR_REPLY_CLOSE_OUT_PROMPT, PR_REVIEW_CLOSE_OUT_PROMPT } from "./prompts" import { makeProgressRecorder, parseToolInput } from "./progress" +import { makeReviewCoverage } from "./review-coverage" +import { reviewFailureError, reviewFailureReason } from "./review-failure" +import { makeReviewLedger } from "./review-ledger" import { investigationForSession, isAutonomousTurn, makeRunUsage, prReplyForSession, prReviewForSession, + savedFindingsRequest, SUBMIT_DIAGNOSIS, } from "./tools" @@ -124,7 +128,6 @@ const NO_DIAGNOSIS_MESSAGE = "Maple ended this investigation without a diagnosis /** What the row records when the pass and its close-out both ended in prose or on an error. */ const NO_DIAGNOSIS_ERROR = "no_diagnosis: the agent ended its pass without submitting a diagnosis; retry" const NO_REVIEW_MESSAGE = "Maple ended this review without a report." -const NO_REVIEW_ERROR = "no_review: the agent ended its pass without submitting a review; retry" const NO_REPLY_MESSAGE = "Maple ended this answer without posting it." const NO_REPLY_ERROR = "no_reply: the agent ended its pass without submitting an answer" @@ -359,6 +362,13 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis const text = latest?.role === "user" ? latest.text : "" const prior = latest?.role === "user" ? spoken.slice(0, -1) : spoken const autonomous = isAutonomousTurn(input.sessionId, origin) + // One per review turn: the close-out sees what the pass read and keeps what it saved. + const review = + prReviewId !== undefined && autonomous + ? { coverage: makeReviewCoverage(text), ledger: makeReviewLedger() } + : undefined + // Why the pass, or else its close-out, stopped; the first reason is the one the PR is told. + let failure: PrReviewFailureReason | undefined const holdsTurn = () => input.session.holdsTurn(input.messageId) // An autonomous pass ends when the runner says so, not when a run does: a run that stopped // in prose or died on a model error gets one close-out turn first, so its terminal is held @@ -381,6 +391,7 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis submitReply: conversations.submitReply, stageEdit: conversations.stageEdit, ...(turn.closeOut === true ? { closeOut: true } : undefined), + ...(review === undefined ? undefined : { review }), text: turn.text, history: turn.history, usage, @@ -417,7 +428,10 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis Effect.catchCause((cause) => Cause.hasInterruptsOnly(cause) ? Effect.failCause(cause) - : Effect.logWarning("Autonomous pass failed; closing it out").pipe( + : Effect.sync(() => { + failure ??= reviewFailureReason(cause) + }).pipe( + Effect.andThen(Effect.logWarning("Autonomous pass failed; closing it out")), Effect.annotateLogs({ sessionId: input.sessionId, messageId: input.messageId, @@ -442,6 +456,8 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis const first = yield* recoverAutonomousFailure(run({ text, history: prior })) observability.compactions = first.compactions let submitted = first.submitted + // Exhausted rather than failed: the engine's final answer ended in prose, not a report. + if (!submitted && held?.reason === "max-steps") failure ??= "step_limit" if (!submitted && holdsTurn()) { held = undefined const closeOut = yield* recoverAutonomousFailure( @@ -491,10 +507,35 @@ export const runChatSessionTurn = async (input: RunChatSessionTurnInput): Promis ), ) } + // Findings the pass saved are posted as a partial rather than lost with the report. + const saved = review?.ledger.findings() ?? [] + if (!submitted && prReviewId !== undefined && review !== undefined && saved.length > 0) { + submitted = yield* reviews + .submitReview( + tenant.orgId, + prReviewId, + savedFindingsRequest({ + findings: saved, + unreviewed: review.coverage.unread(), + modelName: model.name, + usage, + }), + ) + .pipe( + Effect.as(true), + Effect.catchCause((cause) => + Effect.logError("Could not file the saved findings", cause).pipe( + Effect.as(false), + ), + ), + ) + yield* Effect.annotateCurrentSpan("maple.pr_review.filed_saved_findings", submitted) + } if (!submitted && prReviewId !== undefined) { observability.failureReason = "NoReview" + yield* Effect.annotateCurrentSpan("maple.pr_review.failure_reason", failure ?? "no_report") yield* reviews - .failReview(tenant.orgId, prReviewId, NO_REVIEW_ERROR) + .failReview(tenant.orgId, prReviewId, reviewFailureError(failure ?? "no_report")) .pipe( Effect.catchCause((cause) => Effect.logError("Could not record the failed review", cause), diff --git a/apps/ai/src/platform/Llm.ts b/apps/ai/src/platform/Llm.ts index 5e879e6ca7..ba2ef99f8f 100644 --- a/apps/ai/src/platform/Llm.ts +++ b/apps/ai/src/platform/Llm.ts @@ -344,11 +344,12 @@ const instrumentedModel = ( name: string, make: Effect.Effect, telemetry: ModelCallTelemetry, + options: { readonly coalesceDeltas?: boolean } = {}, ): Layer.Layer => AiModel.make( "openrouter", name, - Layer.effect(LanguageModel.LanguageModel, instrumentLanguageModel(make, telemetry)), + Layer.effect(LanguageModel.LanguageModel, instrumentLanguageModel(make, telemetry, options)), ) const openRouterModel = ( @@ -358,6 +359,8 @@ const openRouterModel = ( fallbackEffort: ReasoningEffort | undefined, tags: LlmCallTags | undefined, overridableLimits = true, + /** Nobody watches the run stream, so its deltas are joined; see `coalesceDeltas`. */ + unattended = false, ): ResolvedModel => { const effort = readReasoningEffort(env, effortKey) ?? fallbackEffort return { @@ -371,6 +374,7 @@ const openRouterModel = ( ...(effort === undefined || effort === "off" ? undefined : { reasoningLevel: effort }), sessionAttributes: agentSessionSpanAttributes(tags), }, + { coalesceDeltas: unattended }, ), limits: limitsFor(env, name, overridableLimits), tags, @@ -429,6 +433,7 @@ export const resolveReviewModel = (env: LlmEnv, tags?: LlmCallTags): ResolvedMod undefined, tags, false, + true, ) /** diff --git a/apps/ai/src/platform/coalesce-deltas.test.ts b/apps/ai/src/platform/coalesce-deltas.test.ts new file mode 100644 index 0000000000..39c83cb91d --- /dev/null +++ b/apps/ai/src/platform/coalesce-deltas.test.ts @@ -0,0 +1,42 @@ +/** + * Joined deltas carry the same text in fewer parts, and never cross a part or a kind. + */ +import { Effect, Stream } from "effect" +import { Response } from "effect/unstable/ai" +import { assert, describe, it } from "vitest" +import { COALESCED_DELTA_CHARS, coalesceDeltas } from "./coalesce-deltas" + +const run = (parts: ReadonlyArray) => + Effect.runPromise(Stream.runCollect(coalesceDeltas(Stream.fromIterable(parts)))) + +describe("coalesceDeltas", () => { + it("joins a part's deltas and keeps every other part in order", async () => { + const out = await run([ + Response.makePart("reasoning-start", { id: "r" }), + Response.makePart("reasoning-delta", { id: "r", delta: "Let " }), + Response.makePart("reasoning-delta", { id: "r", delta: "me " }), + Response.makePart("reasoning-delta", { id: "r", delta: "check." }), + Response.makePart("reasoning-end", { id: "r" }), + Response.makePart("text-delta", { id: "t", delta: "Done" }), + Response.makePart("text-delta", { id: "t", delta: "." }), + ]) + assert.deepEqual( + out.map((part) => part.type), + ["reasoning-start", "reasoning-delta", "reasoning-end", "text-delta"], + ) + assert.equal(out[1]!.type === "reasoning-delta" ? out[1]!.delta : "", "Let me check.") + // The last delta is flushed when the stream ends. + assert.equal(out[3]!.type === "text-delta" ? out[3]!.delta : "", "Done.") + }) + + it("never joins two parts, two kinds, or past the size bound", async () => { + const big = "x".repeat(COALESCED_DELTA_CHARS) + const out = await run([ + Response.makePart("text-delta", { id: "a", delta: "1" }), + Response.makePart("text-delta", { id: "b", delta: "2" }), + Response.makePart("reasoning-delta", { id: "b", delta: "3" }), + Response.makePart("reasoning-delta", { id: "b", delta: big }), + ]) + assert.equal(out.length, 4) + }) +}) diff --git a/apps/ai/src/platform/coalesce-deltas.ts b/apps/ai/src/platform/coalesce-deltas.ts new file mode 100644 index 0000000000..8883e61601 --- /dev/null +++ b/apps/ai/src/platform/coalesce-deltas.ts @@ -0,0 +1,41 @@ +/** + * Consecutive streamed deltas of one part, joined before the engine sees them. + * + * effect-agent counts every delta as a run event (65,536 per run) and a response part (16,384 per + * call), and neither ceiling can be raised. A model that streams a delta per token spends them on + * reasoning: a 126-call review died on the event ceiling (2026-09-25). Joined deltas keep the text + * and cost nothing but live streaming, so only unattended runs use this. + */ +import { Stream } from "effect" +import type { Response } from "effect/unstable/ai" + +/** The most a joined delta holds, so one never grows past a response's byte bound in one piece. */ +export const COALESCED_DELTA_CHARS = 4_000 + +type DeltaPart = Response.TextDeltaPart | Response.ReasoningDeltaPart | Response.ToolParamsDeltaPart + +const isDelta = (part: A): part is A & DeltaPart => + part.type === "text-delta" || part.type === "reasoning-delta" || part.type === "tool-params-delta" + +/** A later delta's metadata is dropped: the first carries the part's, and providers put theirs on the end part. */ +export const coalesceDeltas = ( + stream: Stream.Stream, +): Stream.Stream => + stream.pipe( + Stream.mapAccum( + (): (A & DeltaPart) | undefined => undefined, + (pending, part): readonly [(A & DeltaPart) | undefined, ReadonlyArray] => { + if (!isDelta(part)) return [undefined, pending === undefined ? [part] : [pending, part]] + if ( + pending !== undefined && + pending.type === part.type && + pending.id === part.id && + pending.delta.length + part.delta.length <= COALESCED_DELTA_CHARS + ) { + return [{ ...pending, delta: pending.delta + part.delta }, []] + } + return [part, pending === undefined ? [] : [pending]] + }, + { onHalt: (pending) => (pending === undefined ? [] : [pending]) }, + ), + ) diff --git a/apps/ai/src/platform/genai-spans.ts b/apps/ai/src/platform/genai-spans.ts index 4fcee06073..1d58e16846 100644 --- a/apps/ai/src/platform/genai-spans.ts +++ b/apps/ai/src/platform/genai-spans.ts @@ -26,6 +26,7 @@ import type * as Prompt from "effect/unstable/ai/Prompt" import * as Telemetry from "effect/unstable/ai/Telemetry" import * as Tool from "effect/unstable/ai/Tool" import type * as Toolkit from "effect/unstable/ai/Toolkit" +import { coalesceDeltas } from "./coalesce-deltas" /** * Per-attribute size budgets, in JSON characters. @@ -367,6 +368,8 @@ const idleTimeout = (): AiError.AiError => export const instrumentLanguageModel = ( make: Effect.Effect, telemetry: ModelCallTelemetry, + /** Join streamed deltas (see `coalesceDeltas`); for models only unattended runs use. */ + instrumentation: { readonly coalesceDeltas?: boolean } = {}, ): Effect.Effect => make.pipe( Effect.map((service) => ({ @@ -395,6 +398,9 @@ export const instrumentLanguageModel = ( duration: MODEL_STREAM_IDLE_TIMEOUT, orElse: () => Stream.fail(idleTimeout()), }), + // After the idle timeout, which must keep seeing every raw delta. + (stream) => + instrumentation.coalesceDeltas === true ? coalesceDeltas(stream) : stream, Stream.provideService( Telemetry.CurrentSpanTransformer, modelCallTransformer(telemetry, timing), diff --git a/apps/ai/src/platform/model-call-span.test.ts b/apps/ai/src/platform/model-call-span.test.ts index 3edcd1abf7..3fe6c6e239 100644 --- a/apps/ai/src/platform/model-call-span.test.ts +++ b/apps/ai/src/platform/model-call-span.test.ts @@ -609,3 +609,43 @@ describe("the model-call span", () => { }), ) }) + +describe("delta joining", () => { + const deltas = Stream.fromIterable([ + { type: "text-start", id: "t" }, + { type: "text-delta", id: "t", delta: "Looks " }, + { type: "text-delta", id: "t", delta: "safe" }, + { type: "text-delta", id: "t", delta: "." }, + { type: "text-end", id: "t" }, + ]) + const partsSeen = (coalesceDeltas: boolean) => + Effect.gen(function* () { + const provider = LanguageModel.make({ + generateText: () => Effect.succeed([]), + streamText: () => deltas, + }) + const model = Layer.effect( + LanguageModel.LanguageModel, + instrumentLanguageModel( + provider, + { providerName: "fake", sessionAttributes: {} }, + { coalesceDeltas }, + ), + ) + const parts = yield* LanguageModel.streamText({ prompt: "hi" }).pipe( + Stream.runCollect, + Effect.provide(model), + ) + return parts.filter((part) => part.type === "text-delta") + }) + + /** The request's own `options` once shadowed this setting, and nothing was ever joined. */ + it.effect("joins a model's deltas when the model is built to, and only then", () => + Effect.gen(function* () { + const joined = yield* partsSeen(true) + assert.lengthOf(joined, 1) + assert.strictEqual(joined[0]!.type === "text-delta" ? joined[0]!.delta : "", "Looks safe.") + assert.lengthOf(yield* partsSeen(false), 3) + }), + ) +}) diff --git a/apps/sandbox/src/checkout.test.ts b/apps/sandbox/src/checkout.test.ts index 8e6ad9d1e1..c2f0989548 100644 --- a/apps/sandbox/src/checkout.test.ts +++ b/apps/sandbox/src/checkout.test.ts @@ -174,6 +174,12 @@ describe("cloneScript", () => { assert.include(script, "chmod -R a+rX,go-w") }) + it("checks symlinks out as plain files, before the checkout, so none can be followed", () => { + const script = cloneScript(checkout) + assert.include(script, `config core.symlinks false`) + assert.isTrue(script.indexOf("core.symlinks false") < script.indexOf("checkout --quiet --detach")) + }) + it("evicts the oldest checkouts, scratch directories included, so the disk cannot fill", () => { const script = cloneScript(checkout) assert.include(script, "tail -n +4") diff --git a/apps/sandbox/src/checkout.ts b/apps/sandbox/src/checkout.ts index 4d383fa816..690d95c756 100644 --- a/apps/sandbox/src/checkout.ts +++ b/apps/sandbox/src/checkout.ts @@ -270,6 +270,9 @@ export const cloneScript = (checkout: SandboxCheckout): string => { `git clone --quiet --shared --no-checkout ${mirror} "$t"`, "exec 9>&-", `git -C "$t" remote set-url origin ${shellQuote(checkout.remoteUrl)}`, + // A committed symlink is checked out as a file holding its target, so no command the agent + // runs can follow one out of the checkout; the index still records mode 120000. + `git -C "$t" config core.symlinks false`, `git -C "$t" checkout --quiet --detach ${shellQuote(checkout.sha)}`, // Readable and traversable by the agent account, writable by nobody but root. // `mktemp -d` creates the directory mode 700, so read and execute have to be diff --git a/apps/web/src/components/integrations/pr-review-settings.tsx b/apps/web/src/components/integrations/pr-review-settings.tsx index eaba83bb65..960c0f3a85 100644 --- a/apps/web/src/components/integrations/pr-review-settings.tsx +++ b/apps/web/src/components/integrations/pr-review-settings.tsx @@ -47,6 +47,7 @@ import { MapleApiAtomClient, retainedQuery } from "@/lib/services/common/atom-cl const INSTRUCTIONS_MAX = 4_000 const IGNORE_PATH_MAX = 200 const DAILY_LIMIT_MAX = 500 +const AUTOMATIC_LIMIT_MAX = 50 /** How often the review list refetches while a review is queued or running. */ const REVIEWS_POLL_MS = 5_000 @@ -86,6 +87,7 @@ const SKIP_LABELS = { superseded: "a newer push was reviewed", agent_unavailable: "reviewer unavailable", not_rolled_out: "not enabled for this organization", + automatic_limit: "pull request limit reached", } satisfies Record const configKey = (repo: GithubRepoSummary) => `githubPrReviewConfig:${repo.id}` @@ -163,6 +165,7 @@ interface FormState { readonly minInlineSeverity: PrReviewSeverity readonly reviewDrafts: boolean readonly dailyLimit: string + readonly automaticReviewLimit: string readonly feedbackScope: PrReviewFeedbackScope } @@ -173,6 +176,8 @@ const stateFromConfig = (config: PrReviewRepositoryConfig): FormState => ({ minInlineSeverity: config.minInlineSeverity ?? "warn", reviewDrafts: config.reviewDrafts ?? false, dailyLimit: config.dailyLimit === undefined ? "" : String(config.dailyLimit), + automaticReviewLimit: + config.automaticReviewLimit === undefined ? "" : String(config.automaticReviewLimit), feedbackScope: config.feedbackScope ?? "organization", }) @@ -182,6 +187,7 @@ const sameState = (a: FormState, b: FormState) => a.minInlineSeverity === b.minInlineSeverity && a.reviewDrafts === b.reviewDrafts && a.dailyLimit === b.dailyLimit && + a.automaticReviewLimit === b.automaticReviewLimit && a.feedbackScope === b.feedbackScope && a.categories.length === b.categories.length && a.categories.every((category) => b.categories.includes(category)) @@ -205,6 +211,12 @@ const validate = (state: FormState): string | null => { if (!Number.isInteger(value) || value < 1 || value > DAILY_LIMIT_MAX) return `Daily limit must be a whole number from 1 to ${DAILY_LIMIT_MAX}.` } + const perPullRequest = state.automaticReviewLimit.trim() + if (perPullRequest !== "") { + const value = Number(perPullRequest) + if (!Number.isInteger(value) || value < 1 || value > AUTOMATIC_LIMIT_MAX) + return `Reviews per pull request must be a whole number from 1 to ${AUTOMATIC_LIMIT_MAX}.` + } return null } @@ -213,6 +225,7 @@ const configFromState = (state: FormState) => { const instructions = state.instructions.trim() const ignorePaths = parseIgnorePaths(state.ignorePaths) const limit = state.dailyLimit.trim() + const perPullRequest = state.automaticReviewLimit.trim() return new PrReviewRepositoryConfig({ ...(instructions === "" ? undefined : { instructions }), ...(ignorePaths.length === 0 ? undefined : { ignorePaths }), @@ -223,6 +236,7 @@ const configFromState = (state: FormState) => { ...(state.minInlineSeverity === "warn" ? undefined : { minInlineSeverity: state.minInlineSeverity }), ...(state.reviewDrafts ? { reviewDrafts: true } : undefined), ...(limit === "" ? undefined : { dailyLimit: Number(limit) }), + ...(perPullRequest === "" ? undefined : { automaticReviewLimit: Number(perPullRequest) }), ...(state.feedbackScope === "organization" ? undefined : { feedbackScope: state.feedbackScope }), }) } @@ -374,6 +388,24 @@ function ConfigForm({ repo, config }: { repo: GithubRepoSummary; config: PrRevie

+
+ + update({ automaticReviewLimit: event.target.value })} + /> +

+ After this many, pushes stop starting reviews. Comment @maple review to run one. +

+
+