diff --git a/.github/workflows/benchmark-calibration.yml b/.github/workflows/benchmark-calibration.yml index 3dbc661..d12defb 100644 --- a/.github/workflows/benchmark-calibration.yml +++ b/.github/workflows/benchmark-calibration.yml @@ -13,7 +13,7 @@ jobs: permissions: contents: write steps: - - name: Reserve the current main calibration source + - name: Verify the current main calibration source env: GH_TOKEN: ${{ github.token }} run: | @@ -31,10 +31,6 @@ jobs: echo "::error::Calibration source is not the current main commit." exit 1 } - gh api --method POST "repos/${GITHUB_REPOSITORY}/git/refs" \ - -f ref="refs/tags/postil-calibration-${GITHUB_SHA}" \ - -f sha="${GITHUB_SHA}" \ - --silent - uses: actions/checkout@3d3c42e5aac5ba805825da76410c181273ba90b1 # v7.0.1 - uses: oven-sh/setup-bun@0c5077e51419868618aeaa5fe8019c62421857d6 # v2 with: @@ -45,10 +41,76 @@ jobs: - name: Verify embedded model admission working-directory: bench run: bun run verify-admission --allow-provisional - - name: Require an unpopulated Azure/US Luna baseline - run: >- - jq -e '.profiles["openai/gpt-5.6-luna"].populated == false' - bench/baseline-us.json >/dev/null + - name: Fingerprint the current evaluator contract + id: evaluator + working-directory: bench + run: | + bun -e 'import { evaluatorSourceSha256 } from "./src/live"; console.log("sha256=" + await evaluatorSourceSha256())' >> "$GITHUB_OUTPUT" + - name: Verify the calibration baseline input + env: + GH_TOKEN: ${{ github.token }} + CURRENT_EVALUATOR_SHA256: ${{ steps.evaluator.outputs.sha256 }} + run: | + set -euo pipefail + profile="openai/gpt-5.6-luna" + populated="$(jq -er --arg profile "$profile" '.profiles[$profile].populated | if type == "boolean" then tostring else error("Calibration populated state must be boolean") end' bench/baseline-us.json)" + case "$populated" in + false) exit 0 ;; + true) ;; + *) echo "::error::Calibration baseline must declare a boolean populated state."; exit 1 ;; + esac + test -s bench/baseline-us.attestation.json || { + echo "::error::Populated recalibration requires the retained baseline attestation." + exit 1 + } + source_sha="$(jq -er --arg profile "$profile" '.profiles[$profile].calibration.sourceSha' bench/baseline-us.json)" + [[ "$source_sha" =~ ^[0-9a-f]{40,64}$ ]] || { + echo "::error::Prior calibration source digest is invalid." + exit 1 + } + test "$source_sha" != "$GITHUB_SHA" || { + echo "::error::Recalibration requires a new main source commit." + exit 1 + } + prior_evaluator="$(jq -er '.corpus.evaluatorSha256' bench/baseline-us.json)" + [[ "$prior_evaluator" =~ ^[0-9a-f]{64}$ && "$CURRENT_EVALUATOR_SHA256" =~ ^[0-9a-f]{64}$ ]] || { + echo "::error::Calibration evaluator digest is invalid." + exit 1 + } + test "$prior_evaluator" != "$CURRENT_EVALUATOR_SHA256" || { + echo "::error::Populated recalibration requires a changed evaluator contract." + exit 1 + } + registry_sha="$(gh api "repos/${GITHUB_REPOSITORY}/git/ref/tags/postil-calibration-${source_sha}" --jq .object.sha)" + test "$registry_sha" = "$source_sha" || { + echo "::error::Prior calibration registry source mismatch." + exit 1 + } + gh attestation verify bench/baseline-us.json \ + --bundle bench/baseline-us.attestation.json \ + --repo postil-dev/postil-cli \ + --signer-workflow postil-dev/postil-cli/.github/workflows/benchmark-calibration.yml \ + --signer-digest "$source_sha" \ + --source-digest "$source_sha" \ + --source-ref refs/heads/main \ + --cert-oidc-issuer https://token.actions.githubusercontent.com \ + --predicate-type https://slsa.dev/provenance/v1 \ + --deny-self-hosted-runners \ + --format json >/dev/null + - name: Reserve the current main calibration source + env: + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + main_sha="$(gh api "repos/${GITHUB_REPOSITORY}/git/ref/heads/main" --jq .object.sha)" + test "${GITHUB_SHA}" = "$main_sha" || { + echo "::error::Calibration source is no longer the current main commit." + exit 1 + } + gh api --method POST "repos/${GITHUB_REPOSITORY}/git/refs" \ + -f ref="refs/tags/postil-calibration-${GITHUB_SHA}" \ + -f sha="${GITHUB_SHA}" \ + --silent prepare: needs: validate @@ -149,6 +211,7 @@ jobs: name: benchmark-calibration-prepared-${{ github.run_id }} path: ${{ runner.temp }}/prepared - name: Verify prepared calibration provenance + id: verify-prepared env: GH_TOKEN: ${{ github.token }} run: | @@ -173,6 +236,7 @@ jobs: - name: Resolve canonical sample slot run: echo "CALIBRATION_SLOT=$(printf '%02d' '${{ matrix.sample }}')" >> "$GITHUB_ENV" - name: Reserve benchmark sample ${{ matrix.sample }} + id: reserve-sample working-directory: bench run: >- bun run bench:cohort-run -- @@ -193,6 +257,7 @@ jobs: cp "${RUNNER_TEMP}/prepared/slots/${CALIBRATION_SLOT}/receipt.json" \ "${RUNNER_TEMP}/reservation.receipt.json" - name: Execute benchmark sample ${{ matrix.sample }} + id: execute-sample working-directory: bench env: OPENROUTER_API_KEY: ${{ secrets.OPENROUTER_API_KEY }} @@ -213,6 +278,7 @@ jobs: ${{ runner.temp }}/prepared/slots/${{ env.CALIBRATION_SLOT }}/report.json ${{ runner.temp }}/prepared/slots/${{ env.CALIBRATION_SLOT }}/receipt.json - name: Stage benchmark sample evidence + id: stage-sample if: always() run: | set -euo pipefail @@ -228,6 +294,7 @@ jobs: cp "${{ steps.attest-result.outputs.bundle-path }}" "$evidence/result.attestation.json" fi - name: Upload benchmark sample evidence + id: upload-sample if: always() uses: actions/upload-artifact@043fb46d1a93c77aae656e7c1c64a875d1fc6a0a # v7.0.1 with: @@ -235,6 +302,25 @@ jobs: path: ${{ runner.temp }}/benchmark-calibration-sample-${{ env.CALIBRATION_SLOT }} if-no-files-found: error retention-days: 90 + - name: Summarize failed calibration sample + if: failure() + env: + SAMPLE: ${{ matrix.sample }} + STEP_RESULTS: ${{ toJSON(steps) }} + UPLOAD_OUTCOME: ${{ steps.upload-sample.outcome }} + run: | + { + echo "Calibration sample ${SAMPLE} failed; step outcomes:" + jq -r 'to_entries[] | "\(.key): \(.value.outcome)"' <<< "$STEP_RESULTS" + echo "Run logs (job ${GITHUB_JOB}, attempt ${GITHUB_RUN_ATTEMPT}): ${GITHUB_SERVER_URL}/${GITHUB_REPOSITORY}/actions/runs/${GITHUB_RUN_ID}" + if [[ "$UPLOAD_OUTCOME" == "success" ]]; then + echo "Uploaded sample evidence: benchmark-calibration-sample-${GITHUB_RUN_ID}-${SAMPLE}." + else + echo "Sample artifact was not uploaded (upload outcome: ${UPLOAD_OUTCOME:-not run}); use the run/job logs." + fi + echo "Failed evidence is not a calibration." + echo "Fix the source on main and start a new first-attempt run. Do not reuse this reservation or rerun this cohort." + } >> "$GITHUB_STEP_SUMMARY" record: needs: [validate, prepare, sample] diff --git a/Cargo.lock b/Cargo.lock index 7c67047..1291c48 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1358,7 +1358,7 @@ checksum = "05c8b63e8d9609db387f0324918f81d68fe27748f084ef092fb35954d0539a85" [[package]] name = "postil-cli" -version = "0.9.20" +version = "0.9.21" dependencies = [ "aho-corasick", "anyhow", diff --git a/Cargo.toml b/Cargo.toml index c2a6cf8..2d26b48 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "postil-cli" -version = "0.9.20" +version = "0.9.21" edition = "2024" description = "Postil: a low-noise AI review gate. Silent on clean PRs, hard gate on real risk." license = "Apache-2.0" diff --git a/bench/fixtures/cases.ts b/bench/fixtures/cases.ts index f4b5778..af9ee6a 100644 --- a/bench/fixtures/cases.ts +++ b/bench/fixtures/cases.ts @@ -92,11 +92,11 @@ export function makeDiff(path: string, hunks: FixtureHunk[]): string { ].join("\n"); } -function makeDistantStreamingChurn(side: "prefix" | "suffix"): string { +function makeDistantStreamingChurn(side: "prefix" | "suffix", additionalChangeLine?: number): string { return Array.from({ length: 3 }, (_, fileIndex) => { const lines = [` export function ordinary_${side}_${fileIndex}(actor: Actor) {`]; for (let line = 2; line < 130; line += 1) { - if (line === 64) { + if (line === 64 || line === additionalChangeLine) { lines.push( `- const ordinary_${side}_${fileIndex}_${line}=actor.id;`, `+ const ordinary_${side}_${fileIndex}_${line} = actor.id;`, @@ -1109,7 +1109,7 @@ const fixtureSpecs: FixtureSpec[] = [ "export async function bulkEdit(actor: Actor, changeSet: ChangeSet) { await applyBulkEdit(changeSet); }", policy: "Huge diffs still need authorization checks on privileged writes.", scoringLabels: ["huge-low-signal", "multi-hunk", "security", "authorization", "error"], - prefixDiff: makeDistantStreamingChurn("prefix"), + prefixDiff: makeDistantStreamingChurn("prefix", 70), suffixDiff: makeDistantStreamingChurn("suffix"), expectedCoverage: "bounded", finding: { diff --git a/bench/src/compare-baseline.test.ts b/bench/src/compare-baseline.test.ts index 38764d4..c768422 100644 --- a/bench/src/compare-baseline.test.ts +++ b/bench/src/compare-baseline.test.ts @@ -1,4 +1,5 @@ import { describe, expect, test } from "bun:test"; +import { execFileSync } from "node:child_process"; import { createHash } from "node:crypto"; import { mkdir, mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; @@ -48,7 +49,13 @@ import { type ProviderContractEvidence, } from "./livemodels-score"; -test("EU baseline preserves its fixture identity but is stale for the calibration evaluator", async () => { +const historicalImport = "7ae07f4c2c62256d837860deb76e498706e3ac5b"; +const historicalArtifact = (path: string): string => execFileSync( + "git", ["show", `${historicalImport}:${path}`], + { cwd: resolve(import.meta.dir, "../.."), encoding: "utf8" }, +); + +test("retained EU baseline preserves historical identity and rejects the current corpus", async () => { const baseline = parseBaselineFile(JSON.parse( await readFile(resolve(import.meta.dir, "..", "baseline.json"), "utf8"), )); @@ -56,7 +63,9 @@ test("EU baseline preserves its fixture identity but is stale for the calibratio .update(JSON.stringify(cases.map((input) => benchmarkCase.parse(input)))) .digest("hex"); - expect(baseline.corpus.fixtureCorpusSha256).toBe(fixtureCorpusSha256); + const historical = parseBaselineFile(JSON.parse(historicalArtifact("bench/baseline.json"))); + expect(baseline.corpus).toEqual(historical.corpus); + expect(baseline.corpus.fixtureCorpusSha256).not.toBe(fixtureCorpusSha256); expect(baseline.corpus.evaluatorSha256).not.toBe(await evaluatorSourceSha256()); }); @@ -115,12 +124,12 @@ test("release prerequisite rejects an unpopulated US baseline before attestation } }); -test("attested US calibration binds its source and route while the embedded provider stays on Azure/EU", async () => { +test("retained attested US calibration binds its source and route while the embedded provider stays on Azure/EU", async () => { const root = resolve(import.meta.dir, "..", ".."); const [usBytes, euBytes, attestationBytes, us, eu] = await Promise.all([ - readFile(resolve(root, "bench/baseline-us.json"), "utf8"), - readFile(resolve(root, "bench/baseline.json"), "utf8"), - readFile(resolve(root, "bench/baseline-us.attestation.json"), "utf8"), + historicalArtifact("bench/baseline-us.json"), + historicalArtifact("bench/baseline.json"), + historicalArtifact("bench/baseline-us.attestation.json"), screeningProfileMetadata(resolve(root, "provisional-models-us.json")), screeningProfileMetadata(resolve(root, "provisional-models-eu.json")), ]); @@ -163,7 +172,7 @@ test("attested US calibration binds its source and route while the embedded prov expect(await readFile(resolve(root, "provisional-models.json"), "utf8")).toBe(await readFile(resolve(root, "provisional-models-eu.json"), "utf8")); expect(usBaseline.corpus.fixtureCorpusSha256).toBe(euBaseline.corpus.fixtureCorpusSha256); expect(usBaseline.corpus.evaluatorSha256).not.toBe(euBaseline.corpus.evaluatorSha256); - expect(usBaseline.corpus.evaluatorSha256).toBe(await evaluatorSourceSha256()); + expect(usBaseline.corpus.evaluatorSha256).not.toBe(await evaluatorSourceSha256()); expect(eu.upstreamProviderRoute).toBe("azure/eu"); expect(us.upstreamProviderRoute).toBe("azure/us"); for (const profile of [us, eu]) { @@ -322,7 +331,7 @@ function cloneReport(report: LiveReportForComparison): LiveReportForComparison { return structuredClone(report); } -async function inputBoundReport(): Promise { +async function inputBoundReport(screeningProfilePath = resolve(import.meta.dir, "..", "..", "provisional-models.json")): Promise { const report = fakeReport(); const parsedCases = cases.map((input) => benchmarkCase.parse(input)); let detected = 0; @@ -346,7 +355,6 @@ async function inputBoundReport(): Promise { exitCode: caseDetected && finding?.severity === "error" ? 1 : 0, }; }); - const screeningProfilePath = resolve(import.meta.dir, "..", "..", "provisional-models.json"); const [binary, evaluatorSha256, profile] = await Promise.all([ readFile(process.execPath), evaluatorSourceSha256(), @@ -637,6 +645,77 @@ function fakeReleaseCohort(reports: readonly LiveReportForComparison[]): { }; } +test("recording replaces a populated historical baseline only after complete current-input cohort verification", async () => { + const directory = await mkdtemp(resolve(tmpdir(), "postil-populated-record-")); + try { + const profilePath = resolve(import.meta.dir, "../..", "provisional-models-us.json"); + const original = await inputBoundReport(profilePath); + const reports = Array.from({ length: 10 }, (_, index) => { + const report = cloneReport(original); + report.summary.runId = `recalibration-${index + 1}`; + report.summary.ranAt = `2026-08-25T00:10:0${index}.000Z`; + report.summary.providerGenerationIds = [`gen-recalibration-${index + 1}`]; + return report; + }); + const cohort = fakeReleaseCohort(reports); + for (const field of ["binarySha256", "evaluatorSha256", "fixtureCorpusSha256", "screeningProfileSha256", "providerContractSha256"] as const) { + cohort.manifest[field] = original.summary[field]; + } + const manifestBytes = JSON.stringify(cohort.manifest); + const manifestPath = resolve(directory, "cohort.json"); + await writeFile(manifestPath, manifestBytes); + const baselinePath = resolve(directory, "baseline.json"); + const historical = historicalArtifact("bench/baseline-us.json"); + await writeFile(baselinePath, historical); + const arguments_: string[] = []; + for (const [index, report] of reports.entries()) { + const reportPath = resolve(directory, `report-${index}.json`); + const receiptPath = resolve(directory, `receipt-${index}.json`); + cohort.receipts[index]!.receipt.manifestSha256 = sha256(manifestBytes); + await writeFile(reportPath, JSON.stringify(report)); + await writeFile(receiptPath, JSON.stringify(cohort.receipts[index]!.receipt)); + arguments_.push("--expected-run-id", report.summary.runId, "--result", reportPath, "--receipt", receiptPath); + } + const run = async () => { + const execution = cohort.manifest.execution; + const child = Bun.spawn([ + process.execPath, "--no-env-file", resolve(import.meta.dir, "compare-baseline.ts"), + "--binary", process.execPath, "--baseline", baselinePath, "--screen-profile", profilePath, + "--cohort-manifest", manifestPath, ...arguments_, "--record", + ], { stdout: "pipe", stderr: "pipe", timeout: 10_000, env: { + GITHUB_REPOSITORY: execution.repository, GITHUB_SHA: execution.sourceSha, + GITHUB_REF: execution.sourceRef, GITHUB_RUN_ID: execution.runId, GITHUB_RUN_ATTEMPT: execution.runAttempt, + } }); + const [code, stdout, stderr] = await Promise.all([ + child.exited, new Response(child.stdout).text(), new Response(child.stderr).text(), + ]); + return { code, output: stdout + stderr }; + }; + const receiptPath = resolve(directory, "receipt-9.json"); + const receiptBytes = await readFile(receiptPath); + await writeFile(receiptPath, JSON.stringify({ ...cohort.receipts[9]!.receipt, state: "failed", exitCode: 1 })); + expect((await run()).code).not.toBe(0); + expect(await readFile(baselinePath, "utf8")).toBe(historical); + await writeFile(receiptPath, receiptBytes); + const recorded = await run(); + expect(recorded.code).toBe(0); + expect(recorded.output).toContain("Recorded baseline"); + const baseline = parseBaselineFile(JSON.parse(await readFile(baselinePath, "utf8"))); + expect(baseline.corpus.evaluatorSha256).toBe(original.summary.evaluatorSha256); + expect(baseline.corpus.fixtureCorpusSha256).toBe(original.summary.fixtureCorpusSha256); + const profile = baseline.profiles[original.summary.model]!; + expect(profile.populated).toBe(true); + if (!isCalibratedBaselineProfile(profile)) throw new Error("recorded calibration is missing"); + expect(profile.calibration.reportCount).toBe(10); + expect(profile.calibration.sourceSha).toBe(cohort.manifest.execution.sourceSha); + for (const [name, prior] of Object.entries(JSON.parse(historical).profiles)) { + if (name !== original.summary.model) expect(baseline.profiles[name]).toEqual(prior); + } + } finally { + await rm(directory, { recursive: true, force: true }); + } +}); + describe("predeclared cohort evidence", () => { test("accepts every completed immutable slot", () => { const reports = Array.from({ length: 5 }, (_, index) => fakeReport({ diff --git a/bench/src/scorer-eval.test.ts b/bench/src/scorer-eval.test.ts index 28acfe3..7be962f 100644 --- a/bench/src/scorer-eval.test.ts +++ b/bench/src/scorer-eval.test.ts @@ -378,6 +378,8 @@ function genericOpenAiCompatibleRequest( }; } +const ADJUDICATION_CORRECTION_SUFFIX = "\n\nReturn the complete adjudication JSON array for the original candidate IDs. Correct the rejected output using only the supplied evidence. A refutation needs candidate-specific contradictory evidence, not its own citation or unsupported prose. A context-anchored confirmation needs an exact causal change reference. Preserve all structural, scope and publication rules. Never invent evidence or references, silently drop a candidate, or fabricate a verdict."; + function adjudicationResponse(body: string): string | null { let request: { messages?: Array<{ role?: string; content?: string }>; @@ -392,7 +394,9 @@ function adjudicationResponse(body: string): string | null { ) ?? false; if (!isAdjudication) return null; const user = request.messages?.findLast((message) => message.role === "user")?.content ?? "{}"; - const payload = JSON.parse(user) as { + const originalUser = user.endsWith(ADJUDICATION_CORRECTION_SUFFIX) + ? user.slice(0, -ADJUDICATION_CORRECTION_SUFFIX.length) : user; + const payload = JSON.parse(originalUser) as { candidates?: Array<{ candidateId?: string; title?: string; @@ -2387,13 +2391,49 @@ describe("finite adjudication provenance", () => { expect(adjudicationOutput(events.map(provenanceLine).join("\n"))?.records).toEqual(events); }); + test("mock accepts only original JSON or its exact correction suffix", () => { + const request = (user: string) => JSON.stringify({ messages: [ + { role: "system", content: "Postil's single finding adjudicator" }, { role: "user", content: user }, + ] }); + expect(adjudicationResponse(request('{"candidates":[]}'))).toBe("[]"); + expect(adjudicationResponse(request('{"candidates":[]}' + ADJUDICATION_CORRECTION_SUFFIX))).toBe("[]"); + for (const suffix of [" arbitrary", ADJUDICATION_CORRECTION_SUFFIX + " extra", ADJUDICATION_CORRECTION_SUFFIX.repeat(2)]) { + expect(() => adjudicationResponse(request('{"candidates":[]}' + suffix))).toThrow(); + } + }); + + test("accepts exactly the bounded call phases alongside legacy records", () => { + const phased = (call: number, callPhase: unknown) => { + const { attempt, call: _, ...rest } = provenanceRecord(); + return { attempt, call, callPhase, ...rest }; + }; + for (const record of [phased(1, "initial"), phased(2, "schema-repair"), phased(2, "semantic-retry")]) { + const line = provenanceLine(record); + expect(adjudicationOutput(line)).toEqual({ records: [record], malformed: 0, duplicates: 0, omitted: 0 }); + for (const bad of [ + line.replace('"callPhase":', '"callPhase":"initial","callPhase":'), + line.replace('"version":1', '"other":0,"version":1'), + ]) expect(adjudicationOutput(bad)?.malformed).toBe(1); + } + for (const record of [phased(1, "semantic-retry"), phased(1, "schema-repair"), phased(2, "initial"), + phased(3, "semantic-retry"), phased(2, null), phased(1, "unknown"), phased(1, {}), + provenanceRecord({ call: 2 }), provenanceRecord({ callPhase: undefined, call: 2 })]) { + expect(adjudicationOutput(provenanceLine(record))?.records).toEqual([]); + expect(adjudicationOutput(provenanceLine(record))?.malformed).toBe(1); + } + const line = provenanceLine(phased(2, "semantic-retry")); + expect(adjudicationOutput(`${line}\n${line}`)).toMatchObject({ duplicates: 1, malformed: 0 }); + expect(adjudicationOutput(`${line}\n`.repeat(40))).toMatchObject({ duplicates: 31, omitted: 8, malformed: 0 }); + }); + test("real child preserves baseline behavior with enabled counted scorer and mock publication", async () => { const baseBinary = process.env.POSTIL_PROVENANCE_BASE_BIN; const binaries = baseBinary ? [baseBinary, postilBinaryPath()] : [postilBinaryPath()]; const hostile = String.fromCharCode(115, 107, 45) + crypto.randomUUID(); const modes = ["accepted", "fenced", "noArray", "unclosedArray", "jsonSyntax", "resultCountMismatch", "missingKnownField", "unknownField", "invalidEnum", "endpointMalformed", "emptyRecovered", - "missingChoicesRecovered", "emptyTerminal", "length", "tool_calls", "absentFinish"]; + "missingChoicesRecovered", "emptyTerminal", "length", "tool_calls", "absentFinish", + "unknownRecovered", "unknownTerminal", "syntaxThenUnknown"]; const keyName = "POSTIL_PROVENANCE_LOCAL_KEY"; const previous = process.env[keyName]; process.env[keyName] = crypto.randomUUID(); @@ -2417,6 +2457,11 @@ describe("finite adjudication provenance", () => { if (mode === "missingChoicesRecovered" && adjudicationCalls === 1) { res.writeHead(200); res.end(JSON.stringify({ usage: { prompt_tokens: 30, completion_tokens: 10, cost: 0.000045 } })); return; } + if (mode === "unknownTerminal" || mode === "syntaxThenUnknown" && adjudicationCalls === 2 || + mode === "unknownRecovered" && adjudicationCalls === 1) { + content = JSON.stringify([{ ...JSON.parse(adjudication)[0], candidateId: "unknown-candidate" }]); + } + if (mode === "syntaxThenUnknown" && adjudicationCalls === 1) content = "no array"; if (mode === "noArray") content = `refusal ${hostile} https://example.invalid/${hostile}`; if (mode === "unclosedArray") content = "["; if (mode === "jsonSyntax") content = "[?]"; @@ -2446,7 +2491,9 @@ describe("finite adjudication provenance", () => { binary, runRoot, apiBase, keyName, pricing); const artifacts = join(runRoot, safeSegment(TEST_SCORER_MODEL), "repeat-1", c.id, "artifacts"); const envelope = envelopeV1.parse(JSON.parse(await readFile(join(artifacts, "stdout.json"), "utf8"))); - const success = ["accepted", "fenced", "emptyRecovered", "missingChoicesRecovered"].includes(mode); + const success = ["accepted", "fenced", "emptyRecovered", "missingChoicesRecovered", "unknownRecovered"].includes(mode); + const corrected = ["noArray", "unclosedArray", "jsonSyntax", "resultCountMismatch", + "missingKnownField", "unknownField", "invalidEnum", "unknownRecovered", "unknownTerminal", "syntaxThenUnknown"].includes(mode); expect(evaluation.envelopeProduced).toBe(true); expect(envelope.findings.some((finding) => finding.path === ".postil/model-output")).toBe(!success); const publication = JSON.parse(await readFile(join(artifacts, "publication-telemetry.json"), "utf8")); @@ -2464,8 +2511,40 @@ describe("finite adjudication provenance", () => { else { expect(projection?.malformed).toBe(0); expect(projection?.records[0]?.event).toBe("callStarted"); - expect(projection?.records.every((record) => record.call === 1)).toBe(true); - const expectedCode = success ? "accepted" : + const records = projection!.records; + expect(projection?.duplicates).toBe(0); + expect(projection?.omitted).toBe(0); + const phase = ["unknownRecovered", "unknownTerminal"].includes(mode) ? "semantic-retry" : "schema-repair"; + expect(records.filter((record) => record.event === "callStarted") + .map(({ call, callPhase }) => ({ call, callPhase }))).toEqual(corrected + ? [{ call: 1, callPhase: "initial" }, { call: 2, callPhase: phase }] + : [{ call: 1, callPhase: "initial" }]); + expect(records.every((record) => record.call === 1 ? record.callPhase === "initial" + : corrected && record.call === 2 && record.callPhase === phase)).toBe(true); + const transportRecovered = ["emptyRecovered", "missingChoicesRecovered"].includes(mode); + const expectedCalls = corrected || transportRecovered || mode === "emptyTerminal" ? 2 : 1; + expect(adjudicationCalls).toBe(expectedCalls); + const adjudicationRequests = requests.map((body) => JSON.parse(body)) + .filter((body) => body.messages.some((message: { content?: string }) => + message.content?.includes("Postil's single finding adjudicator"))); + expect(adjudicationRequests.length).toBe(expectedCalls); + for (const request of adjudicationRequests.slice(1)) { + const expected = structuredClone(adjudicationRequests[0]); + if (corrected) expected.messages.findLast((message: { role?: string }) => message.role === "user").content += ADJUDICATION_CORRECTION_SUFFIX; + expect(request).toEqual(expected); + } + expect(evaluation.diagnostics?.adjudication.attempts).toBe(expectedCalls); + expect(evaluation.diagnostics?.adjudication.validUsageAttempts).toBe(mode === "endpointMalformed" ? 0 : expectedCalls); + expect(evaluation.diagnostics?.adjudication.exactCostAttempts).toBe(mode === "endpointMalformed" ? 0 : expectedCalls); + const usage = envelope.modelUsage?.filter((usage) => usage.role === "findingScorer") ?? []; + expect(usage.length).toBe(expectedCalls + scorerCalls); + if (mode !== "endpointMalformed") { + expect(usage.every((entry) => entry.accountingComplete)).toBe(true); + expect(usage.reduce((sum, entry) => sum + entry.promptTokens, 0)).toBe(30 * (expectedCalls + scorerCalls)); + expect(usage.reduce((sum, entry) => sum + entry.completionTokens, 0)).toBe(10 * (expectedCalls + scorerCalls)); + expect(usage.reduce((sum, entry) => sum + entry.costMicros!, 0)).toBe(45 * (expectedCalls + scorerCalls)); + } + const expectedCode = success || ["unknownTerminal", "syntaxThenUnknown"].includes(mode) ? "accepted" : mode === "emptyTerminal" ? "emptyContent" : ["length", "tool_calls", "absentFinish"].includes(mode) ? "nonterminalStop" : mode; expect(projection?.records.at(-1)?.code).toBe(expectedCode); @@ -2473,7 +2552,7 @@ describe("finite adjudication provenance", () => { expect(projection?.records.at(-1)?.finish).toBe(mode === "absentFinish" ? "absent" : mode); expect(adjudicationCalls).toBe(1); } - expect(projection?.records.at(-1)?.event).toBe(success ? "parsed" : + expect(projection?.records.at(-1)?.event).toBe(success || ["unknownTerminal", "syntaxThenUnknown"].includes(mode) ? "parsed" : ["endpointMalformed", "emptyTerminal", "length", "tool_calls", "absentFinish"].includes(mode) ? "callFailed" : "parserRejected"); expect(JSON.stringify(projection).includes(hostile)).toBe(false); if (mode === "emptyRecovered" || mode === "missingChoicesRecovered") { diff --git a/bench/src/scorer-eval.ts b/bench/src/scorer-eval.ts index 1676609..72f6247 100644 --- a/bench/src/scorer-eval.ts +++ b/bench/src/scorer-eval.ts @@ -237,7 +237,8 @@ const OUTPUT_CODES = { type OutputCount = number | "overflow" | null; export interface AdjudicationOutputRecord { attempt: number | null; // Local to this adjudication call, never a proxy ordinal. - call: 1; // One adjudication call per review child; not a proxy ordinal. + call: 1 | 2; // Initial call or its one shared correction; not a proxy ordinal. + callPhase?: "initial" | "schema-repair" | "semantic-retry"; code: string; event: keyof typeof OUTPUT_CODES; expected: OutputCount; @@ -276,10 +277,13 @@ export function adjudicationOutput(stderr: string): AdjudicationOutput | undefin try { const value = JSON.parse(raw); if (!value || typeof value !== "object" || Array.isArray(value)) throw 0; - const { attempt, call, code, event, expected, field, finish, observed, phase, stage, version } = value; + const { attempt, call, callPhase, code, event, expected, field, finish, observed, phase, stage, version } = value; + const phased = Object.hasOwn(value, "callPhase"); + if (phased ? !(call === 1 && callPhase === "initial" || + call === 2 && (callPhase === "schema-repair" || callPhase === "semantic-retry")) : call !== 1) throw 0; if (version !== 1 || phase !== "adjudication" || typeof event !== "string" || typeof stage !== "string" || typeof code !== "string" || !Object.hasOwn(OUTPUT_CODES, event) || - call !== 1 || !ordinal(attempt, 13) || !count(expected) || !count(observed) || + !ordinal(attempt, 13) || !count(expected) || !count(observed) || !(field === null || OUTPUT_FIELDS.includes(field)) || !(finish === null || ["stop", "length", "tool_calls", "content_filter", "error", "reported", "absent", "end_turn", "stop_sequence"].includes(finish))) throw 0; const stages = OUTPUT_CODES[event as keyof typeof OUTPUT_CODES] as Record; @@ -299,7 +303,7 @@ export function adjudicationOutput(stderr: string): AdjudicationOutput | undefin if (code !== "nonterminalStop" && finish !== null) throw 0; if (code === "nonterminalStop" && finish === null) throw 0; const record: AdjudicationOutputRecord = { - attempt, call, code, event, expected, field, finish, observed, phase, stage, version, + attempt, call, ...(phased ? { callPhase } : {}), code, event, expected, field, finish, observed, phase, stage, version, }; if (JSON.stringify(record) !== raw) throw 0; if (accepted.has(raw)) duplicates = boundedIncrement(duplicates); diff --git a/bench/src/verify-calibration.test.ts b/bench/src/verify-calibration.test.ts index b013aa3..c021f93 100644 --- a/bench/src/verify-calibration.test.ts +++ b/bench/src/verify-calibration.test.ts @@ -1,13 +1,17 @@ import { afterEach, expect, test } from "bun:test"; import { execFileSync } from "node:child_process"; import { randomBytes } from "node:crypto"; -import { cp, mkdir, mkdtemp, readFile, rm, symlink, writeFile } from "node:fs/promises"; +import { chmod, cp, mkdir, mkdtemp, readFile, rm, symlink, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { dirname, resolve } from "node:path"; import { evaluatorSourceSha256 } from "./live"; import { calibrationMain, validateCalibrationDispatch, verifyLegacyCalibration } from "./verify-calibration"; const root = resolve(import.meta.dir, "../.."); +const historicalSource = "356ba691999c737551506039d2ad4bc26c6c1354"; +const historicalImport = "7ae07f4c2c62256d837860deb76e498706e3ac5b"; +const historicalBytes = (source: string, path: string) => + execFileSync("git", ["show", `${source}:${path}`], { cwd: root }); const temporary: string[] = []; const candidate = randomBytes(20).toString("hex"); const admission = { @@ -27,13 +31,15 @@ async function fixture() { const paths: string[] = JSON.parse(await readFile(resolve(root, "bench/evaluator-contract-sources.json"), "utf8")); for (const path of [...paths, "bench/baseline-us.json", "bench/baseline-us.attestation.json", "provisional-models-us.json"]) { await mkdir(dirname(resolve(directory, path)), { recursive: true }); - await cp(resolve(root, path), resolve(directory, path)); + const source = paths.includes(path) ? historicalSource : historicalImport; + await writeFile(resolve(directory, path), historicalBytes(source, path)); } return directory; } async function transport() { - const bundle = JSON.parse(await readFile(resolve(root, "bench/baseline-us.attestation.json"), "utf8")); + const historicalRoot = await fixture(); + const bundle = JSON.parse(await readFile(resolve(historicalRoot, "bench/baseline-us.attestation.json"), "utf8")); const statement = JSON.parse(Buffer.from(bundle.dsseEnvelope.payload, "base64").toString("utf8")); const calls: string[][] = []; const runGh = async (args: string[]) => { @@ -45,7 +51,7 @@ async function transport() { } }]); }; const runGit = async (args: string[]) => execFileSync("git", args, { cwd: root }); - return { runGh, runGit, calls, statement }; + return { root: historicalRoot, runGh, runGit, calls, statement }; } test("default admission retains all four effective required inputs without granting calibration authority", async () => { @@ -82,7 +88,8 @@ test("frozen historical evidence stays legacy-only and pins the actual signature const context = await transport(); const result = await verifyLegacyCalibration(context); expect(result).toMatchObject({ protocol: "legacy-v2", behaviorCompatibility: "UNKNOWN", observedRuntimeIdentity: "UNKNOWN", executionEnabled: false }); - expect(result.evaluatorSha256).toBe(await evaluatorSourceSha256()); + const baseline = JSON.parse(await readFile(resolve(context.root, "bench/baseline-us.json"), "utf8")); + expect(result.evaluatorSha256).toBe(baseline.corpus.evaluatorSha256); expect(result.workflowRunId).toBe("35951227818"); expect(result).not.toHaveProperty("effectiveConcurrency"); expect(result).not.toHaveProperty("behaviorSha256"); @@ -97,6 +104,127 @@ test("frozen historical evidence stays legacy-only and pins the actual signature expect(verification).toContain("--deny-self-hosted-runners"); }); +test("current evaluator sources cannot reuse the historical calibration", async () => { + const context = await transport(); + const paths: string[] = JSON.parse(await readFile(resolve(root, "bench/evaluator-contract-sources.json"), "utf8")); + for (const path of paths) await cp(resolve(root, path), resolve(context.root, path)); + const baseline = JSON.parse(await readFile(resolve(context.root, "bench/baseline-us.json"), "utf8")); + expect(await evaluatorSourceSha256()).not.toBe(baseline.corpus.evaluatorSha256); + await expect(verifyLegacyCalibration(context)).rejects.toThrow("evaluator is stale"); + expect(context.calls).toHaveLength(0); +}); + +test("calibration workflow verifies retained provenance before reserving a changed current-main evaluator", async () => { + const workflow = Bun.YAML.parse(await readFile(resolve(root, ".github/workflows/benchmark-calibration.yml"), "utf8")) as { + jobs: { validate: { steps: { name?: string; run?: string }[] } }; + }; + const steps = workflow.jobs.validate.steps; + const script = (name: string) => { + const step = steps.find(value => value.name === name); + expect(step?.run).toBeDefined(); + return step!.run!; + }; + const initial = script("Verify the current main calibration source"); + const validation = script("Verify the calibration baseline input"); + const reservation = script("Reserve the current main calibration source"); + expect(steps.findIndex(step => step.name === "Verify embedded model admission")).toBeLessThan( + steps.findIndex(step => step.name === "Verify the calibration baseline input"), + ); + expect(steps.findIndex(step => step.name === "Verify the calibration baseline input")).toBeLessThan( + steps.findIndex(step => step.name === "Reserve the current main calibration source"), + ); + const directory = await fixture(); + const baselinePath = resolve(directory, "bench/baseline-us.json"); + const baselineBytes = await readFile(baselinePath); + const baseline = JSON.parse(baselineBytes.toString()); + const source = baseline.profiles["openai/gpt-5.6-luna"].calibration.sourceSha; + const bin = resolve(directory, "bin"); + await mkdir(bin); + const callsPath = resolve(directory, "calls"); + const shim = resolve(bin, "gh"); + await writeFile(shim, `#!/bin/bash +set -euo pipefail +printf '%s\\t' "$@" >> "$CALLS_FILE" +printf '\\n' >> "$CALLS_FILE" +if [[ "$1" == api ]]; then + case "$*" in + *'--method POST'*) exit "$RESERVATION_EXIT" ;; + *'heads/main'*) printf '%s' "$MAIN_SHA" ;; + *) printf '%s' "$REGISTRY_SHA" ;; + esac +else + exit "$VERIFICATION_EXIT" +fi +`); + await chmod(shim, 0o700); + const currentEvaluator = await evaluatorSourceSha256(); + const run = async (options: Record = {}) => { + await writeFile(callsPath, ""); + const child = Bun.spawn(["bash", "-c", initial + "\n(\n" + validation + "\n)\n" + reservation], { + cwd: directory, stdout: "pipe", stderr: "pipe", timeout: 10_000, + env: { + PATH: `${bin}:${process.env.PATH}`, CALLS_FILE: callsPath, + GITHUB_REF: "refs/heads/main", GITHUB_RUN_ATTEMPT: "1", GITHUB_SHA: candidate, + GITHUB_REPOSITORY: "postil-dev/postil-cli", MAIN_SHA: candidate, + REGISTRY_SHA: source, VERIFICATION_EXIT: "0", RESERVATION_EXIT: "0", + CURRENT_EVALUATOR_SHA256: currentEvaluator, ...options, + }, + }); + const [code, stdout, stderr] = await Promise.all([ + child.exited, new Response(child.stdout).text(), new Response(child.stderr).text(), + ]); + const calls = (await readFile(callsPath, "utf8")).trim().split("\n").filter(Boolean).map(line => line.trim().split("\t")); + return { code, output: stdout + stderr, calls }; + }; + const accepted = await run(); + expect(accepted.code).toBe(0); + const verification = accepted.calls.find(args => args[0] === "attestation")!; + expect(verification).toEqual([ + "attestation", "verify", "bench/baseline-us.json", "--bundle", "bench/baseline-us.attestation.json", + "--repo", "postil-dev/postil-cli", "--signer-workflow", "postil-dev/postil-cli/.github/workflows/benchmark-calibration.yml", + "--signer-digest", source, "--source-digest", source, "--source-ref", "refs/heads/main", + "--cert-oidc-issuer", "https://token.actions.githubusercontent.com", "--predicate-type", "https://slsa.dev/provenance/v1", + "--deny-self-hosted-runners", "--format", "json", + ]); + expect(accepted.calls.at(-1)).toContain(`ref=refs/tags/postil-calibration-${candidate}`); + for (const [options, message] of [ + [{ GITHUB_REF: "refs/heads/topic" }, "only from main"], + [{ GITHUB_RUN_ATTEMPT: "2" }, "retries are not authoritative"], + [{ MAIN_SHA: source }, "not the current main commit"], + [{ GITHUB_SHA: source, MAIN_SHA: source }, "requires a new main source commit"], + [{ CURRENT_EVALUATOR_SHA256: baseline.corpus.evaluatorSha256 }, "changed evaluator contract"], + [{ CURRENT_EVALUATOR_SHA256: "invalid" }, "evaluator digest is invalid"], + [{ REGISTRY_SHA: candidate }, "registry source mismatch"], + [{ VERIFICATION_EXIT: "1" }, ""], + ] as const) { + const refused = await run(options); + expect(refused.code).not.toBe(0); + expect(refused.output).toContain(message); + expect(refused.calls.some(args => args.includes("POST"))).toBe(false); + } + expect((await run({ RESERVATION_EXIT: "1" })).code).not.toBe(0); + for (const update of [ + { profiles: {} }, + { profiles: { "openai/gpt-5.6-luna": { populated: "false" } } }, + { profiles: { "openai/gpt-5.6-luna": { populated: true, calibration: { sourceSha: "invalid" } } } }, + { ...baseline, corpus: { evaluatorSha256: "invalid" } }, + ]) { + await writeFile(baselinePath, JSON.stringify(update)); + const refused = await run(); + expect(refused.code).not.toBe(0); + expect(refused.calls.some(args => args.includes("POST"))).toBe(false); + } + await writeFile(baselinePath, baselineBytes); + await writeFile(resolve(directory, "bench/baseline-us.attestation.json"), ""); + const missingBundle = await run(); + expect(missingBundle.output).toContain("requires the retained baseline attestation"); + expect(missingBundle.calls.some(args => args.includes("POST"))).toBe(false); + await writeFile(baselinePath, JSON.stringify({ profiles: { "openai/gpt-5.6-luna": { populated: false } } })); + const bootstrap = await run(); + expect(bootstrap.code).toBe(0); + expect(bootstrap.calls.some(args => args[0] === "attestation")).toBe(false); +}); + test("actual evaluator, effective-launch and workflow mutations invalidate legacy freshness", async () => { const directory = await fixture(); const context = await transport(); diff --git a/src/adjudication.rs b/src/adjudication.rs index 1da2673..ac7e5ef 100644 --- a/src/adjudication.rs +++ b/src/adjudication.rs @@ -14,6 +14,29 @@ pub(crate) const MAX_ADJUDICATION_CANDIDATES: usize = 20; pub(crate) const MAX_ADJUDICATION_CORPUS_BYTES: usize = 24 * 1024; pub(crate) const MAX_ADJUDICATION_PROMPT_BYTES: usize = 48 * 1024; pub(crate) const MAX_ADJUDICATION_OUTPUT_TOKENS: u32 = 8_000; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum AdjudicationViolation { + UnknownCandidateIdentity, + ContradictoryRefutation, + MissingCausalReference, +} + +impl std::fmt::Display for AdjudicationViolation { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(match self { + Self::UnknownCandidateIdentity => "adjudication returned an unknown candidate identity", + Self::ContradictoryRefutation => { + "refuted adjudication must cite candidate-specific contradictory evidence" + } + Self::MissingCausalReference => { + "context-anchored confirmation requires a causal change reference" + } + }) + } +} + +impl std::error::Error for AdjudicationViolation {} const MAX_DIRECT_EVIDENCE_QUERIES: usize = 128; const MAX_DIRECT_EVIDENCE_QUERY_BYTES: usize = 8 * 1024; const MAX_CITED_EVIDENCE_BYTES: usize = 1_024; @@ -1179,7 +1202,7 @@ fn validate_result_structure( for result in results { ensure!( expected.contains(&result.candidate_id), - "adjudication returned an unknown candidate identity" + AdjudicationViolation::UnknownCandidateIdentity ); ensure!( seen.insert(result.candidate_id.clone()), @@ -1357,7 +1380,7 @@ pub(crate) fn validate_results( ); ensure!( refutation_grounded, - "refuted adjudication must cite candidate-specific contradictory evidence" + AdjudicationViolation::ContradictoryRefutation ); } AdjudicationStatus::Unresolved => ensure!( @@ -1758,9 +1781,7 @@ fn validate_scopes( evidence: evidence.to_string(), }) } else { - return Err(anyhow!( - "context-anchored confirmation requires a causal change reference" - )); + return Err(AdjudicationViolation::MissingCausalReference.into()); }; scopes.insert( result.candidate_id.clone(), diff --git a/src/diff.rs b/src/diff.rs index bd39119..22caab1 100644 --- a/src/diff.rs +++ b/src/diff.rs @@ -1439,6 +1439,8 @@ impl ModelBatchSpool { &mut self, selected_limit: usize, ) -> Result { + #[cfg(test)] + tests::SOURCE_PACKING_RECEIPTS.with(|receipts| receipts.set(receipts.get() + 1)); anyhow::ensure!( selected_limit > 0, "bounded review batch limit must be positive" @@ -2479,7 +2481,126 @@ pub(crate) fn spool_model_batches_with_synthesis_budget( max_manifest_bytes: usize, force_empty: bool, semantic_receipt_limit: usize, +) -> Result { + spool_model_batches_guarded( + prepared, + max_source_batch_bytes, + max_synthesis_batch_bytes, + max_manifest_bytes, + force_empty, + semantic_receipt_limit, + None, + ) +} + +pub(crate) fn spool_model_batches_for_review( + prepared: &mut PreparedReview, + max_source_batch_bytes: usize, + max_synthesis_batch_bytes: usize, + max_manifest_bytes: usize, + force_empty: bool, + semantic_receipt_limit: usize, + bounded: bool, +) -> Result { + spool_model_batches_guarded( + prepared, + max_source_batch_bytes, + max_synthesis_batch_bytes, + max_manifest_bytes, + force_empty, + semantic_receipt_limit, + Some(bounded), + ) +} + +fn spool_model_batches_guarded( + prepared: &mut PreparedReview, + max_source_batch_bytes: usize, + max_synthesis_batch_bytes: usize, + max_manifest_bytes: usize, + force_empty: bool, + semantic_receipt_limit: usize, + bounded: Option, ) -> Result { + anyhow::ensure!( + semantic_receipt_limit > 0, + "bounded review batch limit must be positive" + ); + let mut candidate = spool_model_batches_with_packing( + prepared, + max_source_batch_bytes, + max_synthesis_batch_bytes, + max_manifest_bytes, + force_empty, + semantic_receipt_limit, + true, + )?; + // Mirror the caller's route boundaries. A provider's five-batch choice is + // outside the deterministic coverage proof, so it keeps next-fit inventory. + let provider_selection = bounded.is_some_and(|bounded| { + candidate.count <= semantic_receipt_limit + && !(crate::config::hosted_runtime_mode() + && !bounded + && candidate.source_count > crate::review::MAX_HOSTED_SELECTED_BATCHES) + && (bounded || crate::config::bounded_review_selection_mode()) + && candidate.count > crate::review::MAX_HOSTED_SELECTED_BATCHES + }); + // Mandatory hunks select their materialized source batches first. When all + // source batches fit, their complete ownership fits without a guard receipt. + if !provider_selection && candidate.source_count <= semantic_receipt_limit { + candidate + .file + .seek(SeekFrom::Start(0)) + .context("rewinding capacity-packed model batches")?; + return Ok(candidate); + } + // Releasing both the file and its quota lease precedes the baseline build. + // Return the complete baseline inventory, including its semantic evidence. + drop(candidate); + spool_model_batches_next_fit( + prepared, + max_source_batch_bytes, + max_synthesis_batch_bytes, + max_manifest_bytes, + force_empty, + semantic_receipt_limit, + ) +} + +pub(crate) fn spool_model_batches_next_fit( + prepared: &mut PreparedReview, + max_source_batch_bytes: usize, + max_synthesis_batch_bytes: usize, + max_manifest_bytes: usize, + force_empty: bool, + semantic_receipt_limit: usize, +) -> Result { + spool_model_batches_with_packing( + prepared, + max_source_batch_bytes, + max_synthesis_batch_bytes, + max_manifest_bytes, + force_empty, + semantic_receipt_limit, + false, + ) +} + +fn spool_model_batches_with_packing( + prepared: &mut PreparedReview, + max_source_batch_bytes: usize, + max_synthesis_batch_bytes: usize, + max_manifest_bytes: usize, + force_empty: bool, + semantic_receipt_limit: usize, + reuse_capacity: bool, +) -> Result { + #[cfg(test)] + tests::SOURCE_PACKING_BUILDS.with(|builds| { + builds + .borrow_mut() + .push(prepared._window_lease.budget.retained_bytes()); + }); anyhow::ensure!( max_source_batch_bytes >= MIN_REVIEW_BATCH_BYTES, "source batch limit must leave room for context" @@ -2541,12 +2662,13 @@ pub(crate) fn spool_model_batches_with_synthesis_budget( } } } - let plan = render_review_batches( + let plan = render_review_batches_with_packing( &parsed, &[], &[], max_source_batch_bytes, max_manifest_bytes, + reuse_capacity, ); anyhow::ensure!( !plan.incomplete, @@ -2636,12 +2758,13 @@ pub(crate) fn spool_model_batches_with_synthesis_budget( { let lock_end = (lock_start + 16).min(prepared.lockfiles.len()); let artifact_end = (artifact_start + 16).min(prepared.compacted_artifacts.len()); - let plan = render_review_batches( + let plan = render_review_batches_with_packing( &Diff::default(), &prepared.lockfiles[lock_start..lock_end], &prepared.compacted_artifacts[artifact_start..artifact_end], max_source_batch_bytes, max_manifest_bytes, + reuse_capacity, ); if plan.incomplete { anyhow::bail!("compact artifact metadata could not fit a bounded model request"); @@ -5011,6 +5134,24 @@ pub fn render_review_batches( compacted_artifacts: &[CompactedArtifactEvidence], max_bytes: usize, max_manifest_bytes: usize, +) -> ReviewBatchPlan { + render_review_batches_with_packing( + diff, + lockfiles, + compacted_artifacts, + max_bytes, + max_manifest_bytes, + true, + ) +} + +fn render_review_batches_with_packing( + diff: &Diff, + lockfiles: &[LockfileEvidence], + compacted_artifacts: &[CompactedArtifactEvidence], + max_bytes: usize, + max_manifest_bytes: usize, + reuse_capacity: bool, ) -> ReviewBatchPlan { assert!( max_bytes >= MIN_REVIEW_BATCH_BYTES, @@ -5027,8 +5168,8 @@ pub fn render_review_batches( return plan; } - let mut current = String::new(); - let mut current_hunks = BTreeSet::new(); + let mut packing = SourceBatchPacking::new(&manifest.text); + packing.reuse_capacity = reuse_capacity; for file in &diff.files { if file.binary || file.hunks.is_empty() { continue; @@ -5044,24 +5185,16 @@ pub fn render_review_batches( return plan; }; for unit in units { - if !append_unit( - &mut plan, - &mut current, - &mut current_hunks, - &manifest.text, - &unit, - &hunk_identity, - max_bytes, - ) { + if !append_unit(&mut plan, &mut packing, &unit, &hunk_identity, max_bytes) { plan.incomplete = true; return plan; } } } } - if !current.is_empty() { - plan.batches.push(current); - plan.batch_hunks.push(current_hunks); + if !packing.current.is_empty() { + plan.batches.push(packing.current); + plan.batch_hunks.push(packing.current_hunks); } else if plan.batches.is_empty() && (!diff.files.is_empty() || !lockfiles.is_empty() || !compacted_artifacts.is_empty()) { @@ -5430,29 +5563,69 @@ fn render_line_segments_with_budget( rendered } +struct SourceBatchPacking { + reuse_capacity: bool, + initial: String, + initial_bytes: usize, + current: String, + current_hunks: BTreeSet, + current_bytes: usize, + closed_bytes: Vec, +} + +impl SourceBatchPacking { + fn new(manifest: &str) -> Self { + let initial = format!("{manifest}\n"); + let initial_bytes = json_string_content_bytes(&initial); + Self { + reuse_capacity: true, + initial, + initial_bytes, + current: String::new(), + current_hunks: BTreeSet::new(), + current_bytes: 0, + closed_bytes: Vec::new(), + } + } +} + fn append_unit( plan: &mut ReviewBatchPlan, - current: &mut String, - current_hunks: &mut BTreeSet, - manifest: &str, + packing: &mut SourceBatchPacking, unit: &str, hunk: &HunkIdentity, max_bytes: usize, ) -> bool { - let initial = format!("{manifest}\n"); - if json_string_content_bytes(&initial) + json_string_content_bytes(unit) > max_bytes { + let unit_bytes = json_string_content_bytes(unit); + if packing.initial_bytes + unit_bytes > max_bytes { return false; } - if current.is_empty() { - current.push_str(&initial); + if packing.current.is_empty() { + packing.current.push_str(&packing.initial); + packing.current_bytes = packing.initial_bytes; } - if json_string_content_bytes(current) + json_string_content_bytes(unit) > max_bytes { - plan.batches.push(std::mem::take(current)); - plan.batch_hunks.push(std::mem::take(current_hunks)); - current.push_str(&initial); + if packing.current_bytes + unit_bytes > max_bytes { + if packing.reuse_capacity + && let Some(index) = packing + .closed_bytes + .iter() + .position(|size| size + unit_bytes <= max_bytes) + { + plan.batches[index].push_str(unit); + plan.batch_hunks[index].insert(hunk.clone()); + packing.closed_bytes[index] += unit_bytes; + return true; + } + plan.batches.push(std::mem::take(&mut packing.current)); + plan.batch_hunks + .push(std::mem::take(&mut packing.current_hunks)); + packing.closed_bytes.push(packing.current_bytes); + packing.current.push_str(&packing.initial); + packing.current_bytes = packing.initial_bytes; } - current.push_str(unit); - current_hunks.insert(hunk.clone()); + packing.current.push_str(unit); + packing.current_hunks.insert(hunk.clone()); + packing.current_bytes += unit_bytes; true } @@ -6096,6 +6269,560 @@ pub fn render_hunk_context(diff: &Diff, path: &str, line: u32, radius: u32) -> O mod tests { use super::*; + thread_local! { + pub(super) static SOURCE_PACKING_BUILDS: std::cell::RefCell> = const { std::cell::RefCell::new(Vec::new()) }; + pub(super) static SOURCE_PACKING_RECEIPTS: std::cell::Cell = const { std::cell::Cell::new(0) }; + } + + fn capacity_coverage_fixture(budget: usize, prefix: usize) -> String { + let mut paths = (0..prefix) + .map(|i| format!("src/auth/{i:02}.rs")) + .chain(["src/a.rs", "src/b.rs", "src/auth/c.rs", "src/auth/d.rs"].map(String::from)) + .collect::>(); + let manifest = format!( + "Changed-file manifest:\n{}\n", + paths + .iter() + .map(|p| format!("- {p} [source]\n")) + .collect::() + ); + let padding = 792 - json_string_content_bytes(&manifest); + paths[0] = paths[0].replace(".rs", &format!("{}.rs", "x".repeat(padding))); + let available = budget - 792; + let tail = if budget == 32_768 { + [19_182, 22_379, 12_788, 9_591] + } else { + [ + available * 60 / 100, + available * 70 / 100, + available * 40 / 100, + available * 30 / 100, + ] + }; + let sizes = std::iter::repeat_n(available - 200, prefix).chain(tail); + let lines = if budget == 32_768 { 40 } else { 4 }; + let mut source = String::new(); + for (path, size) in paths.into_iter().zip(sizes) { + let header = format!("### {path}\n@@ segment starting near new line 1 @@\n"); + let overhead = json_string_content_bytes(&header) + + (1..=lines) + .map(|i| json_string_content_bytes(&format!("{i:>6} + \n"))) + .sum::(); + let payload = size - overhead; + source.push_str(&format!( + "diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n@@ -0,0 +1,{lines} @@\n" + )); + for i in 0..lines { + let width = payload / lines + usize::from(i < payload % lines); + assert!(width <= MAX_RENDERED_EVIDENCE_BYTES); + source.push_str(&format!("+{}\n", "x".repeat(width))); + } + } + source + } + + fn model_spool_bytes(spool: &mut ModelBatchSpool) -> Vec { + assert_eq!(spool.file.stream_position().unwrap(), 0); + let mut bytes = Vec::new(); + spool.file.read_to_end(&mut bytes).unwrap(); + spool.file.seek(SeekFrom::Start(0)).unwrap(); + bytes + } + + fn assert_same_model_spool( + left: &mut ModelBatchSpool, + right: &mut ModelBatchSpool, + limit: usize, + ) { + assert_eq!(model_spool_bytes(left), model_spool_bytes(right)); + assert_eq!( + (left.count, left.source_count, left.metadata_count), + (right.count, right.source_count, right.metadata_count) + ); + assert_eq!(left.batch_hunks, right.batch_hunks); + assert_eq!(left.all_hunks, right.all_hunks); + assert_eq!(left.synthesis_ids, right.synthesis_ids); + assert_eq!(left.exact_semantic_ids, right.exact_semantic_ids); + assert_eq!(left.metadata_batch_ids, right.metadata_batch_ids); + let risks = |s: &ModelBatchSpool| { + s.hunk_risk + .iter() + .map(|(h, r)| (h.clone(), (r.mandatory, r.semantic_eligible, r.score))) + .collect::>() + }; + assert_eq!(risks(left), risks(right)); + let receipt = left.deterministic_bounded_receipt(limit).unwrap(); + assert_eq!(receipt, left.deterministic_bounded_receipt(limit).unwrap()); + assert_eq!(receipt, right.deterministic_bounded_receipt(limit).unwrap()); + for _ in 0..left.count { + assert_eq!(left.next_batch().unwrap(), right.next_batch().unwrap()); + } + assert!(left.next_batch().unwrap().is_none()); + assert!(right.next_batch().unwrap().is_none()); + left.file.seek(SeekFrom::Start(0)).unwrap(); + right.file.seek(SeekFrom::Start(0)).unwrap(); + } + + fn mandatory_direct(spool: &mut ModelBatchSpool, limit: usize) -> BTreeSet { + spool + .deterministic_bounded_receipt(limit) + .unwrap() + .entries + .into_iter() + .filter_map(|e| { + (spool.hunk_risk.get(&e.hunk).is_none_or(|r| r.mandatory) + && matches!(e.disposition, HunkDisposition::Direct { .. })) + .then_some(e.hunk) + }) + .collect() + } + + #[test] + fn source_packing_preserves_mandatory_identity_at_actual_admission() { + let source = capacity_coverage_fixture(32_768, 22); + let parsed = parse(&source); + assert_eq!( + json_string_content_bytes(&format!( + "{}\n", + build_manifest(&parsed, &[], &[], 10_922).text + )), + 792 + ); + let sizes = parsed + .files + .iter() + .flat_map(|f| { + f.hunks + .iter() + .flat_map(|h| render_hunk_units(f, h, 31_974).unwrap()) + }) + .map(|u| serde_json::to_vec(&u).unwrap().len() - 2) + .collect::>(); + assert_eq!( + sizes, + std::iter::repeat_n(31_776, 22) + .chain([19_182, 22_379, 12_788, 9_591]) + .collect::>() + ); + let cfg = crate::config::Config { + scorer_enabled: true, + scorer: crate::config::default_scorer_model().to_string(), + ..Default::default() + }; + let limit = crate::review::MAX_LARGE_DIFF_SELECTED_BATCHES + .min(crate::llm::max_hosted_review_batches(&cfg, false).unwrap()); + assert_eq!(limit, 23); + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + let mut baseline = + spool_model_batches_next_fit(&mut prepared, 32_768, 32_768, 10_922, false, limit) + .unwrap(); + let mut candidate = spool_model_batches_with_packing( + &mut prepared, + 32_768, + 32_768, + 10_922, + false, + limit, + true, + ) + .unwrap(); + assert_eq!((baseline.source_count, candidate.source_count), (25, 24)); + let d = HunkIdentity::new("src/auth/d.rs", &parsed.files[25].hunks[0]); + assert_eq!( + d.canonical(), + [ + "src/auth/d.rs", + "0", + "0", + "1", + "40", + "ae0333e57ce9e78124abb29d2d3191486b4ce269f0994756b92b943e38be01f6" + ] + .join("\0") + ); + let b = baseline.deterministic_bounded_receipt(limit).unwrap(); + let c = candidate.deterministic_bounded_receipt(limit).unwrap(); + assert_eq!((b.direct_hunks(), b.unreviewed_hunks()), (24, 2)); + assert_eq!((c.direct_hunks(), c.unreviewed_hunks()), (24, 2)); + assert!( + b.entries + .iter() + .any(|e| e.hunk == d && matches!(e.disposition, HunkDisposition::Direct { .. })) + ); + assert!(c.entries.iter().any(|e| e.hunk==d && matches!(e.disposition,HunkDisposition::Unreviewed {..}))); + let mut guarded = spool_model_batches_with_synthesis_budget( + &mut prepared, + 32_768, + 32_768, + 10_922, + false, + limit, + ) + .unwrap(); + assert!(mandatory_direct(&mut guarded, limit).contains(&d)); + assert_same_model_spool(&mut guarded, &mut baseline, limit); + } + + #[test] + fn source_packing_guard_matches_whole_layout_and_releases_lease() { + for budget in [4096, 32_768] { + for prefix in [0, 3, 8] { + let source = capacity_coverage_fixture(budget, prefix); + for limit in [1, 3, 5, 23, 96] { + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + let quota = prepared._window_lease.budget.clone(); + let preparation = quota.retained_bytes(); + let synthesis = if prefix == 3 { 4096 } else { budget }; + let mut candidate = spool_model_batches_with_packing( + &mut prepared, + budget, + synthesis, + 1024, + false, + limit, + true, + ) + .unwrap(); + let c_bytes = candidate._lease.bytes; + let keep_candidate = candidate.source_count <= limit; + if keep_candidate { + let receipt = candidate.deterministic_bounded_receipt(limit).unwrap(); + assert!( + candidate + .hunk_risk + .iter() + .filter(|(_, risk)| risk.mandatory) + .all(|(hunk, _)| receipt.entries.iter().any(|entry| &entry.hunk + == hunk + && matches!( + entry.disposition, + HunkDisposition::Direct { .. } + ))) + ); + } + let mut baseline = spool_model_batches_next_fit( + &mut prepared, + budget, + synthesis, + 1024, + false, + limit, + ) + .unwrap(); + let b_bytes = baseline._lease.bytes; + let expected_direct = mandatory_direct(&mut baseline, limit); + let mut reference = if keep_candidate { + drop(baseline); + candidate + } else { + drop(candidate); + baseline + }; + let expected_bytes = model_spool_bytes(&mut reference); + drop(reference); + assert_eq!(quota.retained_bytes(), preparation); + // Exactly one model spool fits. A retained C lease would make fallback fail. + prepared._window_lease.budget.limit = preparation + c_bytes.max(b_bytes); + SOURCE_PACKING_BUILDS.with(|b| b.borrow_mut().clear()); + SOURCE_PACKING_RECEIPTS.with(|receipts| receipts.set(0)); + let mut guarded = spool_model_batches_with_synthesis_budget( + &mut prepared, + budget, + synthesis, + 1024, + false, + limit, + ) + .unwrap(); + assert_eq!(SOURCE_PACKING_RECEIPTS.with(|receipts| receipts.get()), 0); + let builds = SOURCE_PACKING_BUILDS.with(|b| b.borrow().clone()); + assert_eq!( + builds, + vec![preparation; if keep_candidate { 1 } else { 2 }] + ); + assert_eq!(model_spool_bytes(&mut guarded), expected_bytes); + assert!(expected_direct.is_subset(&mandatory_direct(&mut guarded, limit))); + assert_eq!(quota.retained_bytes(), preparation + guarded._lease.bytes); + drop(guarded); + assert_eq!(quota.retained_bytes(), preparation); + prepared._window_lease.budget.limit = MAX_REVIEW_SPOOL_BYTES; + let mut first = spool_model_batches_with_synthesis_budget( + &mut prepared, + budget, + synthesis, + 1024, + false, + limit, + ) + .unwrap(); + let mut repeated = spool_model_batches_with_synthesis_budget( + &mut prepared, + budget, + synthesis, + 1024, + false, + limit, + ) + .unwrap(); + assert_same_model_spool(&mut first, &mut repeated, limit); + } + } + } + } + + #[test] + fn source_packing_source_count_keeps_candidate_above_total_limit_without_receipt() { + let source = capacity_coverage_fixture(32_768, 3); + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + let candidate = + spool_model_batches_with_packing(&mut prepared, 32_768, 32_768, 1024, false, 24, true) + .unwrap(); + let limit = candidate.source_count; + drop(candidate); + let mut candidate = spool_model_batches_with_packing( + &mut prepared, + 32_768, + 32_768, + 1024, + false, + limit, + true, + ) + .unwrap(); + assert!(candidate.count > limit); + assert_eq!(candidate.source_count, limit); + SOURCE_PACKING_BUILDS.with(|builds| builds.borrow_mut().clear()); + SOURCE_PACKING_RECEIPTS.with(|receipts| receipts.set(0)); + let mut returned = spool_model_batches_for_review( + &mut prepared, + 32_768, + 32_768, + 1024, + false, + limit, + false, + ) + .unwrap(); + assert_eq!(SOURCE_PACKING_RECEIPTS.with(|receipts| receipts.get()), 0); + assert_eq!( + SOURCE_PACKING_BUILDS.with(|builds| builds.borrow().len()), + 1 + ); + let receipt = returned.deterministic_bounded_receipt(limit).unwrap(); + assert_eq!(SOURCE_PACKING_RECEIPTS.with(|receipts| receipts.get()), 1); + returned.validate_coverage_receipt(&receipt).unwrap(); + assert!( + returned + .hunk_risk + .iter() + .filter(|(_, risk)| risk.mandatory) + .all( + |(hunk, _)| receipt.entries.iter().any(|entry| &entry.hunk == hunk + && matches!(entry.disposition, HunkDisposition::Direct { .. })) + ) + ); + assert_same_model_spool(&mut returned, &mut candidate, limit); + } + + #[test] + fn source_packing_guard_mult_window_fragments_and_provider_boundary() { + let mut source = String::from( + "diff --git a/src/auth/large.rs b/src/auth/large.rs\n--- a/src/auth/large.rs\n+++ b/src/auth/large.rs\n@@ -0,0 +1,2400 @@\n", + ); + for _ in 0..2400 { + source.push_str(&format!("+{}\n", "x".repeat(900))); + } + source.push_str( + "diff --git a/logo.bin b/logo.bin\nBinary files a/logo.bin and b/logo.bin differ\n", + ); + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + let mut windows = 0; + while prepared.next_window().unwrap().is_some() { + windows += 1; + } + assert!(windows > 1); + for budget in [4096, 32_768] { + let mut baseline = + spool_model_batches_next_fit(&mut prepared, budget, 4096, 1024, false, 5).unwrap(); + let expected = mandatory_direct(&mut baseline, 5); + let mut guarded = spool_model_batches_with_synthesis_budget( + &mut prepared, + budget, + 4096, + 1024, + false, + 5, + ) + .unwrap(); + assert!(expected.is_subset(&mandatory_direct(&mut guarded, 5))); + assert_same_model_spool(&mut guarded, &mut baseline, 5); + } + // Explicit provider-bounded selection retains baseline inventory, even + // when deterministic selection at 24 would cover every mandatory hunk. + let source = capacity_coverage_fixture(32_768, 3); + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + SOURCE_PACKING_RECEIPTS.with(|receipts| receipts.set(0)); + let mut provider = + spool_model_batches_for_review(&mut prepared, 32_768, 32_768, 1024, false, 24, true) + .unwrap(); + assert_eq!(SOURCE_PACKING_RECEIPTS.with(|receipts| receipts.get()), 0); + let mut baseline = + spool_model_batches_next_fit(&mut prepared, 32_768, 32_768, 1024, false, 24).unwrap(); + assert_same_model_spool(&mut provider, &mut baseline, 24); + let candidates = provider.hosted_candidates(5, 96).unwrap(); + assert!(candidates.mandatory_ids.len() <= 5); + assert!( + candidates + .candidate_ids + .iter() + .all(|id| *id > 0 && *id <= provider.count) + ); + let source = source.replace("/auth/", "/logic/"); + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + let mut no_mandatory = + spool_model_batches_with_synthesis_budget(&mut prepared, 32_768, 4096, 1024, false, 1) + .unwrap(); + assert!(no_mandatory.hunk_risk.values().all(|r| !r.mandatory)); + let mut baseline = + spool_model_batches_next_fit(&mut prepared, 32_768, 4096, 1024, false, 1).unwrap(); + assert_same_model_spool(&mut no_mandatory, &mut baseline, 1); + } + + #[test] + fn source_packing_guard_small_empty_metadata_and_errors() { + for source in [ + "", + "diff --git a/logo.bin b/logo.bin\nBinary files a/logo.bin and b/logo.bin differ\n", + "diff --git a/src/auth/a.rs b/src/auth/a.rs\n--- a/src/auth/a.rs\n+++ b/src/auth/a.rs\n@@ -0,0 +1 @@\n+require_admin();\n", + ] { + for force in [false, true] { + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + let mut guarded = spool_model_batches_with_synthesis_budget( + &mut prepared, + 4096, + 4096, + 1024, + force, + 24, + ) + .unwrap(); + assert!(guarded.count <= 24); + for hunk in &guarded.all_hunks { + assert!( + guarded + .batch_hunks + .iter() + .any( + |(id, hs)| !guarded.synthesis_ids.contains(id) && hs.contains(hunk) + ) + ); + } + let receipt = guarded.deterministic_bounded_receipt(24).unwrap(); + assert_eq!(receipt.direct_hunks(), guarded.all_hunks.len()); + let mut baseline = + spool_model_batches_next_fit(&mut prepared, 4096, 4096, 1024, force, 24) + .unwrap(); + assert_same_model_spool(&mut guarded, &mut baseline, 24); + } + } + let source = capacity_coverage_fixture(32_768, 22); + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + let quota = prepared._window_lease.budget.clone(); + let preparation = quota.retained_bytes(); + let candidate = + spool_model_batches_with_packing(&mut prepared, 32_768, 32_768, 1024, false, 23, true) + .unwrap(); + let candidate_bytes = candidate._lease.bytes; + drop(candidate); + // C fits, but B's extra source batch and semantic inventory do not. + prepared._window_lease.budget.limit = preparation + candidate_bytes; + let error = spool_model_batches_with_synthesis_budget( + &mut prepared, + 32_768, + 32_768, + 1024, + false, + 23, + ) + .err() + .unwrap(); + assert!( + format!("{error:#}").contains("operation quota"), + "{error:#}" + ); + assert_eq!(quota.retained_bytes(), preparation); + prepared._window_lease.budget.limit = MAX_REVIEW_SPOOL_BYTES; + let mut invalid = + spool_model_batches_with_packing(&mut prepared, 32_768, 32_768, 1024, false, 23, true) + .unwrap(); + let hunk = invalid.all_hunks.first().unwrap().clone(); + invalid.hunk_risk.remove(&hunk); + assert!(invalid.deterministic_bounded_receipt(23).is_err()); + drop(invalid); + assert_eq!(quota.retained_bytes(), preparation); + // A truncated model file is an error, never interpreted as capacity loss. + let mut truncated = + spool_model_batches_with_packing(&mut prepared, 32_768, 32_768, 1024, false, 23, true) + .unwrap(); + truncated.file.set_len(1).unwrap(); + assert!(truncated.deterministic_bounded_receipt(23).is_err()); + drop(truncated); + assert_eq!(quota.retained_bytes(), preparation); + assert!( + spool_model_batches_with_synthesis_budget( + &mut prepared, + 32_768, + 32_768, + 1024, + false, + 0 + ) + .is_err() + ); + assert_eq!(quota.retained_bytes(), preparation); + prepared.windows.seek(SeekFrom::Start(0)).unwrap(); + prepared.windows.write_all(&4096u64.to_le_bytes()).unwrap(); + prepared.windows.set_len(9).unwrap(); + assert!( + spool_model_batches_with_synthesis_budget( + &mut prepared, + 32_768, + 32_768, + 1024, + false, + 23 + ) + .is_err() + ); + assert_eq!(quota.retained_bytes(), preparation); + #[cfg(unix)] + { + let (stream, _peer) = std::os::unix::net::UnixStream::pair().unwrap(); + let descriptor: std::os::fd::OwnedFd = stream.into(); + prepared.windows = File::from(descriptor); + let error = spool_model_batches_with_synthesis_budget( + &mut prepared, + 32_768, + 32_768, + 1024, + false, + 23, + ) + .err() + .unwrap(); + assert!(format!("{error:#}").contains("rewinding review windows")); + assert_eq!(quota.retained_bytes(), preparation); + } + } + const SAMPLE: &str = "\ diff --git a/src/lib.rs b/src/lib.rs index 1111..2222 100644 @@ -7278,6 +8005,528 @@ diff --git a/src/multi.rs b/src/multi.rs ); } + fn packing_units(batch: &str) -> Vec { + batch + .split_inclusive('\n') + .fold(Vec::::new(), |mut units, line| { + if line.starts_with("### ") { + units.push(String::new()); + } + if let Some(unit) = units.last_mut() { + unit.push_str(line); + } + units + }) + } + + fn reference_append_unit( + plan: &mut ReviewBatchPlan, + current: &mut String, + hunks: &mut BTreeSet, + initial: &str, + evidence: (&str, &HunkIdentity), + budget: usize, + ) -> bool { + let (unit, hunk) = evidence; + let size = |text: &str| serde_json::to_vec(text).unwrap().len() - 2; + if size(initial) + size(unit) > budget { + return false; + } + if current.is_empty() { + current.push_str(initial); + } + if size(current) + size(unit) > budget { + if let Some(index) = plan + .batches + .iter() + .position(|batch| size(batch) + size(unit) <= budget) + { + plan.batches[index].push_str(unit); + plan.batch_hunks[index].insert(hunk.clone()); + return true; + } + plan.batches.push(std::mem::take(current)); + plan.batch_hunks.push(std::mem::take(hunks)); + current.push_str(initial); + } + current.push_str(unit); + hunks.insert(hunk.clone()); + true + } + + #[test] + fn source_packing_generated_transitions_match_serialized_reference() { + let parsed = parse( + "diff --git a/src/a.rs b/src/a.rs\n--- a/src/a.rs\n+++ b/src/a.rs\n@@ -1 +1 @@\n-old\n+first\n@@ -1 +1 @@\n-old\n+second\ndiff --git a/src/b.rs b/src/b.rs\n--- a/src/b.rs\n+++ b/src/b.rs\n@@ -1 +1 @@\n-old\n+third\n", + ); + let identities = parsed + .files + .iter() + .flat_map(|file| { + file.hunks + .iter() + .map(|hunk| HunkIdentity::new(&file.path, hunk)) + }) + .collect::>(); + assert_ne!(identities[0], identities[1]); + let manifest = "Changed-file manifest:\n- \"src/quote\\\"界.rs\" [source]\n"; + for budget in [4_096, 32_768] { + for seed in 1..=12u64 { + let mut state = seed; + let mut plan = ReviewBatchPlan::default(); + let mut packing = SourceBatchPacking::new(manifest); + let mut reference = ReviewBatchPlan::default(); + let mut current = String::new(); + let mut hunks = BTreeSet::new(); + let mut expected = Vec::::new(); + let mut owners = BTreeMap::::new(); + let mut next_fit = vec![packing.initial_bytes]; + for index in 0..48 { + state = state.wrapping_mul(6364136223846793005).wrapping_add(1); + let owner = &identities[(state as usize) % identities.len()]; + let heading = format!( + "### {}\n@@ generated {index} @@\n 1 + ", + display_path(&owner.path) + ); + let pattern = ["界", "\0", "\\", "\"", "\t", "é"][(state as usize / 3) % 6]; + let desired = (state as usize / 7) % (budget + 80); + let mut unit = format!( + "{heading}{}\n", + pattern.repeat(desired / json_string_content_bytes(pattern)) + ); + if index % 13 == 0 && !expected.is_empty() { + unit = expected[0].clone(); + } + let owner = owners.get(&unit).cloned().unwrap_or_else(|| owner.clone()); + let accepted = append_unit(&mut plan, &mut packing, &unit, &owner, budget); + assert_eq!( + accepted, + reference_append_unit( + &mut reference, + &mut current, + &mut hunks, + &packing.initial, + (&unit, &owner), + budget + ) + ); + if accepted { + let size = serde_json::to_vec(&unit).unwrap().len() - 2; + if *next_fit.last().unwrap() + size > budget { + next_fit.push(packing.initial_bytes); + } + *next_fit.last_mut().unwrap() += size; + owners.insert(unit.clone(), owner.clone()); + expected.push(unit); + } + assert_eq!(plan.batches, reference.batches); + assert_eq!(plan.batch_hunks, reference.batch_hunks); + assert_eq!(packing.current, current); + assert_eq!(packing.current_hunks, hunks); + assert_eq!(packing.closed_bytes.len(), plan.batches.len()); + assert_eq!( + packing.current_bytes, + serde_json::to_vec(&packing.current).unwrap().len() - 2 + ); + assert_eq!(plan.batches.len(), plan.batch_hunks.len()); + for (batch, cached) in plan.batches.iter().zip(&packing.closed_bytes) { + assert_eq!(*cached, serde_json::to_vec(batch).unwrap().len() - 2); + assert!(*cached <= budget); + assert!(batch.starts_with(&packing.initial)); + } + let mut actual = Vec::new(); + for (batch, credited) in plan + .batches + .iter() + .zip(&plan.batch_hunks) + .chain(std::iter::once((&packing.current, &packing.current_hunks))) + { + let units = packing_units(batch); + assert_eq!( + *credited, + units.iter().map(|unit| owners[unit].clone()).collect() + ); + actual.extend(units); + } + let mut sorted_expected = expected.clone(); + sorted_expected.sort(); + actual.sort(); + assert_eq!(actual, sorted_expected); + assert!( + plan.batches.len() + usize::from(!packing.current.is_empty()) + <= next_fit.len() + ); + } + // Rejection does not flush or mutate the returned source plan. + let before = ( + plan.batches.clone(), + plan.batch_hunks.clone(), + packing.current.clone(), + packing.current_bytes, + packing.closed_bytes.clone(), + ); + assert!(!append_unit( + &mut plan, + &mut packing, + &"x".repeat(budget), + &identities[0], + budget + )); + assert_eq!( + before, + ( + plan.batches, + plan.batch_hunks, + packing.current, + packing.current_bytes, + packing.closed_bytes + ) + ); + } + let mut plan = ReviewBatchPlan::default(); + let mut packing = SourceBatchPacking::new(manifest); + let available = budget - packing.initial_bytes; + let mut unit = "\0\"\\界".repeat(available / 13); + unit.push_str(&"x".repeat(available - (serde_json::to_vec(&unit).unwrap().len() - 2))); + assert!(append_unit( + &mut plan, + &mut packing, + &unit, + &identities[0], + budget + )); + assert_eq!(packing.current_bytes, budget); + assert_eq!( + serde_json::to_vec(&packing.current).unwrap().len() - 2, + budget + ); + } + } + + #[test] + fn source_packing_reuses_capacity_without_changing_units_or_ownership() { + let mut source = String::new(); + for (path, lines, marker) in [ + ("src/a.rs", 20, 'a'), + ("src/b.rs", 25, 'b'), + ("src/a.rs", 10, 'c'), + ] { + source.push_str(&format!( + "diff --git a/{path} b/{path}\n--- a/{path}\n+++ b/{path}\n@@ -0,0 +1,{lines} @@\n" + )); + for _ in 0..lines { + source.push_str(&format!("+{}\n", marker.to_string().repeat(990))); + } + } + source.push_str("diff --git a/logo.bin b/logo.bin\nold mode 100644\nnew mode 100755\nBinary files a/logo.bin and b/logo.bin differ\n"); + let parsed = parse(&source); + let manifest = build_manifest(&parsed, &[], &[], 4096); + let plan = render_review_batches(&parsed, &[], &[], 32_768, 4096); + let again = render_review_batches(&parsed, &[], &[], 32_768, 4096); + assert!(!plan.incomplete); + assert_eq!(plan.batches.len(), 2); + assert_eq!(plan.batches, again.batches); + assert_eq!(plan.batch_hunks, again.batch_hunks); + let mut expected = Vec::new(); + let mut owners = BTreeMap::new(); + let mut next_fit_sizes = vec![json_string_content_bytes(&format!("{}\n", manifest.text))]; + for file in &parsed.files { + for hunk in &file.hunks { + for unit in render_hunk_units( + file, + hunk, + 32_768 - json_string_content_bytes(&manifest.text) - 2, + ) + .unwrap() + { + let size = json_string_content_bytes(&unit); + if *next_fit_sizes.last().unwrap() + size > 32_768 { + next_fit_sizes + .push(json_string_content_bytes(&format!("{}\n", manifest.text))); + } + *next_fit_sizes.last_mut().unwrap() += size; + owners.insert(unit.clone(), HunkIdentity::new(&file.path, hunk)); + expected.push(unit); + } + } + } + assert_eq!(next_fit_sizes.len(), 3); + let mut actual = Vec::new(); + for (batch, hunks) in plan.batches.iter().zip(&plan.batch_hunks) { + assert!(batch.starts_with(&format!("{}\n", manifest.text))); + assert!(batch.contains("mode 100644 -> 100755")); + let units = packing_units(batch.strip_prefix(&format!("{}\n", manifest.text)).unwrap()); + assert_eq!( + *hunks, + units.iter().map(|unit| owners[unit].clone()).collect() + ); + actual.extend(units); + } + actual.sort(); + expected.sort(); + assert_eq!(actual, expected); + assert_eq!(plan.batch_hunks[0].len(), 2); + assert_eq!(plan.batch_hunks[1].len(), 1); + assert!(review_batch_contains_exact_evidence( + &plan.batches[0], + "src/a.rs", + 1, + Some(&"c".repeat(990)) + )); + assert!(!review_batch_contains_exact_evidence( + &plan.batches[0], + "src/b.rs", + 1, + Some(&"c".repeat(990)) + )); + } + + #[test] + fn source_packing_reordered_segments_preserve_grounding_and_mandatory_fragments() { + let path = "src/auth/quote\"界.rs"; + let header = format!("### {}\n", display_path(path)); + let parsed = parse( + "diff --git a/src/auth/a.rs b/src/auth/a.rs\n--- a/src/auth/a.rs\n+++ b/src/auth/a.rs\n@@ -1 +1 @@\n-old\n+require_admin();\n", + ); + let identity = HunkIdentity::new(path, &parsed.files[0].hunks[0]); + let mut plan = ReviewBatchPlan::default(); + let mut packing = SourceBatchPacking::new("manifest"); + let units = [ + format!( + "{header}@@ segment one @@\n 1 + {}\n", + "a".repeat(2_300) + ), + format!( + "{header}@@ segment two @@\n 20 + {}\n", + "b".repeat(3_100) + ), + format!( + "{header}@@ segment three @@\n 30 + load_order_for_update(id);\n 31 + {}\n", + "c".repeat(1000) + ), + ]; + for unit in &units { + assert!(append_unit(&mut plan, &mut packing, unit, &identity, 4096)); + } + plan.batches.push(packing.current); + plan.batch_hunks.push(packing.current_hunks); + assert_eq!(plan.batches.len(), 2); + assert!(plan.batches[0].contains(&units[2])); + let decorated = format!( + "Synthetic title and feedback\n{}\nRetry instructions", + plan.batches[0] + ); + assert!(!review_batch_contains_range(&decorated, path, 1, 30)); + assert!(review_batch_contains_exact_evidence( + &decorated, + path, + 30, + Some("load_order_for_update(id);") + )); + assert!(!review_batch_contains_exact_evidence( + &decorated, + "src/other.rs", + 30, + Some("load_order_for_update(id);") + )); + let local = render_review_batch_context(&decorated, path, 30, 2, 1024).unwrap(); + assert!(local.contains("load_order_for_update")); + let related = render_related_scorer_evidence( + &plan.batches, + path, + "Verify load_order_for_update", + 1024, + ) + .unwrap(); + assert!(related.len() <= 1024); + assert!(related.contains(&display_path(path))); + + let long_line = format!("{}TAIL", "界".repeat(4_000)); + let mut source = String::from( + "diff --git a/src/auth/a.rs b/src/auth/a.rs\n--- a/src/auth/a.rs\n+++ b/src/auth/a.rs\n@@ -1 +1,61 @@\n-old\n", + ); + source.push_str(&format!("+{long_line}\n")); + for _ in 0..60 { + source.push_str(&format!("+require_admin({});\n", "x".repeat(80))); + } + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + let mut batches = spool_model_batches(&mut prepared, 4096, 1024, false).unwrap(); + let hunk = batches.hunk_risk.keys().next().unwrap().clone(); + assert!(batches.hunk_risk[&hunk].mandatory); + let required = batches + .batch_hunks + .iter() + .filter(|(id, hs)| !batches.synthesis_ids.contains(id) && hs.contains(&hunk)) + .map(|(id, _)| *id) + .collect::>(); + assert!(required.len() > 1); + let partial = batches.deterministic_bounded_receipt(1).unwrap(); + assert_eq!(partial.direct_hunks(), 0); + assert_eq!(partial.unreviewed_hunks(), 1); + let complete = batches + .deterministic_bounded_receipt(batches.count) + .unwrap(); + assert_eq!(complete.direct_hunks(), 1); + assert_eq!(complete.unreviewed_hunks(), 0); + assert!(required.is_subset(&complete.selected_batch_ids)); + assert!( + matches!(&complete.entries[0].disposition, HunkDisposition::Direct { batch_ids } if batch_ids.iter().copied().collect::>() == required) + ); + for limit in 1..required.len() { + let receipt = batches.deterministic_bounded_receipt(limit).unwrap(); + assert_eq!(receipt.direct_hunks(), 0); + assert_eq!(receipt.unreviewed_hunks(), 1); + } + let mut repeated_prepared = prepare_review(&snapshot).unwrap(); + let mut repeated = spool_model_batches(&mut repeated_prepared, 4096, 1024, false).unwrap(); + let repeated_receipt = repeated + .deterministic_bounded_receipt(repeated.count) + .unwrap(); + assert_eq!(complete.plan_sha256, repeated_receipt.plan_sha256); + assert_eq!(complete.entries, repeated_receipt.entries); + let mut wider_prepared = prepare_review(&snapshot).unwrap(); + let wider = spool_model_batches(&mut wider_prepared, 32768, 1024, false).unwrap(); + assert_eq!( + batches.hunk_risk.keys().collect::>(), + wider.hunk_risk.keys().collect() + ); + for (hunk, risk) in &batches.hunk_risk { + let other = &wider.hunk_risk[hunk]; + assert_eq!( + (risk.mandatory, risk.semantic_eligible, risk.score), + (other.mandatory, other.semantic_eligible, other.score) + ); + } + batches.validate_coverage_receipt(&complete).unwrap(); + let texts = batches.batch_text_by_id(&required).unwrap(); + let mut coverage = vec![false; long_line.len()]; + for batch in texts.values() { + let mut slice = None; + for line in batch.lines() { + if let Some(range) = line + .strip_prefix("@@slice ") + .and_then(|r| r.strip_suffix("@@")) + { + slice = range + .split_once("..") + .map(|(a, b)| (a.parse::().unwrap(), b.parse::().unwrap())); + } else if let Some((start, end)) = slice.take() { + let payload = line.trim_start().split_once(" + ").unwrap().1; + assert_eq!(payload, &long_line[start..end]); + coverage[start..end].fill(true); + assert!(review_batch_contains_exact_evidence( + batch, + "src/auth/a.rs", + 1, + Some(payload) + )); + } + } + } + assert!(coverage.into_iter().all(|covered| covered)); + } + + #[test] + #[ignore = "native scaling comparison"] + fn source_packing_near_window_failed_fits_match_reference() { + for budget in [4_096, 32_768] { + let mut source = String::from( + "diff --git a/src/many.rs b/src/many.rs\n--- a/src/many.rs\n+++ b/src/many.rs\n", + ); + let lines = budget / 1_800 + 1; + let mut index = 0; + while source.len() < 1_900_000 { + source.push_str(&format!("@@ -0,0 +{},{lines} @@\n", 1 + index * lines)); + index += 1; + for _ in 0..lines { + source.push_str(&format!("+{}\n", "x".repeat(900))); + } + } + assert!(source.len() < STREAM_WINDOW_BYTES); + let snapshot = DiffSnapshot::from_bytes(source.as_bytes()).unwrap(); + let mut prepared = prepare_review(&snapshot).unwrap(); + let window = prepared.next_window().unwrap().unwrap(); + assert!(window.len() < STREAM_WINDOW_BYTES); + assert!(prepared.next_window().unwrap().is_none()); + let diff = parse(&window); + assert!(diff.complete); + // Use a bounded common prefix to isolate fit-search work from manifest construction. + let manifest = "Changed-file manifest:\n"; + let mut packing = SourceBatchPacking::new(manifest); + let mut plan = ReviewBatchPlan::default(); + let units = diff + .files + .iter() + .flat_map(|file| { + file.hunks.iter().flat_map(|hunk| { + render_hunk_units(file, hunk, budget - packing.initial_bytes) + .unwrap() + .into_iter() + .map(|unit| (unit, HunkIdentity::new(&file.path, hunk))) + .collect::>() + }) + }) + .collect::>(); + let started = std::time::Instant::now(); + for (unit, hunk) in &units { + assert!(append_unit(&mut plan, &mut packing, unit, hunk, budget)); + } + let cached_elapsed = started.elapsed(); + let failed_fit_checks = (plan.batches.len() * plan.batches.len().saturating_sub(1)) / 2; + assert!(failed_fit_checks > 1_000); + let mut reference = ReviewBatchPlan::default(); + let mut current = String::new(); + let mut hunks = BTreeSet::new(); + let started = std::time::Instant::now(); + for (unit, hunk) in &units { + assert!(reference_append_unit( + &mut reference, + &mut current, + &mut hunks, + &packing.initial, + (unit, hunk), + budget + )); + } + let reference_elapsed = started.elapsed(); + let started = std::time::Instant::now(); + let mut next_fit = Vec::new(); + let mut next_current = String::new(); + for (unit, _) in &units { + if next_current.is_empty() { + next_current.push_str(&packing.initial); + } + if json_string_content_bytes(&next_current) + json_string_content_bytes(unit) + > budget + { + next_fit.push(std::mem::take(&mut next_current)); + next_current.push_str(&packing.initial); + } + next_current.push_str(unit); + } + let next_fit_elapsed = started.elapsed(); + assert_eq!(next_fit, plan.batches); + assert_eq!(next_current, packing.current); + assert_eq!(plan.batches, reference.batches); + assert_eq!(plan.batch_hunks, reference.batch_hunks); + assert_eq!(packing.current, current); + assert_eq!(packing.current_hunks, hunks); + for (batch, size) in plan.batches.iter().zip(&packing.closed_bytes) { + assert_eq!(*size, serde_json::to_vec(batch).unwrap().len() - 2); + } + eprintln!( + "source_budget={budget} normalized_bytes={} units={} closed_batches={} failed_fit_checks={failed_fit_checks} cached_us={} serialized_reference_us={} next_fit_us={}", + window.len(), + units.len(), + plan.batches.len(), + cached_elapsed.as_micros(), + reference_elapsed.as_micros(), + next_fit_elapsed.as_micros() + ); + } + } + #[test] fn batch_grounding_requires_both_range_endpoints_in_one_segment() { let batch = "### src/a.rs\n@@ first @@\n 10 + first\n 11 + second\n@@ second @@\n 20 + third\n"; diff --git a/src/llm.rs b/src/llm.rs index a88f510..90ed530 100644 --- a/src/llm.rs +++ b/src/llm.rs @@ -69,7 +69,7 @@ pub struct ScorerReview { #[derive(Debug, Clone)] pub(crate) struct AdjudicationReview { - pub(crate) results: Vec, + pub(crate) application: crate::adjudication::AdjudicationApplication, pub(crate) usage: Usage, pub(crate) model_usage: Vec, pub(crate) model_incidents: Vec, @@ -197,6 +197,17 @@ impl std::error::Error for AtomicAttributionRequestTooLarge {} #[derive(Debug)] struct ProviderHttpFailure(reqwest::StatusCode); +#[derive(Debug)] +struct ProviderAdmissionFailure; + +impl std::fmt::Display for ProviderAdmissionFailure { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str("model request has no residual admitted capacity") + } +} + +impl std::error::Error for ProviderAdmissionFailure {} + impl std::fmt::Display for ProviderHttpFailure { fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(formatter, "model endpoint returned {}", self.0) @@ -1891,6 +1902,7 @@ impl std::error::Error for ModelContentFailure {} fn classify_chat_error(error: anyhow::Error) -> anyhow::Error { if error.downcast_ref::().is_some() + || error.downcast_ref::().is_some() || error.downcast_ref::().is_some() || error .downcast_ref::() @@ -3264,20 +3276,30 @@ impl LlmClient { eprintln!("postil: adjudication-output {}", record); } - pub(crate) async fn adjudicate_findings( + fn adjudication_trace_for_call(mut record: serde_json::Value, phase: LlmCallPhase) { + record["call"] = json!(if matches!(phase, LlmCallPhase::Initial) { + 1 + } else { + 2 + }); + record["callPhase"] = json!(phase.as_str()); + Self::adjudication_trace(record); + } + + #[allow(clippy::too_many_arguments)] + pub(crate) async fn adjudicate_findings( &self, model: &str, system: &str, user: &str, expected_len: usize, timeout: Duration, - ) -> std::result::Result { + validate: F, + ) -> std::result::Result + where + F: Fn(Vec) -> Result, + { let mut client = self.clone(); - Self::adjudication_trace(adjudication_record( - "callStarted", - "chatTerminal", - "started", - )); let deadline = Instant::now() + timeout; client.scorer_deadline = Some( self.total_deadline @@ -3285,58 +3307,178 @@ impl LlmClient { ); let mut usage = Usage::default(); let mut model_usage = Vec::new(); + let mut model_incidents: Vec = Vec::new(); let mut usage_accounting_complete = true; - let content = client - .chat_with_temperature( - model, - system, - user, - &mut usage, - &mut model_usage, - &mut usage_accounting_complete, - MAX_ADJUDICATION_OUTPUT_TOKENS, - 0.0, - LlmPhase::Adjudication, - LlmCallPhase::Initial, - ) - .await - .map_err(|error| { - Self::adjudication_trace(adjudication_chat_record( - &error, - "callFailed", - "chatTerminal", - None, - )); - let mut model_error = ModelError::new(error, usage, usage_accounting_complete); - model_error.model_usage = model_usage.clone(); - model_error - .model_incidents - .push(model_error.incident(ModelIncidentPhase::Scorer)); - model_error - })?; - let results = - parse_adjudication_results(&content, expected_len, true).map_err(|detail| { - let mut error = ModelError::new( - anyhow!("finding adjudication output invalid: {detail}"), - usage, - usage_accounting_complete, - ); - error.model_usage = model_usage.clone(); - error.model_incidents.push(ModelIncident { - phase: ModelIncidentPhase::Scorer, - category: ModelIncidentCategory::InvalidOutput, - recovered: false, - recovery: None, - }); - error - })?; - Ok(AdjudicationReview { - results, + let mut call_phase = LlmCallPhase::Initial; + let mut request_user = user.to_owned(); + let mut last_validation_error = None; + let failure = |cause: anyhow::Error, + usage, + events: Vec, + incidents, + complete, + validation_error: Option<&anyhow::Error>| { + let cause = match validation_error { + Some(validation) => { + match validation.downcast_ref::() { + Some(violation) => cause.context(*violation), + None => cause.context(validation.to_string()), + } + } + None => cause, + }; + let mut error = ModelError::new(cause, usage, complete); + error.model_usage = events; + error.model_incidents = incidents; + error + }; + // Syntax and semantic correction share this one operation's budget. + for call in 0..2 { + Self::adjudication_trace_for_call( + adjudication_record("callStarted", "chatTerminal", "started"), + call_phase, + ); + let previous_events = model_usage.len(); + let content = client + .chat_with_temperature( + model, + system, + &request_user, + &mut usage, + &mut model_usage, + &mut usage_accounting_complete, + MAX_ADJUDICATION_OUTPUT_TOKENS, + 0.0, + LlmPhase::Adjudication, + call_phase, + ) + .await + .map_err(|cause| { + Self::adjudication_trace_for_call( + adjudication_chat_record(&cause, "callFailed", "chatTerminal", None), + call_phase, + ); + let mut error = failure( + cause, + usage, + model_usage.clone(), + model_incidents.clone(), + usage_accounting_complete, + last_validation_error.as_ref(), + ); + let incident = error.incident(ModelIncidentPhase::Scorer); + // A refused local admission is not another invalid response. + if model_incidents.is_empty() + || model_usage.len() > previous_events + || incident.category != ModelIncidentCategory::InvalidOutput + { + error.model_incidents.push(incident); + } + error + })?; + let (application, correction_phase) = + match parse_adjudication_results(&content, expected_len, Some(call_phase)) { + Ok(results) => { + let application = validate(results); + let recoverable = application.as_ref().err().is_some_and(|error| { + error + .downcast_ref::() + .is_some() + }); + ( + application, + recoverable.then_some(LlmCallPhase::SemanticRetry), + ) + } + Err(_) => ( + Err(anyhow!("finding adjudication syntax is invalid")), + Some(LlmCallPhase::SchemaRepair), + ), + }; + match application { + Ok(application) => { + client + .remaining_budget(LlmPhase::Adjudication) + .map_err(|cause| { + let mut error = failure( + cause, + usage, + model_usage.clone(), + model_incidents.clone(), + usage_accounting_complete, + last_validation_error.as_ref(), + ); + error + .model_incidents + .push(error.incident(ModelIncidentPhase::Scorer)); + error + })?; + for incident in &mut model_incidents { + incident.recovered = true; + incident.recovery = Some(ModelIncidentRecovery::Repair); + } + return Ok(AdjudicationReview { + application, + usage, + model_usage, + model_incidents, + usage_accounting_complete, + }); + } + Err(cause) => { + eprintln!("postil: finding adjudication validation failed: {cause:#}"); + last_validation_error = Some(cause); + model_incidents.push(ModelIncident { + phase: ModelIncidentPhase::Scorer, + category: ModelIncidentCategory::InvalidOutput, + recovered: false, + recovery: None, + }); + } + } + let Some(next_phase) = correction_phase.filter(|_| call == 0) else { + break; + }; + let corrected = client + .remaining_budget(LlmPhase::Adjudication) + .and_then(|remaining| { + ensure!( + remaining.is_none_or(|remaining| remaining >= RETRY_ATTEMPT_RESERVE), + "finding adjudication has no viable correction slot before its deadline" + ); + adjudication_correction_user(user) + }) + .map_err(|cause| { + let mut error = failure( + cause, + usage, + model_usage.clone(), + model_incidents.clone(), + usage_accounting_complete, + last_validation_error.as_ref(), + ); + if error.is_deadline_exceeded() { + error + .model_incidents + .push(error.incident(ModelIncidentPhase::Scorer)); + } + error + })?; + request_user = corrected; + call_phase = next_phase; + // Actual residual exposure is reserved atomically by the same + // provider path as the initial call, immediately before contact. + } + Err(failure( + last_validation_error.unwrap_or_else(|| { + anyhow!("finding adjudication output did not satisfy its admitted contract") + }), usage, model_usage, - model_incidents: Vec::new(), + model_incidents, usage_accounting_complete, - }) + None, + )) } pub async fn resolve_uncertainty( @@ -4531,12 +4673,15 @@ impl LlmClient { Ok(status) => status.unwrap_or(response.status), Err(error) => { if matches!(phase, LlmPhase::Adjudication) { - Self::adjudication_trace(adjudication_chat_record( - &error, - "attemptRejected", - "endpointErrorContract", - Some(attempt), - )); + Self::adjudication_trace_for_call( + adjudication_chat_record( + &error, + "attemptRejected", + "endpointErrorContract", + Some(attempt), + ), + call_phase, + ); } if let Some(response_usage) = summary.usage { add_usage(usage, response_usage); @@ -4593,12 +4738,15 @@ impl LlmClient { ) { if matches!(phase, LlmPhase::Adjudication) { - Self::adjudication_trace(adjudication_chat_record( - &error, - "attemptRejected", - "endpointContent", - Some(attempt), - )); + Self::adjudication_trace_for_call( + adjudication_chat_record( + &error, + "attemptRejected", + "endpointContent", + Some(attempt), + ), + call_phase, + ); } if let Some(response_usage) = summary.usage { add_usage(usage, response_usage); @@ -4658,12 +4806,15 @@ impl LlmClient { } Err(error) => { if matches!(phase, LlmPhase::Adjudication) { - Self::adjudication_trace(adjudication_chat_record( - &error, - "attemptRejected", - "endpointContent", - Some(attempt), - )); + Self::adjudication_trace_for_call( + adjudication_chat_record( + &error, + "attemptRejected", + "endpointContent", + Some(attempt), + ), + call_phase, + ); } let parse_added_usage = usage.prompt_tokens != usage_before_parse.prompt_tokens @@ -4880,6 +5031,9 @@ impl LlmClient { } return Err(error.context("request to model endpoint failed")); } + Err(error) if error.downcast_ref::().is_some() => { + return Err(error); + } Err(error) => { *usage_accounting_complete = false; call_usage @@ -5019,7 +5173,8 @@ impl LlmClient { review_route: Option, call_phase: LlmCallPhase, ) -> Result { - self.reserve_provider_attempt(body)?; + self.reserve_provider_attempt(body) + .context(ProviderAdmissionFailure)?; let http = self.http_client()?; let mut request = match self.request_decorations.api_format { ApiFormat::OpenaiCompatible => { @@ -5130,9 +5285,14 @@ impl LlmClient { .ok_or_else(|| anyhow!("model provider spend exposure overflowed"))?; let total_projected_cost = admission .projected_cost_exposure_micros + .max(admission.reported_cost_micros) .checked_add(projected_cost_micros) .ok_or_else(|| anyhow!("model provider projected cost exposure overflowed"))?; if self.request_decorations.hosted_price_bounds.is_some() { + ensure!( + admission.reported_token_spend < MAX_REPORTED_TOKEN_SPEND, + "model token spend exhausted the {MAX_REPORTED_TOKEN_SPEND} token hard cap" + ); ensure!( attempts <= MAX_PROVIDER_ATTEMPTS, "model provider attempt hard cap ({MAX_PROVIDER_ATTEMPTS}) exceeded" @@ -6337,8 +6497,8 @@ fn parse_scores(content: &str, expected_len: usize) -> Result, // This wire object has fixed keys and finite values only. Null means unavailable; // counts saturate independently of the authoritative parser and retries. The -// private review entry point calls adjudicate_findings once per child process; -// call=1 is scoped to that child, not a shared or proxy request counter. +// call is the adjudication logical call (1 initial, 2 correction), scoped to +// one operation; attempt remains the per-call transport attempt counter. fn adjudication_record( event: &'static str, stage: &'static str, @@ -6486,15 +6646,27 @@ fn adjudication_count(value: usize) -> serde_json::Value { } } +const ADJUDICATION_CORRECTION_SUFFIX: &str = "\n\nReturn the complete adjudication JSON array for the original candidate IDs. Correct the rejected output using only the supplied evidence. A refutation needs candidate-specific contradictory evidence, not its own citation or unsupported prose. A context-anchored confirmation needs an exact causal change reference. Preserve all structural, scope and publication rules. Never invent evidence or references, silently drop a candidate, or fabricate a verdict."; + +fn adjudication_correction_user(user: &str) -> Result { + ensure!( + user.len() + .saturating_add(ADJUDICATION_CORRECTION_SUFFIX.len()) + <= crate::adjudication::MAX_ADJUDICATION_PROMPT_BYTES, + "finding adjudication correction exceeds its admitted input bound" + ); + Ok(format!("{user}{ADJUDICATION_CORRECTION_SUFFIX}")) +} + fn parse_adjudication_results( content: &str, expected_len: usize, - trace: bool, + call_phase: Option, ) -> Result, String> { let emit = |mut record: serde_json::Value| { record["expected"] = adjudication_count(expected_len); - if trace { - LlmClient::adjudication_trace(record); + if let Some(phase) = call_phase { + LlmClient::adjudication_trace_for_call(record, phase); } }; let json = extract_json_array(content).ok_or_else(|| { @@ -9359,6 +9531,212 @@ mod tests { assert_eq!(overflow.model_costs_micros[&first.model], u64::MAX); } + #[test] + fn source_packing_phase_admission_uses_original_deadline() { + let config = Config { + scorer_enabled: true, + scorer: crate::config::default_scorer_model().to_string(), + ..Config::default() + }; + let generator = crate::review::hosted_review_timeout_secs(&config); + let limit = crate::review::MAX_LARGE_DIFF_SELECTED_BATCHES + .min(max_hosted_review_batches(&config, false).unwrap()); + let concurrency = crate::review::large_diff_batch_concurrency(&config); + assert_eq!( + ( + generator, + limit, + concurrency, + planned_review_models(&config).len() + ), + (240, 23, 4, 1) + ); + for elapsed in [0, 70, 119, 121, 240] { + let started = Instant::now() - Duration::from_secs(elapsed); + let mut client = LlmClient::build( + &config, + format!("fixture-{}", std::process::id()), + Duration::from_secs(crate::review::HOSTED_LLM_REQUEST_TIMEOUT_SECS), + Some(started + Duration::from_secs(generator)), + Some(started + Duration::from_secs(crate::review::HOSTED_LLM_TOTAL_TIMEOUT_SECS)), + ) + .unwrap(); + client.review_model_timeout = Some(Duration::from_secs( + crate::review::LARGE_DIFF_LLM_REQUEST_TIMEOUT_SECS, + )); + let admitted = client.ensure_hosted_review_schedule(&config, limit, false, concurrency); + if elapsed < 120 { + admitted.unwrap(); + let slot = client.scheduled_review_model_slot().unwrap(); + assert!(slot >= RETRY_ATTEMPT_RESERVE); + assert!( + slot.as_secs() + <= (generator + - elapsed + - crate::review::HOSTED_REVIEW_SCHEDULING_RESERVE_SECS) + / 6 + ); + } else { + let error = admitted.unwrap_err(); + assert!(format!("{error:#}").contains(if elapsed == 240 { + "deadline has already elapsed" + } else { + "no provider request was made" + })); + } + } + } + + #[test] + fn packed_source_bounds_do_not_admit_oversized_decorated_requests() { + let config = Config { + model: "openai/gpt-5-mini".into(), + scorer_enabled: false, + uncertainty_resolution: false, + concise_findings: false, + ..Config::default() + }; + let mut client = LlmClient::build( + &config, + format!("fixture-{}", std::process::id()), + Duration::from_secs(1), + None, + None, + ) + .unwrap(); + client.request_decorations.hosted_price_bounds = Some(Arc::new(HashMap::from([( + config.model.clone(), + ModelPriceBound { + model: config.model.clone(), + input_micros_per_million_tokens: 1, + output_micros_per_million_tokens: 1, + }, + )]))); + let title = "Synthetic \"title\" 界"; + let context = crate::prompt::PrContext { + repo: Some("example/project"), + title: Some(title), + body: Some("Synthetic body with \\ and Unicode"), + incremental: false, + content_policy: false, + change_context: None, + }; + let system = crate::prompt::system_prompt(&config, trusted_date()); + for budget in [4096, 32768] { + let source = format!( + "diff --git a/src/payload.rs b/src/payload.rs\n--- a/src/payload.rs\n+++ b/src/payload.rs\n@@ -0,0 +1 @@\n+{}\n", + "界\0\"\\".repeat(budget / 12) + ); + let parsed = crate::diff::parse(&source); + let plan = crate::diff::render_review_batches(&parsed, &[], &[], budget, 1024); + assert!(!plan.incomplete); + for batch in plan.batches { + assert!(serde_json::to_vec(&batch).unwrap().len() - 2 <= budget); + let user = crate::prompt::user_prompt(&context, &batch, config.max_findings); + let first = vec![format!( + "{user}\nSynthetic feedback: preserve exact citation grounding." + )]; + let later = vec![format!("{user}\nSynthetic later-request context.")]; + let outputs = [REVIEW_MAX_TOKENS]; + let preflight = |first_users: &[String], later_users: &[String]| { + client.preflight_review_plan_with_output_limits( + &config, + 1, + &system, + ReviewPreflightPrompts { + first_users, + later_users, + output_tokens: &outputs, + scorer_system: "unused", + current_utc_date: trusted_date(), + }, + ReviewPlanSchedule { + planner: None, + batch_concurrency: 1, + }, + ) + }; + preflight(&first, &later).unwrap(); + let oversized = vec![format!( + "{user}\nSynthetic feedback: {}", + "\0".repeat(32768) + )]; + for (first_users, later_users) in [(&oversized, &later), (&first, &oversized)] { + let error = preflight(first_users, later_users).unwrap_err(); + assert!( + error.downcast_ref::().is_some(), + "{error:#}" + ); + } + let body = client.request_body( + &config.model, + &system, + &first[0], + REVIEW_MAX_TOKENS, + 0.1, + LlmPhase::Review, + ); + assert_eq!( + client + .planned_request_bytes( + &config.model, + &system, + &first[0], + REVIEW_MAX_TOKENS, + 0.1, + LlmPhase::Review + ) + .unwrap(), + serde_json::to_vec(&body).unwrap().len() + ); + // The independent output-retry expansion remains subject to the same context check. + let initial_bytes = client + .planned_request_bytes( + &config.model, + &system, + &user, + REVIEW_MAX_TOKENS, + 0.1, + LlmPhase::Review, + ) + .unwrap(); + let retry_user = format!( + "{user}{}", + "x".repeat( + conservative_context_tokens(&config.model) + - REVIEW_MAX_OUTPUT_TOKENS as usize + + 1 + - initial_bytes + ) + ); + client + .ensure_review_request_fits_context( + &config.model, + &client.request_body( + &config.model, + &system, + &retry_user, + REVIEW_MAX_TOKENS, + 0.1, + LlmPhase::Review, + ), + REVIEW_MAX_TOKENS, + ) + .unwrap(); + assert!( + client + .planned_review_request_exposure( + &config.model, + &system, + &retry_user, + REVIEW_MAX_TOKENS + ) + .is_err() + ); + } + } + } + #[test] fn hosted_preflight_counts_exact_hostile_json_expansion() { let config = Config { @@ -9539,7 +9917,7 @@ mod tests { #[test] fn extracts_adjudication_array_from_fenced_output() { let text = "Here you go:\n```json\n[{\"candidateId\":\"candidate\",\"status\":\"unresolved\"}]\n```"; - let results = parse_adjudication_results(text, 1, false).unwrap(); + let results = parse_adjudication_results(text, 1, None).unwrap(); assert_eq!(results.len(), 1); assert_eq!(results[0].candidate_id, "candidate"); } @@ -9581,6 +9959,652 @@ mod tests { assert_eq!(adjudication_count(usize::MAX), json!("overflow")); } + struct AdjudicationFixture { + corpus: &'static str, + candidates: Vec, + ids: Vec, + receipt: crate::adjudication::DiffCorpusReceipt, + valid: AdjudicationResult, + cfg: Config, + fresh: usize, + } + + impl AdjudicationFixture { + fn new(context: bool, fresh: usize) -> Self { + let corpus = "diff --git a/src/access.js b/src/access.js\n--- a/src/access.js\n+++ b/src/access.js\n@@ -1,4 +1,4 @@\n function authorized(age) {\n- const ttl = 10;\n+ const ttl = 100000;\n return age < ttl;\n }\n"; + let candidate = Finding { + path: "src/access.js".into(), line: if context { 3 } else { 2 }, + end_line: None, severity: crate::envelope::Severity::Error, kind: Kind::Risk, + confidence: 0.9, generator_confidence: None, scorer_confidence: None, + generator_kind: None, scorer_kind: None, scorer_reason: None, + repository_claim: None, machine_claim: None, machine_claim_deferred: false, + title: "Bound authorization lifetime".into(), + body: "The increased lifetime accepts expired access grants. Restore the short lifetime.".into(), + evidence: Some(if context { " return age < ttl;" } else { " const ttl = 100000;" }.into()), id: None, + }; + let candidates = vec![candidate]; + let ids = crate::adjudication::stable_candidate_ids("test-snapshot", &candidates); + let receipt = crate::adjudication::build_diff_corpus_receipt( + "test-snapshot", + corpus, + &candidates, + &ids, + fresh, + ); + let valid = AdjudicationResult { + candidate_id: ids[0].clone(), + status: crate::adjudication::AdjudicationStatus::Confirmed, + revised_title: candidates[0].title.clone(), + revised_body: candidates[0].body.clone(), + evidence: candidates[0].evidence.clone().unwrap(), + duplicate_of: None, + scope: Some(crate::adjudication::ScopeAssessment { + disposition: crate::adjudication::ScopeDisposition::IntroducedOrWorsened, + cause: Some(crate::adjudication::CausalReference::DiffLine( + crate::adjudication::DiffLineReference { + diff_line: 7, + corpus_sha256: receipt.corpus_sha256.clone(), + }, + )), + reason: "The changed lifetime extends authorization validity.".into(), + }), + }; + Self { + corpus, + candidates, + ids, + receipt, + valid, + fresh, + cfg: Config { + model: "provider/model".into(), + scorer_enabled: false, + api_format: ApiFormat::OpenaiCompatible, + ..Config::default() + }, + } + } + + fn validate( + &self, + results: Vec, + ) -> Result { + crate::review::apply_adjudication_results( + "test-snapshot", + &self.candidates, + &self.ids, + results, + self.corpus, + &self.receipt, + &crate::envelope::RepositorySearchReceipt::default(), + self.fresh, + &self.cfg, + &crate::diff::DiffIndex::build(&crate::diff::parse(self.corpus)), + ) + } + + fn output(&self, outcome: usize) -> String { + let mut result = self.valid.clone(); + match outcome { + 0 => {} + 1 => return "invalid JSON".into(), + 2 => { + result.status = crate::adjudication::AdjudicationStatus::Refuted; + result.revised_title.clear(); + result.revised_body.clear(); + result.evidence = "unsupported evidence".into(); + result.scope = None; + } + 3 => result.candidate_id = "unknown-candidate".into(), + 4 => result.scope = None, + _ => unreachable!(), + } + serde_json::to_string(&vec![result]).unwrap() + } + } + + async fn adjudication_server(outputs: Vec) -> MockServer { + let server = MockServer::start().await; + let calls = Arc::new(AtomicU32::new(0)); + Mock::given(method("POST")) + .respond_with(move |_: &wiremock::Request| { + let index = calls.fetch_add(1, Ordering::Relaxed) as usize; + assert!(index < outputs.len(), "unexpected extra adjudication call"); + ResponseTemplate::new(200).set_body_json(json!({ + "choices": [{"finish_reason":"stop", "message":{"content": outputs[index]}}], + "usage":{"prompt_tokens":10,"completion_tokens":2,"cost":0.000003} + })) + }) + .mount(&server) + .await; + server + } + + fn adjudication_client(fixture: &AdjudicationFixture, server: &MockServer) -> LlmClient { + let cfg = Config { + api_base: server.uri(), + ..fixture.cfg.clone() + }; + let client = LlmClient::build( + &cfg, + std::process::id().to_string(), + Duration::from_secs(30), + None, + None, + ) + .unwrap(); + *client.http.lock().unwrap() = Some(reqwest::Client::new()); + client + } + + #[tokio::test] + async fn adjudication_recovery_generated_outcomes_share_one_correction() { + // Exhaustive generated outcome pairs exercise the real parser and the + // complete caller validator, including its applied-result guard. + for context in [false, true] { + for fresh in [0, 1] { + let fixture = AdjudicationFixture::new(context, fresh); + for first in 0..5 { + for second in 0..5 { + let first_valid = first == 0 || (first == 4 && !context); + let eligible = (1..=3).contains(&first) || (first == 4 && context); + let second_valid = second == 0 || (second == 4 && !context); + let server = adjudication_server(vec![ + fixture.output(first), + fixture.output(second), + ]) + .await; + let client = adjudication_client(&fixture, &server); + let original = serde_json::to_value(&fixture.candidates).unwrap(); + let user = "original immutable evidence"; + let result = client + .adjudicate_findings( + &fixture.cfg.model, + "system", + user, + 1, + Duration::from_secs(60), + |results| fixture.validate(results), + ) + .await; + let count = if first_valid || !eligible { 1 } else { 2 }; + let requests = server.received_requests().await.unwrap(); + assert_eq!(requests.len(), count, "{context}/{fresh}/{first}/{second}"); + assert_eq!( + result.is_ok(), + first_valid || (eligible && second_valid), + "{context}/{fresh}/{first}/{second}: {result:?}" + ); + assert_eq!(serde_json::to_value(&fixture.candidates).unwrap(), original); + let (usage, events, incidents) = match &result { + Ok(review) => { + assert_eq!(review.application.kept.len(), 1); + assert!(review.application.suppressed.is_empty()); + assert!(review.application.resolved_indices.is_empty()); + assert_eq!( + review.application.kept[0].severity, + crate::envelope::Severity::Error + ); + assert_eq!(review.model_incidents.len(), count - 1); + assert!(review.model_incidents.iter().all(|i| i.recovered + && i.recovery == Some(ModelIncidentRecovery::Repair))); + (review.usage, &review.model_usage, &review.model_incidents) + } + Err(error) => { + assert!(error.model_incidents.iter().all(|i| !i.recovered)); + assert_eq!(error.model_incidents.len(), count); + (error.usage(), &error.model_usage, &error.model_incidents) + } + }; + assert_eq!(usage.prompt_tokens, 10 * count as u64); + assert_eq!(usage.completion_tokens, 2 * count as u64); + assert_eq!(usage.cost_micros, Some(3 * count as u64)); + assert_eq!(events.len(), count); + assert_eq!(client.admission.lock().unwrap().attempts, count); + assert_eq!( + incidents.len(), + if result.is_ok() { count - 1 } else { count } + ); + for (ordinal, event) in events.iter().enumerate() { + assert_eq!(event.call_ordinal, Some(ordinal as u32 + 1)); + assert_eq!(event.attempt, Some(1)); + assert!(event.accounting_complete); + } + if count == 2 { + assert_eq!( + events[1].phase, + Some(if first == 1 { + ModelUsagePhase::SchemaRepair + } else { + ModelUsagePhase::SemanticRetry + }) + ); + let mut expected: serde_json::Value = + serde_json::from_slice(&requests[0].body).unwrap(); + expected["messages"][1]["content"] = + json!(format!("{user}{ADJUDICATION_CORRECTION_SUFFIX}")); + assert_eq!( + serde_json::from_slice::(&requests[1].body) + .unwrap(), + expected + ); + } + } + } + } + } + } + + #[tokio::test] + async fn adjudication_recovery_residual_caps_are_atomic_and_do_not_change_initial_admission() { + for (cap, outcome) in (0..7).flat_map(|cap| [(cap, 2), (cap, 3)]) { + let fixture = AdjudicationFixture::new(false, 1); + let server = adjudication_server(vec![fixture.output(outcome)]).await; + let mut client = adjudication_client(&fixture, &server); + client.request_decorations.hosted_price_bounds = Some(Arc::new(HashMap::from([( + fixture.cfg.model.clone(), + ModelPriceBound { + model: fixture.cfg.model.clone(), + input_micros_per_million_tokens: 1, + output_micros_per_million_tokens: 1, + }, + )]))); + let competing = client.clone(); + let error = client + .adjudicate_findings( + &fixture.cfg.model, + "system", + "user", + 1, + Duration::from_secs(60), + |results| { + let mut admission = competing.admission.lock().unwrap(); + match cap { + 0 => admission.attempts = MAX_PROVIDER_ATTEMPTS, + 1 => admission.input_bytes = MAX_PROVIDER_INPUT_BYTES, + 2 => { + admission.output_token_exposure = MAX_PROVIDER_OUTPUT_TOKEN_EXPOSURE + } + 3 => admission.token_exposure_upper_bound = MAX_PROVIDER_TOKEN_EXPOSURE, + 4 => { + admission.projected_cost_exposure_micros = + HOSTED_OPERATION_COST_CAP_MICROS + } + 5 => admission.reported_cost_micros = HOSTED_OPERATION_COST_CAP_MICROS, + 6 => admission.reported_token_spend = MAX_REPORTED_TOKEN_SPEND, + _ => unreachable!(), + } + drop(admission); + fixture.validate(results) + }, + ) + .await + .unwrap_err(); + assert_eq!( + server.received_requests().await.unwrap().len(), + 1, + "cap={cap}" + ); + assert_eq!(error.usage().cost_micros, Some(3)); + assert_eq!(error.model_usage.len(), 1); + assert_eq!(error.model_incidents.len(), 1); + assert_eq!( + error.model_incidents[0].category, + ModelIncidentCategory::InvalidOutput + ); + assert!(!error.model_incidents[0].recovered); + assert!(error.usage_accounting_complete()); + assert!(!error.is_provider()); + assert!(!error.is_deadline_exceeded()); + assert!(!error.is_timeout()); + assert!( + error + .error + .downcast_ref::() + .is_some() + ); + assert_eq!( + error + .error + .downcast_ref::(), + Some(&if outcome == 3 { + crate::adjudication::AdjudicationViolation::UnknownCandidateIdentity + } else { + crate::adjudication::AdjudicationViolation::ContradictoryRefutation + }) + ); + assert!(format!("{error:#}").contains("no residual admitted capacity")); + assert!(format!("{error:#}").contains(match cap { + 0 => "model provider attempt hard cap", + 1 => "model provider input hard cap", + 2 => "model provider output exposure hard cap", + 3 => "model token exposure exceeded", + 4 | 5 => "model provider projected cost exposure exceeded", + 6 => "model token spend exhausted", + _ => unreachable!(), + })); + assert_eq!( + crate::review::adjudication_failure_finding(&error).title, + "Model output could not be validated" + ); + } + } + + #[tokio::test] + async fn adjudication_recovery_deadline_and_cancellation_never_accept_an_application() { + for outcome in [0, 2, 3] { + let fixture = AdjudicationFixture::new(false, 1); + let server = adjudication_server(vec![fixture.output(outcome)]).await; + let client = adjudication_client(&fixture, &server); + let error = client + .adjudicate_findings( + &fixture.cfg.model, + "system", + "user", + 1, + Duration::from_millis(100), + |results| { + std::thread::sleep(Duration::from_millis(110)); + fixture.validate(results) + }, + ) + .await + .unwrap_err(); + assert!(error.is_deadline_exceeded()); + assert!(!error.is_provider()); + assert!(error.is_timeout()); + assert!(error.error.downcast_ref::().is_some()); + assert!(format!("{error:#}").contains("LLM total deadline exceeded")); + assert_eq!( + error + .error + .downcast_ref::(), + match outcome { + 2 => Some(&crate::adjudication::AdjudicationViolation::ContradictoryRefutation), + 3 => + Some(&crate::adjudication::AdjudicationViolation::UnknownCandidateIdentity), + _ => None, + } + ); + assert_eq!(error.model_usage.len(), 1); + assert!(error.usage_accounting_complete()); + assert_eq!(server.received_requests().await.unwrap().len(), 1); + assert_eq!(error.usage().cost_micros, Some(3)); + assert!(error.model_incidents.iter().all(|i| !i.recovered)); + } + let fixture = AdjudicationFixture::new(false, 1); + let server = adjudication_server(vec![fixture.output(2), fixture.output(0)]).await; + let client = adjudication_client(&fixture, &server); + let notify = tokio::sync::Notify::new(); + let operation = client.adjudicate_findings( + &fixture.cfg.model, + "system", + "user", + 1, + Duration::from_secs(60), + |results| { + notify.notify_one(); + fixture.validate(results) + }, + ); + tokio::pin!(operation); + tokio::select! { biased; + _ = notify.notified() => {}, + _ = &mut operation => panic!("cancelled adjudication returned an application"), + } + assert!(server.received_requests().await.unwrap().len() <= 2); + } + + #[tokio::test] + async fn adjudication_recovery_preparation_refusals_retain_both_causes() { + for prompt_bound in [false, true] { + let mut fixture = AdjudicationFixture::new(false, 1); + // Reach the prompt bound within an already supported context window. + fixture.cfg.model = "openai/gpt-5-mini".into(); + let server = adjudication_server(vec![fixture.output(2)]).await; + let client = adjudication_client(&fixture, &server); + let user = if prompt_bound { + "x".repeat(crate::adjudication::MAX_ADJUDICATION_PROMPT_BYTES) + } else { + "user".to_owned() + }; + let error = client + .adjudicate_findings( + &fixture.cfg.model, + "system", + &user, + 1, + if prompt_bound { + Duration::from_secs(60) + } else { + Duration::from_secs(1) + }, + |results| fixture.validate(results), + ) + .await + .unwrap_err(); + assert_eq!(server.received_requests().await.unwrap().len(), 1); + assert_eq!( + error + .error + .downcast_ref::(), + Some(&crate::adjudication::AdjudicationViolation::ContradictoryRefutation) + ); + assert!(format!("{error:#}").contains(if prompt_bound { + "finding adjudication correction exceeds its admitted input bound" + } else { + "finding adjudication has no viable correction slot before its deadline" + })); + assert!(!error.is_provider()); + assert!(!error.is_deadline_exceeded()); + assert!(!error.is_timeout()); + assert_eq!(error.usage().cost_micros, Some(3)); + assert_eq!(error.model_usage.len(), 1); + assert!(error.usage_accounting_complete()); + assert_eq!(error.model_incidents.len(), 1); + assert_eq!( + error.model_incidents[0].category, + ModelIncidentCategory::InvalidOutput + ); + assert!(!error.model_incidents[0].recovered); + assert_eq!( + crate::review::adjudication_failure_finding(&error).title, + "Model output could not be validated" + ); + } + } + + #[tokio::test] + async fn adjudication_recovery_provider_failures_never_create_another_semantic_attempt() { + for (repair, context, unknown) in [ + (false, false, false), + (true, false, false), + (true, true, false), + (true, false, true), + ] { + let fixture = AdjudicationFixture::new(context, 1); + let server = MockServer::start().await; + if repair { + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(200).set_body_json(json!({ + "choices": [{"finish_reason":"stop","message":{"content":fixture.output(if unknown { 3 } else if context { 4 } else { 2 })}}], + "usage":{"prompt_tokens":10,"completion_tokens":2,"cost":0.000003} + }))) + .with_priority(1).up_to_n_times(1).mount(&server).await; + } + Mock::given(method("POST")) + .respond_with(ResponseTemplate::new(403)) + .with_priority(2) + .expect(1) + .mount(&server) + .await; + let client = adjudication_client(&fixture, &server); + let error = client + .adjudicate_findings( + &fixture.cfg.model, + "system", + "user", + 1, + Duration::from_secs(60), + |results| fixture.validate(results), + ) + .await + .unwrap_err(); + assert!(error.is_provider()); + assert!(!error.is_deadline_exceeded()); + assert!(!error.is_timeout()); + assert_eq!( + error.error.downcast_ref::().unwrap().0, + reqwest::StatusCode::FORBIDDEN + ); + assert!(error.error.downcast_ref::().is_some()); + if !repair { + assert!( + error + .error + .downcast_ref::() + .is_none() + ); + } + assert_eq!( + server.received_requests().await.unwrap().len(), + if repair { 2 } else { 1 } + ); + assert_eq!( + error + .model_incidents + .iter() + .filter(|i| i.category == ModelIncidentCategory::InvalidOutput) + .count(), + usize::from(repair) + ); + assert!(error.model_incidents.iter().all(|i| !i.recovered)); + let finding = crate::review::adjudication_failure_finding(&error); + assert_eq!( + finding.path, + if repair { + crate::envelope::OPERATIONAL_PATH + } else { + crate::envelope::PROVIDER_PATH + } + ); + if repair { + assert_eq!(finding.title, "Model output could not be validated"); + assert_eq!(error.usage().cost_micros, Some(3)); + assert_eq!(error.model_usage.len(), 2); + assert_eq!( + error.model_usage[1].phase, + Some(ModelUsagePhase::SemanticRetry) + ); + assert!(!error.usage_accounting_complete()); + assert_eq!( + error + .error + .downcast_ref::(), + Some(&if unknown { + crate::adjudication::AdjudicationViolation::UnknownCandidateIdentity + } else if context { + crate::adjudication::AdjudicationViolation::MissingCausalReference + } else { + crate::adjudication::AdjudicationViolation::ContradictoryRefutation + }) + ); + assert!(format!("{error:#}").contains("403")); + } + } + } + + #[tokio::test] + async fn adjudication_recovery_only_typed_rule_errors_take_the_semantic_edge() { + let fixture = AdjudicationFixture::new(true, 1); + let missing = serde_json::from_str::>(&fixture.output(4)).unwrap(); + let error = fixture.validate(missing).unwrap_err(); + assert_eq!( + error.downcast_ref::(), + Some(&crate::adjudication::AdjudicationViolation::MissingCausalReference) + ); + let fixture = AdjudicationFixture::new(false, 0); + let unknown = serde_json::from_str::>(&fixture.output(3)).unwrap(); + let error = fixture.validate(unknown).unwrap_err(); + assert_eq!( + error.downcast_ref::(), + Some(&crate::adjudication::AdjudicationViolation::UnknownCandidateIdentity) + ); + assert_eq!( + error.to_string(), + "adjudication returned an unknown candidate identity" + ); + let refuted = serde_json::from_str::>(&fixture.output(2)).unwrap(); + let error = crate::adjudication::validate_results( + "test-snapshot", + &fixture.candidates, + &fixture.ids, + &refuted, + fixture.corpus, + &fixture.receipt, + &crate::envelope::RepositorySearchReceipt::default(), + ) + .unwrap_err(); + assert_eq!( + error.downcast_ref::(), + Some(&crate::adjudication::AdjudicationViolation::ContradictoryRefutation) + ); + let error = fixture.validate(refuted).unwrap_err(); + assert_eq!( + error.downcast_ref::(), + Some(&crate::adjudication::AdjudicationViolation::ContradictoryRefutation) + ); + let expected_cause = error.to_string(); + let server = adjudication_server(vec![fixture.output(0)]).await; + let client = adjudication_client(&fixture, &server); + let error = client + .adjudicate_findings( + &fixture.cfg.model, + "system", + "user", + 1, + Duration::from_secs(60), + |_| Err(anyhow!(error.to_string())), + ) + .await + .unwrap_err(); + assert_eq!(server.received_requests().await.unwrap().len(), 1); + assert_eq!(error.model_incidents.len(), 1); + assert!(format!("{error:#}").contains(&expected_cause)); + } + + #[test] + fn adjudication_recovery_prompt_bound_and_untyped_errors_fail_closed() { + let maximum = crate::adjudication::MAX_ADJUDICATION_PROMPT_BYTES; + assert_eq!( + adjudication_correction_user( + &"x".repeat(maximum - ADJUDICATION_CORRECTION_SUFFIX.len()) + ) + .unwrap() + .len(), + maximum + ); + assert!( + adjudication_correction_user( + &"x".repeat(maximum - ADJUDICATION_CORRECTION_SUFFIX.len() + 1) + ) + .is_err() + ); + assert!(adjudication_correction_user(&"é".repeat(maximum / 2)).is_err()); + let violation = crate::adjudication::AdjudicationViolation::ContradictoryRefutation; + assert!( + anyhow!(violation.to_string()) + .downcast_ref::() + .is_none() + ); + assert!( + anyhow!(violation) + .context("outer context") + .downcast_ref::() + .is_some() + ); + } + #[tokio::test] async fn adjudication_provenance_pre_request_failures_have_no_attempt_usage() { for (oversized, code) in [(true, "contextExceeded"), (false, "deadline")] { @@ -9607,7 +10631,14 @@ mod tests { "[]".to_owned() }; let error = match client - .adjudicate_findings(&config.model, "system", &user, 1, Duration::from_secs(1)) + .adjudicate_findings( + &config.model, + "system", + &user, + 1, + Duration::from_secs(1), + |_| unreachable!(), + ) .await { Ok(_) => panic!("pre-request rejection unexpectedly succeeded"), @@ -9670,7 +10701,7 @@ mod tests { #[test] fn rejects_adjudication_output_without_a_json_array() { assert_eq!( - parse_adjudication_results("I could not adjudicate this finding.", 1, false) + parse_adjudication_results("I could not adjudicate this finding.", 1, None) .unwrap_err(), "no JSON array found" ); @@ -11821,8 +12852,25 @@ mod tests { Ok(_) => panic!("oversized request reached the provider transport"), Err(error) => error, }; - assert!(error.to_string().contains("per-request cap")); assert!(server.received_requests().await.unwrap().is_empty()); + assert_eq!(client.admission.lock().unwrap().attempts, 0); + let expected = format!( + "model provider request needs {} serialized bytes, exceeding the {MAX_PROVIDER_REQUEST_BYTES} byte per-request cap", + MAX_PROVIDER_REQUEST_BYTES + 1 + ); + assert_eq!(error.root_cause().to_string(), expected); + assert!(error.downcast_ref::().is_some()); + let error = ModelError::new(classify_chat_error(error), Usage::default(), true); + assert!( + error + .error + .downcast_ref::() + .is_some() + ); + assert!(!error.is_provider()); + assert!(!error.is_timeout()); + assert!(!error.is_deadline_exceeded()); + assert!(format!("{error:#}").contains(&expected)); } #[test] diff --git a/src/review.rs b/src/review.rs index 38ab9dd..febf2db 100644 --- a/src/review.rs +++ b/src/review.rs @@ -2029,13 +2029,14 @@ async fn review_diff_at( } else { MAX_LARGE_DIFF_SELECTED_BATCHES }; - let mut batches = diff::spool_model_batches_with_synthesis_budget( + let mut batches = diff::spool_model_batches_for_review( &mut prepared, batch_budgets.source, batch_budgets.synthesis, MAX_REVIEW_MANIFEST_BYTES.min(batch_budgets.source / 3), force_model || pr_desc_lines > 0, large_diff_selected_limit, + args.bounded, )?; index.add_change_metadata(batches.metadata_count); if batches.count == 0 { @@ -2748,6 +2749,20 @@ async fn review_diff_at( &adjudication_user, all_adjudication_candidates.len(), Duration::from_secs(FINDING_ADJUDICATION_TIMEOUT_SECS), + |results| { + apply_adjudication_results( + &snapshot_id, + &all_adjudication_candidates, + &candidate_ids, + results, + diff_snapshot.as_str(), + &diff_receipt, + receipt, + fresh_candidate_count, + cfg, + &index, + ) + }, ) .await, ), @@ -2770,69 +2785,22 @@ async fn review_diff_at( model_incidents.extend(adjudicated.model_incidents); usage_accounting_complete &= adjudicated.usage_accounting_complete; - match crate::adjudication::apply_results( - &snapshot_id, - all_adjudication_candidates.clone(), - candidate_ids.clone(), - adjudicated.results, - diff_snapshot.as_str(), - &diff_receipt, - receipt, - ).and_then(|mut application| { - suppress_fresh_unresolved_repository_claims( - &mut application, - fresh_candidate_count, + let application = adjudicated.application; + if !application.recovered_duplicate_indices.is_empty() { + eprintln!( + "postil: repaired {} non-confirmed adjudication duplicate reference(s) to unresolved; applying candidate publication policy", + application.recovered_duplicate_indices.len() ); - suppress_fresh_lockfile_platform_claims( - &mut application, - fresh_candidate_count, - &index, - ); - anyhow::ensure!( - application.invalid_refutation_indices.iter().all(|candidate_index| { - cfg.scorer_enabled() - && *candidate_index < fresh_candidate_count - && application.kept_indices.contains(candidate_index) - }), - "refuted adjudication must cite candidate-specific contradictory evidence" - ); - Ok(application) - }) { - Ok(application) => { - if !application.recovered_duplicate_indices.is_empty() { - eprintln!( - "postil: repaired {} non-confirmed adjudication duplicate reference(s) to unresolved; applying candidate publication policy", - application.recovered_duplicate_indices.len() - ); - model_incidents.push(ModelIncident { - phase: ModelIncidentPhase::Scorer, - category: ModelIncidentCategory::InvalidOutput, - recovered: true, - recovery: Some(crate::envelope::ModelIncidentRecovery::Repair), - }); - } - application - }, - Err(error) => { - adjudication_incomplete = true; - review_trust = filter::ReviewTrust::Failed; - eprintln!( - "postil: finding adjudication validation failed; preserving all generated findings: {error:#}" - ); - model_incidents.push(ModelIncident { - phase: ModelIncidentPhase::Scorer, - category: ModelIncidentCategory::InvalidOutput, - recovered: false, - recovery: None, - }); - adjudication_failure = Some(fail_closed_finding( - "finding adjudication output did not satisfy its admitted contract", - )); - preserve_unadjudicated_findings( - all_adjudication_candidates, - ) - } + model_incidents.push(ModelIncident { + phase: ModelIncidentPhase::Scorer, + category: ModelIncidentCategory::InvalidOutput, + recovered: true, + recovery: Some( + crate::envelope::ModelIncidentRecovery::Repair, + ), + }); } + application } Some(Err(error)) => { adjudication_incomplete = true; @@ -2844,15 +2812,8 @@ async fn review_diff_at( model_usage.extend_from_slice(error.model_usage()); model_incidents.extend_from_slice(error.model_incidents()); usage_accounting_complete &= error.usage_accounting_complete(); - adjudication_failure = Some(if error.is_provider() { - crate::envelope::provider_error_finding( - "finding adjudication did not complete", - ) - } else { - fail_closed_finding( - "finding adjudication output did not satisfy its admitted contract", - ) - }); + adjudication_failure = + Some(adjudication_failure_finding(&error)); preserve_unadjudicated_findings(all_adjudication_candidates) } None => { @@ -3918,6 +3879,56 @@ fn promote_incomplete_adjudication_receipts( } } +pub(crate) fn adjudication_failure_finding(error: &crate::llm::ModelError) -> Finding { + if classify_exhausted_scorer_failure(error.model_incidents(), error.is_provider()) + == ReviewFailureKind::Provider + { + crate::envelope::provider_error_finding("finding adjudication did not complete") + } else { + fail_closed_finding("finding adjudication output did not satisfy its admitted contract") + } +} + +#[allow(clippy::too_many_arguments)] +pub(crate) fn apply_adjudication_results( + snapshot_id: &str, + candidates: &[Finding], + candidate_ids: &[String], + results: Vec, + corpus: &str, + diff_receipt: &crate::adjudication::DiffCorpusReceipt, + receipt: &crate::envelope::RepositorySearchReceipt, + fresh_candidate_count: usize, + cfg: &Config, + index: &diff::DiffIndex, +) -> Result { + crate::adjudication::apply_results( + snapshot_id, + candidates.to_vec(), + candidate_ids.to_vec(), + results, + corpus, + diff_receipt, + receipt, + ) + .and_then(|mut application| { + suppress_fresh_unresolved_repository_claims(&mut application, fresh_candidate_count); + suppress_fresh_lockfile_platform_claims(&mut application, fresh_candidate_count, index); + anyhow::ensure!( + application + .invalid_refutation_indices + .iter() + .all(|candidate_index| { + cfg.scorer_enabled() + && *candidate_index < fresh_candidate_count + && application.kept_indices.contains(candidate_index) + }), + crate::adjudication::AdjudicationViolation::ContradictoryRefutation + ); + Ok(application) + }) +} + fn preserve_unadjudicated_findings( findings: Vec, ) -> crate::adjudication::AdjudicationApplication { @@ -4894,6 +4905,8 @@ mod tests { assert!(envelope.findings.iter().any(|finding| { finding.path == crate::envelope::OPERATIONAL_PATH && finding.title == "Large review coverage is incomplete" + && finding.body.contains("1 normalized hunk unreviewed") + && finding.body.contains("cannot be trusted as a pass") })); assert_eq!( envelope diff --git a/tests/e2e.rs b/tests/e2e.rs index 9b0a25b..dfa187e 100644 --- a/tests/e2e.rs +++ b/tests/e2e.rs @@ -195,20 +195,66 @@ impl Respond for AllUnresolvedAdjudicator { } } +const ADJUDICATION_CORRECTION_SUFFIX: &str = "\n\nReturn the complete adjudication JSON array for the original candidate IDs. Correct the rejected output using only the supplied evidence. A refutation needs candidate-specific contradictory evidence, not its own citation or unsupported prose. A context-anchored confirmation needs an exact causal change reference. Preserve all structural, scope and publication rules. Never invent evidence or references, silently drop a candidate, or fabricate a verdict."; + +fn adjudication_payload(request: &Value) -> Value { + let text = request["messages"].as_array().unwrap().last().unwrap()["content"] + .as_str() + .unwrap(); + let mut stream = serde_json::Deserializer::from_str(text).into_iter::(); + let payload = stream.next().unwrap().unwrap(); + let suffix = &text[stream.byte_offset()..]; + assert!(suffix.is_empty() || suffix == ADJUDICATION_CORRECTION_SUFFIX); + payload +} + +fn assert_adjudication_calls(requests: &[Request], phases: &[&str], envelope: &Value) { + let adjudicators = requests + .iter() + .filter(|request| request_system_contains(request, "single finding adjudicator")) + .collect::>(); + assert_eq!(adjudicators.len(), phases.len()); + let initial: Value = adjudicators[0].body_json().unwrap(); + adjudication_payload(&initial); + for (index, request) in adjudicators.iter().enumerate() { + let actual: Value = request.body_json().unwrap(); + let mut expected = initial.clone(); + if index > 0 { + let content = expected["messages"] + .as_array_mut() + .unwrap() + .last_mut() + .unwrap(); + content["content"] = json!(format!( + "{}{}", + content["content"].as_str().unwrap(), + ADJUDICATION_CORRECTION_SUFFIX + )); + } + assert_eq!(actual, expected); + } + // All calls in these fixtures return known usage; one source precedes adjudication. + assert_eq!(envelope["usageAccountingComplete"], true); + assert_model_usage_matches_aggregate(envelope); + let events = envelope["modelUsage"].as_array().unwrap(); + assert_eq!(events.len(), requests.len()); + for (index, phase) in phases.iter().enumerate() { + assert_eq!(events[index + 1]["role"], "findingScorer"); + assert_eq!(events[index + 1]["phase"], *phase); + assert_eq!(events[index + 1]["accountingComplete"], true); + assert_eq!(events[index + 1]["promptTokens"], 30); + assert_eq!(events[index + 1]["completionTokens"], 10); + assert_eq!(events[index + 1]["costMicros"], 45); + } +} + #[derive(Clone, Copy)] struct AllRefutedAdjudicator; impl Respond for AllRefutedAdjudicator { fn respond(&self, request: &Request) -> ResponseTemplate { let request_body: Value = request.body_json().unwrap(); - let payload: Value = serde_json::from_str( - request_body["messages"] - .as_array() - .and_then(|messages| messages.last()) - .and_then(|message| message["content"].as_str()) - .unwrap(), - ) - .unwrap(); + let payload = adjudication_payload(&request_body); let results = payload["candidates"] .as_array() .unwrap() @@ -3779,12 +3825,7 @@ async fn ungrounded_refutation_keeps_candidates_for_the_existing_scorer() { .and(body_string_contains("single finding adjudicator")) .respond_with(move |request: &Request| { let request: Value = request.body_json().unwrap(); - let payload: Value = serde_json::from_str( - request["messages"].as_array().unwrap().last().unwrap()["content"] - .as_str() - .unwrap(), - ) - .unwrap(); + let payload = adjudication_payload(&request); let candidate = &payload["candidates"][0]; ResponseTemplate::new(200).set_body_json(scorer_text( &json!([{ @@ -3796,7 +3837,7 @@ async fn ungrounded_refutation_keeps_candidates_for_the_existing_scorer() { )) }) .with_priority(1) - .expect(1) + .expect(if scorer_disabled { 2 } else { 1 }) .mount(&server) .await; Mock::given(method("POST")) @@ -3842,6 +3883,15 @@ async fn ungrounded_refutation_keeps_candidates_for_the_existing_scorer() { "{label}: {stderr}" ); let requests = server.received_requests().await.unwrap(); + assert_adjudication_calls( + &requests, + if scorer_disabled { + &["initial", "semanticRetry"] + } else { + &["initial"] + }, + &envelope, + ); let scorers = requests .iter() .filter(|request| request_system_contains(request, "independent second-model scorer")) @@ -4191,6 +4241,7 @@ async fn malformed_adjudication_output_blocks_under_advisory_provider_policy() { .and(body_string_contains("single finding adjudicator")) .respond_with(ResponseTemplate::new(200).set_body_json(scorer_text("not json"))) .with_priority(1) + .expect(2) .mount(&server) .await; mock_review( @@ -4226,12 +4277,19 @@ async fn malformed_adjudication_output_blocks_under_advisory_provider_policy() { .filter_map(|line| line.strip_prefix("postil: adjudication-output ")) .map(|line| serde_json::from_str(line).unwrap()) .collect(); - assert_eq!(records.len(), 2); - assert_eq!(records[0]["event"], "callStarted"); - assert_eq!(records[1]["event"], "parserRejected"); - assert_eq!(records[1]["code"], "noArray"); - assert_eq!(records[1]["call"], records[0]["call"]); - assert_eq!(records[1]["attempt"], Value::Null); + assert_eq!(records.len(), 4); + for (call, phase) in [(1, "initial"), (2, "schema-repair")] { + let started = &records[(call - 1) * 2]; + let rejected = &records[(call - 1) * 2 + 1]; + assert_eq!(started["event"], "callStarted"); + assert_eq!(rejected["event"], "parserRejected"); + assert_eq!(rejected["code"], "noArray"); + for record in [started, rejected] { + assert_eq!(record["call"], call); + assert_eq!(record["callPhase"], phase); + assert_eq!(record["attempt"], Value::Null); + } + } let envelope: Value = serde_json::from_slice(&out.get_output().stdout).unwrap(); assert!( @@ -4251,6 +4309,21 @@ async fn malformed_adjudication_output_blocks_under_advisory_provider_policy() { .iter() .any(|finding| finding["path"] == ".postil/model-output") ); + let requests = server.received_requests().await.unwrap(); + assert_adjudication_calls(&requests, &["initial", "schemaRepair"], &envelope); + assert!( + requests + .iter() + .all(|request| !request_system_contains(request, "independent second-model scorer")) + ); + let incidents = envelope["modelIncidents"].as_array().unwrap(); + assert_eq!(incidents.len(), 2); + assert!( + incidents + .iter() + .all(|incident| incident["category"] == "invalidOutput" + && incident["recovered"] == false) + ); assert_eq!(envelope["counts"]["warn"], 1); assert_eq!(envelope["counts"]["error"], 1); assert_eq!(envelope["gate"]["failing"], true); @@ -6922,7 +6995,7 @@ async fn compact_lockfile_platform_claim_survives_failed_adjudication() { .and(body_string_contains("single finding adjudicator")) .respond_with(AllRefutedAdjudicator) .with_priority(1) - .expect(1) + .expect(2) .mount(&server) .await; } @@ -7000,6 +7073,9 @@ async fn compact_lockfile_platform_claim_survives_failed_adjudication() { ); assert_eq!(envelope["gate"]["failing"], true); let requests = server.received_requests().await.unwrap(); + if refuted { + assert_adjudication_calls(&requests, &["initial", "semanticRetry"], &envelope); + } assert!( requests.iter().all(|request| !request_system_contains( request, @@ -14392,7 +14468,7 @@ async fn fresh_unresolved_repository_claims_are_suppressed() { .and(body_string_contains("single finding adjudicator")) .respond_with(AllRefutedAdjudicator) .with_priority(1) - .expect(1) + .expect(2) .mount(&server) .await; } @@ -14449,6 +14525,15 @@ async fn fresh_unresolved_repository_claims_are_suppressed() { ); } let requests = server.received_requests().await.unwrap(); + assert_adjudication_calls( + &requests, + if refuted { + &["initial", "semanticRetry"] + } else { + &["initial"] + }, + &envelope, + ); assert!(requests.iter().all(|request| !request_system_contains( request, "independent second-model scorer" @@ -14806,12 +14891,7 @@ async fn scorer_cannot_suppress_an_unresolved_full_rereview_baseline() { .and(body_string_contains("single finding adjudicator")) .respond_with(move |request: &Request| { let request: Value = request.body_json().unwrap(); - let payload: Value = serde_json::from_str( - request["messages"].as_array().unwrap().last().unwrap()["content"] - .as_str() - .unwrap(), - ) - .unwrap(); + let payload = adjudication_payload(&request); let results = payload["candidates"] .as_array() .unwrap() @@ -14827,7 +14907,7 @@ async fn scorer_cannot_suppress_an_unresolved_full_rereview_baseline() { ResponseTemplate::new(200).set_body_json(scorer_content(json!(results))) }) .with_priority(1) - .expect(1) + .expect(if status == "refuted" { 2 } else { 1 }) .mount(&server) .await; let directory = tempfile::tempdir().unwrap(); @@ -14900,6 +14980,15 @@ async fn scorer_cannot_suppress_an_unresolved_full_rereview_baseline() { assert_eq!(envelope["usageAccountingComplete"], true); assert_model_usage_matches_aggregate(&envelope); let requests = server.received_requests().await.unwrap(); + assert_adjudication_calls( + &requests, + if status == "refuted" { + &["initial", "semanticRetry"] + } else { + &["initial"] + }, + &envelope, + ); assert!(requests.iter().all(|request| { !String::from_utf8_lossy(&request.body).contains("independent second-model scorer") })); @@ -16363,7 +16452,7 @@ async fn causal_scope_contract_prevents_context_relabeling_before_scoring() { let mut response = if request_system_contains(request, "single finding adjudicator") { let body: Value = request.body_json().unwrap(); assert!(body["messages"][0]["content"].as_str().unwrap().starts_with("You are Postil's single finding adjudicator. ")); - let payload: Value = serde_json::from_str(body["messages"].as_array().unwrap().last().unwrap()["content"].as_str().unwrap()).unwrap(); + let payload = adjudication_payload(&body); let result = payload["candidates"].as_array().unwrap().iter().map(|candidate| { let mut result = json!({"candidateId":candidate["candidateId"], "status":"confirmed", "revisedTitle":candidate["title"], "revisedBody":"Untrusted input bypasses the authorization boundary. Restore the guard before dispatch.", "evidence":candidate["citedEvidence"], "duplicateOf":null}); if excluded { @@ -16592,14 +16681,43 @@ async fn causal_scope_contract_prevents_context_relabeling_before_scoring() { } assert_eq!(envelope["findings"][0]["scorerConfidence"], 0.99, "{mode}"); } - assert!( - requests - .iter() - .filter(|request| request_system_contains(request, "single finding adjudicator")) - .count() - <= 1, - "{mode}" - ); + let phases: &[&str] = if matches!(mode, "missing" | "unresolved" | "demoted") { + &["initial", "semanticRetry"] + } else { + &["initial"] + }; + if mode == "carried-incremental" { + assert!( + requests + .iter() + .all(|request| !request_system_contains(request, "single finding adjudicator")), + "{mode}" + ); + } else if !scope_failure { + assert_eq!( + requests + .iter() + .filter(|request| request_system_contains( + request, + "single finding adjudicator" + )) + .count(), + phases.len(), + "{mode}" + ); + assert_adjudication_calls(&requests, phases, &envelope); + } else { + assert_eq!( + requests + .iter() + .filter(|request| request_system_contains( + request, + "single finding adjudicator" + )) + .count(), + 1 + ); + } } }