diff --git a/packages/evals/README.md b/packages/evals/README.md index 098ad64..e9020da 100644 --- a/packages/evals/README.md +++ b/packages/evals/README.md @@ -373,9 +373,39 @@ or measured native-plugin activation. Each live run writes a v1 aggregate below the ignored `.plugin-eval-runs/` directory and a sibling `.journal-v1.jsonl`, both mode `0600`. OpenRouter writes four siblings: the v1 aggregate `.json`, `.requested-routing-v1.json`, -`.configuration-v2.json`, and `.journal-v1.jsonl`. Keep all four. Other live -runners write only the aggregate and journal. `--attempts-output` remains an -optional grader companion and does not replace those siblings. +`.configuration-v2.json`, and `.journal-v1.jsonl`. Keep all four. OMP also writes +`.transcripts-v1.jsonl` automatically, in both authentication modes. +Other live runners write only the aggregate and journal. `--attempts-output` +remains an optional grader companion and does not replace those siblings. + +The OMP transcript is a private, mode-`0600` JSONL file. It contains a run header, +one transcript per trial, and a final `report-bound` record with the exact +aggregate report SHA-256. Each transcript records the user prompt, ordered +assistant text, tool calls and results, trial identity, and settled status. +Reasoning and raw provider frames are excluded. Capture does not change grading +or the v1 aggregate and public attempt schemas. + +Supplied credentials and recognizable credential text are redacted before +persistence. Other private text, tool data, and local paths can remain: do not +publish this file or import it into the public leaderboard. Each trial is capped +at 512 messages and 1 MiB, with 64 KiB field limits. Omitted or shortened content +sets `truncated: true`. + +The transcript header is fsynced before credentials load. Completed and partial +trial transcripts are written after session cleanup, including on timeout or +interruption; this is not a per-token crash-recovery log. A killed process can +leave only earlier trial records. Persistence has a five-second callback budget. +A capture failure stops the run before the next dispatch, even if the completed +observation has a failed grading status. An existing execution error or +interruption remains authoritative. The final report binding requires the +complete planned trial set. + +Library callers opt in with `OmpHarnessTrialOptions.onTranscript`. The package +exports `OmpTrialTranscript`, its schemas, and `createOmpTranscriptWriter` for +private storage. The callback receives a separate `AbortSignal` for its +persistence window. Pass that signal to asynchronous storage so a timed-out +callback does not leave a pending writer. Without that callback, library trials +retain the existing non-streaming generation path. The journal writes its run header, fsynced, before credentials load. The header captures suite, catalog, reasoning, account class, selected cases, diff --git a/packages/evals/__tests__/live-cli.test.ts b/packages/evals/__tests__/live-cli.test.ts index 0224fda..ebc2881 100644 --- a/packages/evals/__tests__/live-cli.test.ts +++ b/packages/evals/__tests__/live-cli.test.ts @@ -24,6 +24,7 @@ import { makeLiveEvalConfigurationEvidence } from "../src/configuration"; import { ALPHA_GINA_READ_SERVER_URL } from "../src/server-url"; import type { LiveEvalConfigurationCaptureType } from "../src/configuration"; import { DEFAULT_OPENROUTER_MAX_TOOL_CALLS } from "../src/openrouter"; +import { OMP_TRANSCRIPT_SCHEMA_VERSION } from "../src/omp-transcript"; import type { SanitizedEvalRunReport } from "../src/report"; import aggregateFixture from "../src/fixtures/sanitized-aggregate.json"; import { SanitizedEvalAggregateSchema } from "../src/sanitize"; @@ -847,6 +848,174 @@ describe("live eval CLI subprocess", () => { ), ); + it.effect("rejects preexisting and attempts-colliding OMP transcripts before credentials", () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const pathValue = yield* Config.string("PATH"); + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "live-cli-omp-transcript-" }); + const liveCli = path.join(process.cwd(), "packages/evals/src/bin/live.ts"); + const outputDirectory = path.join(cwd, ".plugin-eval-runs"); + yield* fs.makeDirectory(outputDirectory); + + const preexistingRunId = "run-preexisting"; + const preexistingTranscript = path.join( + outputDirectory, + `omp_harness-cand-1-${preexistingRunId}.transcripts-v1.jsonl`, + ); + yield* fs.writeFileString(preexistingTranscript, "foreign bytes\n", { + flag: "wx", + mode: 0o600, + }); + const preexistingChild = yield* ChildProcess.make( + "bun", + [ + liveCli, + ...requiredFlags("omp", ["--provider", "openai"]).map((value) => + value === "run-1" ? preexistingRunId : value, + ), + ], + { + cwd, + env: { PATH: pathValue }, + extendEnv: false, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }, + ); + const [preexistingStdout, preexistingStderr, preexistingExitCode] = yield* Effect.all( + [ + collectBoundedUtf8Output(preexistingChild.stdout, 65_536), + collectBoundedUtf8Output(preexistingChild.stderr, 65_536), + preexistingChild.exitCode, + ], + { concurrency: "unbounded" }, + ); + const preexistingOutput = `${preexistingStdout.text}\n${preexistingStderr.text}`; + assert.notStrictEqual(preexistingExitCode, 0, preexistingOutput); + assert.include(preexistingOutput, "PublicEvalAttemptWriteError"); + assert.notInclude(preexistingOutput, "ASK_GINA_ACCESS_TOKEN"); + assert.notInclude(preexistingOutput, "OMP_EVAL_API_KEY"); + assert.strictEqual(yield* fs.readFileString(preexistingTranscript), "foreign bytes\n"); + assert.isFalse( + yield* fs.exists( + path.join(outputDirectory, `omp_harness-cand-1-${preexistingRunId}.journal-v1.jsonl`), + ), + ); + + const collisionRunId = "run-collision"; + const collisionTranscript = path.join( + outputDirectory, + `omp_harness-cand-1-${collisionRunId}.transcripts-v1.jsonl`, + ); + const collisionChild = yield* ChildProcess.make( + "bun", + [ + liveCli, + ...requiredFlags("omp", [ + "--provider", + "openai", + "--attempts-output", + collisionTranscript, + ]).map((value) => (value === "run-1" ? collisionRunId : value)), + ], + { + cwd, + env: { PATH: pathValue }, + extendEnv: false, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }, + ); + const [collisionStdout, collisionStderr, collisionExitCode] = yield* Effect.all( + [ + collectBoundedUtf8Output(collisionChild.stdout, 65_536), + collectBoundedUtf8Output(collisionChild.stderr, 65_536), + collisionChild.exitCode, + ], + { concurrency: "unbounded" }, + ); + const collisionOutput = `${collisionStdout.text}\n${collisionStderr.text}`; + assert.notStrictEqual(collisionExitCode, 0, collisionOutput); + assert.include(collisionOutput, "PublicEvalAttemptWriteError"); + assert.notInclude(collisionOutput, "ASK_GINA_ACCESS_TOKEN"); + assert.isFalse(yield* fs.exists(collisionTranscript)); + assert.isFalse( + yield* fs.exists( + path.join(outputDirectory, `omp_harness-cand-1-${collisionRunId}.journal-v1.jsonl`), + ), + ); + }), + ), + ); + + it.effect("reserves a transcript companion only for OMP before credential loading", () => + Effect.scoped( + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const pathValue = yield* Config.string("PATH"); + const cwd = yield* fs.makeTempDirectoryScoped({ prefix: "live-cli-omp-only-" }); + const liveCli = path.join(process.cwd(), "packages/evals/src/bin/live.ts"); + yield* fs.copyFile( + path.join(process.cwd(), "packages/evals/src/fixtures/ask-gina-routing-smoke.yaml"), + path.join(cwd, "suite.yaml"), + ); + + for (const [runner, runId, extra] of [ + ["omp", "run-omp-companion", ["--provider", "openai"]], + ["responses", "run-responses-no-companion", []], + ] as const) { + const child = yield* ChildProcess.make( + "bun", + [ + liveCli, + ...requiredFlags(runner, extra).map((value) => (value === "run-1" ? runId : value)), + ], + { + cwd, + env: { PATH: pathValue }, + extendEnv: false, + stdin: "ignore", + stdout: "pipe", + stderr: "pipe", + }, + ); + const [stdout, stderr, exitCode] = yield* Effect.all( + [ + collectBoundedUtf8Output(child.stdout, 65_536), + collectBoundedUtf8Output(child.stderr, 65_536), + child.exitCode, + ], + { concurrency: "unbounded" }, + ); + const output = `${stdout.text}\n${stderr.text}`; + assert.notStrictEqual(exitCode, 0, output); + assert.include(output, "ASK_GINA_ACCESS_TOKEN"); + } + + const outputDirectory = path.join(cwd, ".plugin-eval-runs"); + const ompTranscriptPath = path.join( + outputDirectory, + "omp_harness-cand-1-run-omp-companion.transcripts-v1.jsonl", + ); + const responsesTranscriptPath = path.join( + outputDirectory, + "responses_api-cand-1-run-responses-no-companion.transcripts-v1.jsonl", + ); + assert.isTrue(yield* fs.exists(ompTranscriptPath)); + assert.include( + yield* fs.readFileString(ompTranscriptPath), + `"schemaVersion":"${OMP_TRANSCRIPT_SCHEMA_VERSION}"`, + ); + assert.isFalse(yield* fs.exists(responsesTranscriptPath)); + }), + ), + ); + it.effect("rejects a missing native profile before authenticated MCP dispatch", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/evals/__tests__/omp-harness.test.ts b/packages/evals/__tests__/omp-harness.test.ts index 18f4009..9bc7903 100644 --- a/packages/evals/__tests__/omp-harness.test.ts +++ b/packages/evals/__tests__/omp-harness.test.ts @@ -15,12 +15,31 @@ import * as BunFileSystem from "@effect/platform-bun/BunFileSystem"; import * as BunPath from "@effect/platform-bun/BunPath"; import { assert, describe, it } from "@effect/vitest"; import { jsonSchema, type GenerateTextResult, type StepResult, type ToolSet } from "ai"; -import { DateTime, Effect, FileSystem, Layer, Path, Redacted } from "effect"; +import { + Cause, + DateTime, + Deferred, + Duration, + Effect, + Exit, + Fiber, + FileSystem, + Layer, + Path, + Redacted, +} from "effect"; +import { TestClock } from "effect/testing"; import { beforeEach, vi } from "vitest"; import { gradePluginEvalObservation } from "../src/grading"; import { loadPluginEvalSuite } from "../src/load-suite"; -import { runOmpHarnessPluginEvalTrial } from "../src/omp-harness"; +import { + PluginEvalOmpHarnessProcessError, + PluginEvalOmpHarnessTranscriptError, + runOmpHarnessPluginEvalTrial, + type OmpHarnessTrialOptions, +} from "../src/omp-harness"; +import type { OmpTrialTranscript } from "../src/omp-transcript"; const fixture = vi.hoisted(() => ({ failedRead: false, @@ -28,6 +47,14 @@ const fixture = vi.hoisted(() => ({ priceCallCount: 1, extraPriceMirror: false, mismatchedMirrorArguments: false, + generateCalls: 0, + streamCalls: 0, + destroyCalls: 0, + destroyFails: false, + streamMode: "success" as "success" | "terminal-error" | "hang", + streamSecretText: false, + lastPrompt: undefined as string | undefined, + onStreamWait: undefined as (() => void) | undefined, })); const PRICE_TOOL = "spot.getSimplePrice"; const PRICE_ARGUMENTS = { ids: "ethereum", vs_currencies: "usd" }; @@ -172,12 +199,18 @@ vi.mock("@ai-sdk/harness/agent", () => ({ return Promise.resolve(sandbox.createSession(options)).then( (session) => ({ - destroy: () => Promise.resolve(session.destroy()), + destroy: () => { + fixture.destroyCalls += 1; + return fixture.destroyFails + ? Promise.reject(new Error("Fixture session cleanup failed")) + : Promise.resolve(session.destroy()); + }, }) satisfies Pick, ); } generate(): Promise { + fixture.generateCalls += 1; const readCall = { type: "tool-call", toolCallId: "native-skill-read", @@ -272,12 +305,87 @@ vi.mock("@ai-sdk/harness/agent", () => ({ }), ); } + + stream(options: { readonly prompt: string; readonly abortSignal?: AbortSignal }) { + fixture.streamCalls += 1; + fixture.lastPrompt = options.prompt; + return this.generate().then((evidence) => { + const parts: Array> = [ + { type: "text-delta", id: "answer", text: "Checking " }, + { type: "reasoning-delta", id: "reasoning", text: "private chain" }, + { type: "text-delta", id: "answer", text: "Ethereum." }, + { + type: "tool-call", + toolCallId: "stream-price-call", + toolName: PRICE_TOOL, + input: PRICE_ARGUMENTS, + providerExecuted: false, + }, + { type: "raw", rawValue: { private: "provider frame" } }, + { + type: "tool-result", + toolCallId: "stream-price-call", + toolName: PRICE_TOOL, + input: PRICE_ARGUMENTS, + output: PRICE_RESULT, + providerExecuted: false, + }, + { + type: "text-delta", + id: "answer", + text: fixture.streamSecretText + ? "Bearer synthetic-mcp-authorization" + : "Ethereum is $3,200 USD.", + }, + ]; + let index = 0; + let pending: PromiseWithResolvers<{ done: true; value: undefined }> | undefined; + const stream = { + [Symbol.asyncIterator]: () => ({ + next: () => { + const hangAfter = 3; + if (fixture.streamMode !== "hang" || index < hangAfter) { + if (index < parts.length) { + const value = parts[index]; + index += 1; + return Promise.resolve({ done: false as const, value }); + } + return Promise.resolve({ done: true as const, value: undefined }); + } + fixture.onStreamWait?.(); + if (pending === undefined) { + pending = Promise.withResolvers<{ done: true; value: undefined }>(); + options.abortSignal?.addEventListener( + "abort", + () => pending?.resolve({ done: true, value: undefined }), + { once: true }, + ); + } + return pending.promise; + }, + return: () => + pending?.promise ?? Promise.resolve({ done: true as const, value: undefined }), + }), + }; + return { + stream, + get text() { + return fixture.streamMode === "terminal-error" + ? Promise.reject(new Error("Fixture terminal accessor failed")) + : Promise.resolve(evidence.text); + }, + finishReason: Promise.resolve(evidence.finishReason), + usage: Promise.resolve(evidence.usage), + steps: Promise.resolve(evidence.steps), + }; + }); + } }, })); const TestPlatformLayer = Layer.merge(BunFileSystem.layer, BunPath.layer); -const runPriceTrial = Effect.gen(function* () { +const preparePriceTrial = Effect.gen(function* () { const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const runtimeDirectory = yield* fs.makeTempDirectoryScoped({ prefix: "omp-read-evidence-" }); @@ -296,17 +404,23 @@ const runPriceTrial = Effect.gen(function* () { ); const evalCase = suite.cases.find((candidate) => candidate.id === "spot-simple-price"); if (evalCase === undefined) return yield* Effect.die("Missing spot-simple-price suite case"); - const observation = yield* runOmpHarnessPluginEvalTrial(evalCase, { + const options = { runId: "omp-read-evidence", repetition: 1, availableTools: GINA_CONNECTED_TOOL_NAMES, runtimeDirectory, - auth: { mode: "native", provider: "fixture", agentDirectory }, + auth: { mode: "native" as const, provider: "fixture", agentDirectory }, model: "fixture-model", reasoning: "off", mcpAuthorization: Redacted.make("synthetic-mcp-authorization"), timeoutMs: 30_000, - }); + } satisfies OmpHarnessTrialOptions; + return { evalCase, options }; +}); + +const runPriceTrial = Effect.gen(function* () { + const { evalCase, options } = yield* preparePriceTrial; + const observation = yield* runOmpHarnessPluginEvalTrial(evalCase, options); const score = yield* gradePluginEvalObservation(evalCase, observation); return { observation, score }; }); @@ -318,6 +432,14 @@ describe("OMP harness native read evidence", () => { fixture.priceCallCount = 1; fixture.extraPriceMirror = false; fixture.mismatchedMirrorArguments = false; + fixture.generateCalls = 0; + fixture.streamCalls = 0; + fixture.destroyCalls = 0; + fixture.destroyFails = false; + fixture.streamMode = "success"; + fixture.streamSecretText = false; + fixture.lastPrompt = undefined; + fixture.onStreamWait = undefined; }); it.layer(TestPlatformLayer)((it) => { @@ -336,9 +458,283 @@ describe("OMP harness native read evidence", () => { observation.tool_calls.map(({ name }) => name), [PRICE_TOOL], ); + assert.strictEqual(fixture.generateCalls, 1); + assert.strictEqual(fixture.streamCalls, 0); }), ); + it.effect("captures ordered stream chat without changing terminal grading evidence", () => + Effect.gen(function* () { + const { evalCase, options } = yield* preparePriceTrial; + const transcripts: OmpTrialTranscript[] = []; + const observation = yield* runOmpHarnessPluginEvalTrial(evalCase, { + ...options, + onTranscript: (transcript) => { + transcripts.push(transcript); + return Promise.resolve(); + }, + }); + const score = yield* gradePluginEvalObservation(evalCase, observation); + const expectedPrompt = evalCase.turns + .filter((turn) => turn.role === "user") + .map((turn) => turn.content) + .join("\n\n"); + + assert.strictEqual(fixture.streamCalls, 1); + assert.strictEqual(fixture.lastPrompt, expectedPrompt); + assert.strictEqual(observation.final_answer, "Ethereum is $3,200 USD."); + assert.strictEqual(score.routing.score, 1); + assert.strictEqual(score.arguments.score, 1); + assert.isTrue(score.overall_pass); + assert.lengthOf(transcripts, 1); + const [transcript] = transcripts; + if (transcript === undefined) return; + assert.strictEqual(transcript.status, "completed"); + assert.isFalse(transcript.truncated); + assert.deepStrictEqual(transcript.messages, [ + { role: "user", type: "text", text: expectedPrompt }, + { role: "assistant", type: "text", text: "Checking Ethereum." }, + { + role: "assistant", + type: "tool-call", + toolCallId: "stream-price-call", + toolName: PRICE_TOOL, + input: PRICE_ARGUMENTS, + }, + { + role: "tool", + type: "tool-result", + toolCallId: "stream-price-call", + toolName: PRICE_TOOL, + output: PRICE_RESULT, + isError: false, + }, + { + role: "assistant", + type: "text", + text: "Ethereum is $3,200 USD.", + }, + ]); + }), + ); + + it.effect("retains partial stream when terminal accessors reject", () => + Effect.gen(function* () { + fixture.streamMode = "terminal-error"; + const callbackSecret = "terminal-callback-secret-must-not-win"; + const { evalCase, options } = yield* preparePriceTrial; + const transcripts: OmpTrialTranscript[] = []; + const result = yield* Effect.result( + runOmpHarnessPluginEvalTrial(evalCase, { + ...options, + onTranscript: (transcript) => { + transcripts.push(transcript); + return Promise.reject(new Error(callbackSecret)); + }, + }), + ); + + assert.strictEqual(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.instanceOf(result.failure, PluginEvalOmpHarnessProcessError); + assert.notInclude(String(result.failure), callbackSecret); + } + assert.lengthOf(transcripts, 1); + const [transcript] = transcripts; + if (transcript === undefined) return; + assert.strictEqual(transcript.status, "failed"); + assert.deepInclude(transcript.messages, { + role: "assistant", + type: "text", + text: "Ethereum is $3,200 USD.", + }); + assert.strictEqual(fixture.destroyCalls, 1); + }), + ); + + it.effect("retains partial stream and reports timeout after cleanup", () => + Effect.gen(function* () { + fixture.streamMode = "hang"; + const ready = yield* Deferred.make(); + fixture.onStreamWait = () => { + void Deferred.doneUnsafe(ready, Effect.void); + }; + const { evalCase, options } = yield* preparePriceTrial; + const transcripts: OmpTrialTranscript[] = []; + const fiber = yield* Effect.forkChild( + runOmpHarnessPluginEvalTrial(evalCase, { + ...options, + timeoutMs: 1_000, + onTranscript: (transcript, signal) => { + assert.isFalse(signal.aborted); + transcripts.push(transcript); + return Promise.resolve(); + }, + }), + ); + yield* Deferred.await(ready); + yield* TestClock.adjust(Duration.millis(1_000)); + const exit = yield* Fiber.await(fiber); + + assert.isTrue(Exit.isFailure(exit)); + assert.lengthOf(transcripts, 1); + const [transcript] = transcripts; + if (transcript === undefined) return; + assert.strictEqual(transcript.status, "timeout"); + assert.deepInclude(transcript.messages, { + role: "assistant", + type: "text", + text: "Checking Ethereum.", + }); + assert.strictEqual(fixture.destroyCalls, 1); + }), + ); + + it.effect("retains partial stream without swallowing parent interruption", () => + Effect.gen(function* () { + fixture.streamMode = "hang"; + const ready = yield* Deferred.make(); + fixture.onStreamWait = () => { + void Deferred.doneUnsafe(ready, Effect.void); + }; + const callbackSecret = "interrupt-callback-secret-must-not-win"; + const { evalCase, options } = yield* preparePriceTrial; + const transcripts: OmpTrialTranscript[] = []; + const fiber = yield* Effect.forkChild( + runOmpHarnessPluginEvalTrial(evalCase, { + ...options, + onTranscript: (transcript, signal) => { + assert.isFalse(signal.aborted); + transcripts.push(transcript); + return Promise.reject(new Error(callbackSecret)); + }, + }), + ); + yield* Deferred.await(ready); + yield* Fiber.interrupt(fiber); + const exit = yield* Fiber.await(fiber); + + assert.isTrue(Exit.isFailure(exit)); + if (Exit.isFailure(exit)) { + assert.isTrue(Cause.hasInterrupts(exit.cause)); + assert.notInclude(String(exit.cause), callbackSecret); + } + assert.lengthOf(transcripts, 1); + const [transcript] = transcripts; + if (transcript === undefined) return; + assert.strictEqual(transcript.status, "interruption"); + assert.deepInclude(transcript.messages, { + role: "assistant", + type: "text", + text: "Checking Ethereum.", + }); + }), + ); + + it.effect("fails a successful trial with a value-free callback rejection after cleanup", () => + Effect.gen(function* () { + fixture.streamSecretText = true; + const callbackSecret = "callback-secret-must-not-leak"; + const { evalCase, options } = yield* preparePriceTrial; + const transcripts: OmpTrialTranscript[] = []; + let cleanupCountAtCallback = 0; + const result = yield* Effect.result( + runOmpHarnessPluginEvalTrial(evalCase, { + ...options, + onTranscript: (transcript) => { + cleanupCountAtCallback = fixture.destroyCalls; + transcripts.push(transcript); + return Promise.reject(new Error(callbackSecret)); + }, + }), + ); + + assert.strictEqual(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.instanceOf(result.failure, PluginEvalOmpHarnessTranscriptError); + if (result.failure instanceof PluginEvalOmpHarnessTranscriptError) { + assert.strictEqual(result.failure.reason, "write-failed"); + } + const failureText = String(result.failure); + assert.notInclude(failureText, callbackSecret); + assert.notInclude(failureText, "synthetic-mcp-authorization"); + } + assert.strictEqual(cleanupCountAtCallback, 1); + assert.lengthOf(transcripts, 1); + const [transcript] = transcripts; + if (transcript === undefined) return; + assert.strictEqual(transcript.status, "completed"); + const finalMessage = transcript.messages.at(-1); + assert.strictEqual(finalMessage?.type, "text"); + if (finalMessage?.type === "text") { + assert.notInclude(finalMessage.text, "synthetic-mcp-authorization"); + assert.include(finalMessage.text, "[redacted]"); + } + }), + ); + + it.effect("stops a successful trial when transcript persistence never settles", () => + Effect.gen(function* () { + const ready = yield* Deferred.make(); + const pending = Promise.withResolvers(); + let callbackCancelled = false; + const { evalCase, options } = yield* preparePriceTrial; + const fiber = yield* Effect.forkChild( + runOmpHarnessPluginEvalTrial(evalCase, { + ...options, + onTranscript: (_transcript, signal) => { + signal.addEventListener( + "abort", + () => { + callbackCancelled = true; + pending.resolve(); + }, + { once: true }, + ); + void Deferred.doneUnsafe(ready, Effect.void); + return pending.promise; + }, + }), + ); + yield* Deferred.await(ready); + yield* TestClock.adjust(Duration.millis(5_000)); + const result = yield* Fiber.join(fiber).pipe(Effect.result); + + assert.isTrue(callbackCancelled); + assert.strictEqual(result._tag, "Failure"); + if (result._tag === "Failure") { + assert.instanceOf(result.failure, PluginEvalOmpHarnessTranscriptError); + if (result.failure instanceof PluginEvalOmpHarnessTranscriptError) { + assert.strictEqual(result.failure.reason, "write-timeout"); + } + } + }), + ); + + it.effect("reports failed transcript status when settled cleanup fails", () => + Effect.gen(function* () { + fixture.destroyFails = true; + const { evalCase, options } = yield* preparePriceTrial; + const transcripts: OmpTrialTranscript[] = []; + let cleanupCountAtCallback = 0; + const result = yield* Effect.result( + runOmpHarnessPluginEvalTrial(evalCase, { + ...options, + onTranscript: (transcript) => { + cleanupCountAtCallback = fixture.destroyCalls; + transcripts.push(transcript); + return Promise.resolve(); + }, + }), + ); + + assert.strictEqual(result._tag, "Failure"); + assert.strictEqual(cleanupCountAtCallback, 1); + assert.lengthOf(transcripts, 1); + assert.strictEqual(transcripts[0]?.status, "failed"); + }), + ); + it.effect( "keeps a failed native skill read incomplete without contaminating Gina routing", () => diff --git a/packages/evals/__tests__/omp-transcript.test.ts b/packages/evals/__tests__/omp-transcript.test.ts new file mode 100644 index 0000000..0b02c01 --- /dev/null +++ b/packages/evals/__tests__/omp-transcript.test.ts @@ -0,0 +1,373 @@ +import { assert, describe, it } from "@effect/vitest"; +import { Effect, Schema } from "effect"; + +import { + createOmpTranscriptCollector, + OMP_TRANSCRIPT_MAX_BYTES, + OMP_TRANSCRIPT_MAX_FIELD_BYTES, + OMP_TRANSCRIPT_MAX_MESSAGES, + OMP_TRANSCRIPT_REDACTION, + OmpTranscriptToolCallMessageSchema, + OmpTrialTranscriptSchema, +} from "../src/omp-transcript"; + +describe("OMP private transcript collection", () => { + it("coalesces text deltas while preserving chronological tool events", () => { + const collector = createOmpTranscriptCollector([]); + + collector.user("Find "); + collector.user("the market"); + collector.assistant("I will "); + collector.assistant("look it up."); + collector.toolCall("call-1", "searchMarkets", { query: "weather" }); + collector.toolResult("call-1", "searchMarkets", { markets: ["rain"] }, false); + collector.assistant("One market matched."); + + assert.deepStrictEqual(collector.finish().messages, [ + { role: "user", type: "text", text: "Find the market" }, + { role: "assistant", type: "text", text: "I will look it up." }, + { + role: "assistant", + type: "tool-call", + toolCallId: "call-1", + toolName: "searchMarkets", + input: { query: "weather" }, + }, + { + role: "tool", + type: "tool-result", + toolCallId: "call-1", + toolName: "searchMarkets", + output: { markets: ["rain"] }, + isError: false, + }, + { role: "assistant", type: "text", text: "One market matched." }, + ]); + }); + + it("redacts secrets split across deltas and incomplete terminal prefixes", () => { + const secret = ["sk", "-live-abcdef0123456789abcdef0123456789"].join(""); + const collector = createOmpTranscriptCollector([secret]); + collector.assistant(`full=${secret.slice(0, 13)}`); + collector.assistant(`${secret.slice(13)} done`); + + const complete = collector.finish().messages[0]; + assert.strictEqual(complete?.type, "text"); + if (complete?.type !== "text") return; + assert.strictEqual(complete.text.includes(secret), false); + assert.strictEqual(complete.text.includes(OMP_TRANSCRIPT_REDACTION), true); + + const interrupted = createOmpTranscriptCollector([secret]); + const terminalPrefix = secret.slice(0, 17); + interrupted.assistant(`partial=${terminalPrefix}`); + const partial = interrupted.finish().messages[0]; + assert.strictEqual(partial?.type, "text"); + if (partial?.type !== "text") return; + assert.strictEqual(partial.text.includes(terminalPrefix), false); + assert.strictEqual(partial.text.endsWith(OMP_TRANSCRIPT_REDACTION), true); + + const shortSecret = "a"; + const short = createOmpTranscriptCollector([shortSecret]); + short.user(shortSecret); + const shortMessage = short.finish().messages[0]; + assert.strictEqual(shortMessage?.type, "text"); + if (shortMessage?.type !== "text") return; + assert.strictEqual(shortMessage.text.includes(shortSecret), false); + }); + + it("unions assignment and header spans with complete or truncated private-key blocks", () => { + const genericBody = "GENERIC_KEY_BODY_SHOULD_NOT_SURVIVE"; + const openSshBody = "OPENSSH_KEY_BODY_SHOULD_NOT_SURVIVE"; + const truncatedBody = "TRUNCATED_KEY_BODY_SHOULD_NOT_SURVIVE"; + const openSshHeader = ["-----BEGIN OPENSSH", " PRIVATE KEY-----"].join(""); + const collector = createOmpTranscriptCollector([]); + + collector.assistant("safe text before\nPRIVATE_KEY=-----BEGIN PRIVATE"); + collector.assistant(` KEY-----\n${genericBody}\n-----END PRIVATE KEY-----\nsafe text after`); + collector.toolResult( + "call-key", + "inspect", + { + complete: + `safe JSON before\nAuthorization: ${openSshHeader}\n` + + `${openSshBody}\n-----END OPENSSH PRIVATE KEY-----\nsafe JSON after`, + truncated: `safe truncated prefix\nOPENSSH_PRIVATE_KEY=${openSshHeader}\n` + truncatedBody, + }, + false, + ); + + const result = collector.finish(); + assert.deepStrictEqual(result.messages, [ + { + role: "assistant", + type: "text", + text: `safe text before\n${OMP_TRANSCRIPT_REDACTION}\nsafe text after`, + }, + { + role: "tool", + type: "tool-result", + toolCallId: "call-key", + toolName: "inspect", + output: { + complete: `safe JSON before\n${OMP_TRANSCRIPT_REDACTION}\nsafe JSON after`, + truncated: `safe truncated prefix\n${OMP_TRANSCRIPT_REDACTION}`, + }, + isError: false, + }, + ]); + const encoded = JSON.stringify(result.messages); + assert.strictEqual(encoded.includes(genericBody), false); + assert.strictEqual(encoded.includes(openSshBody), false); + assert.strictEqual(encoded.includes(truncatedBody), false); + }); + + it("redacts a split secret at the text cap without breaking UTF-8", () => { + const secret = "secret-value-abcdef0123456789abcdef0123456789"; + const split = Math.floor(secret.length / 2); + const collector = createOmpTranscriptCollector([secret]); + const prefix = "safe ".repeat(Math.ceil((OMP_TRANSCRIPT_MAX_FIELD_BYTES - split) / 5)); + collector.assistant(prefix + secret.slice(0, split)); + collector.assistant(secret.slice(split) + " trailing words ".repeat(64)); + + const result = collector.finish(); + const message = result.messages[0]; + assert.strictEqual(message?.type, "text"); + if (message?.type !== "text") return; + assert.strictEqual(result.truncated, true); + assert.strictEqual(message.text.includes(secret), false); + assert.strictEqual(message.text.includes(secret.slice(0, split)), false); + assert.strictEqual(message.text.includes(OMP_TRANSCRIPT_REDACTION), true); + assert.strictEqual(message.text.includes("�"), false); + }); + + it("keeps JSON arguments and results as cloned redacted snapshots", () => { + const token = ["gh", "p_abcdefghijklmnopqrstuvwxyz0123456789"].join(""); + const input = { + path: "/private/worktree/input.json", + nested: { api_key: "caller-owned-secret", keep: [1, "two"] }, + }; + const output = { + note: `provider returned ${token}`, + authorization: ["Bearer", "caller-owned-secret"].join(" "), + rows: [{ value: 3 }], + }; + const collector = createOmpTranscriptCollector([]); + collector.toolCall("call-2", "query", input); + collector.toolResult("call-2", "query", output, true); + + input.nested.api_key = "mutated"; + input.nested.keep.push(9); + output.note = "mutated"; + output.rows[0].value = 99; + + const result = collector.finish(); + assert.deepStrictEqual(result.messages, [ + { + role: "assistant", + type: "tool-call", + toolCallId: "call-2", + toolName: "query", + input: { + path: "/private/worktree/input.json", + nested: { api_key: OMP_TRANSCRIPT_REDACTION, keep: [1, "two"] }, + }, + }, + { + role: "tool", + type: "tool-result", + toolCallId: "call-2", + toolName: "query", + output: { + note: `provider returned ${OMP_TRANSCRIPT_REDACTION}`, + authorization: OMP_TRANSCRIPT_REDACTION, + rows: [{ value: 3 }], + }, + isError: true, + }, + ]); + const encoded = JSON.stringify(result.messages); + assert.strictEqual(encoded.includes(token), false); + assert.strictEqual(encoded.includes("caller-owned-secret"), false); + assert.strictEqual(encoded.includes("/private/worktree/input.json"), true); + }); + + it("replaces unsupported and oversized JSON without retaining SDK objects", () => { + const circular: Record = { value: 1 }; + circular.self = circular; + const collector = createOmpTranscriptCollector([]); + collector.toolCall("call-3", "unsafe", { + circular, + bigint: 12n, + callback: () => "raw", + instance: /sdk-object/u, + huge: "x".repeat(OMP_TRANSCRIPT_MAX_FIELD_BYTES + 1), + }); + + const result = collector.finish(); + const call = result.messages[0]; + assert.strictEqual(call?.type, "tool-call"); + if (call?.type !== "tool-call") return; + assert.strictEqual(result.truncated, true); + const encoded = JSON.stringify(call.input); + assert.strictEqual(encoded.includes("[omitted:"), true); + assert.strictEqual(encoded.includes("raw"), false); + assert.strictEqual(encoded.includes("x".repeat(256)), false); + }); + + it("stops traversing JSON as soon as its cumulative text budget is exhausted", () => { + const chunk = "x".repeat(Math.floor(OMP_TRANSCRIPT_MAX_FIELD_BYTES * 0.6)); + let readAfterBudget = false; + const input: Record = { first: chunk, second: chunk }; + Object.defineProperty(input, "afterBudget", { + enumerable: true, + get: () => { + readAfterBudget = true; + return "must not be read"; + }, + }); + const collector = createOmpTranscriptCollector([]); + collector.toolCall("call-budget", "query", input); + + const result = collector.finish(); + assert.strictEqual(readAfterBudget, false); + assert.strictEqual(result.truncated, true); + const call = result.messages[0]; + assert.strictEqual(call?.type, "tool-call"); + if (call?.type !== "tool-call") return; + assert.strictEqual(typeof call.input, "string"); + assert.strictEqual(String(call.input).length < 64, true); + }); + + it("seals immutable snapshots against late deltas and caller mutation", () => { + const input = { nested: { value: 1 } }; + const collector = createOmpTranscriptCollector([]); + collector.toolCall("call-4", "read", input); + const first = collector.finish(); + + input.nested.value = 2; + collector.assistant("late text"); + collector.toolResult("call-4", "read", { value: 2 }, false); + const second = collector.finish(); + + assert.strictEqual(first, second); + assert.deepStrictEqual(second.messages, [ + { + role: "assistant", + type: "tool-call", + toolCallId: "call-4", + toolName: "read", + input: { nested: { value: 1 } }, + }, + ]); + assert.strictEqual(Object.isFrozen(second.messages), true); + }); + + it("bounds entry count and encoded UTF-8 output", () => { + const entries = createOmpTranscriptCollector([]); + for (let index = 0; index < OMP_TRANSCRIPT_MAX_MESSAGES + 20; index += 1) { + if (index % 2 === 0) entries.user(`u${index}`); + else entries.assistant(`a${index}`); + } + let readAfterCapacity = false; + const ignoredPayload: Record = {}; + Object.defineProperty(ignoredPayload, "value", { + enumerable: true, + get: () => { + readAfterCapacity = true; + return "must not be read"; + }, + }); + entries.toolCall("ignored", "ignored", ignoredPayload); + assert.strictEqual(readAfterCapacity, false); + const cappedEntries = entries.finish(); + assert.strictEqual(cappedEntries.messages.length, OMP_TRANSCRIPT_MAX_MESSAGES); + assert.strictEqual(cappedEntries.truncated, true); + + const bytes = createOmpTranscriptCollector([]); + for (let index = 0; index < 40; index += 1) { + bytes.toolResult( + `call-${index}`, + "bulk", + { + value: `${index}:` + "🦉 ".repeat(8_000), + }, + false, + ); + } + let readAfterByteCapacity = false; + const ignoredBytePayload: Record = {}; + Object.defineProperty(ignoredBytePayload, "value", { + enumerable: true, + get: () => { + readAfterByteCapacity = true; + return "must not be read"; + }, + }); + bytes.toolResult("ignored", "bulk", ignoredBytePayload, false); + assert.strictEqual(readAfterByteCapacity, false); + const cappedBytes = bytes.finish(); + assert.strictEqual(cappedBytes.truncated, true); + assert.strictEqual( + new TextEncoder().encode(JSON.stringify(cappedBytes.messages)).byteLength <= + OMP_TRANSCRIPT_MAX_BYTES, + true, + ); + assert.strictEqual(JSON.stringify(cappedBytes.messages).includes("�"), false); + }); + + it("bounds labels after redaction expansion and validates the stored message", () => { + const collector = createOmpTranscriptCollector(["x"]); + collector.toolCall("call-label", "x".repeat(256), { ok: true }); + + const result = collector.finish(); + const message = result.messages[0]; + assert.strictEqual(result.truncated, true); + assert.deepStrictEqual(message, { + role: "assistant", + type: "tool-call", + toolCallId: "call-label", + toolName: "[omitted:too-large]", + input: { ok: true }, + }); + const decoded = Effect.runSync( + Schema.decodeUnknownEffect(OmpTranscriptToolCallMessageSchema, { + onExcessProperty: "error", + })(message), + ); + assert.deepStrictEqual(decoded, message); + }); + + it("decodes a collected transcript through the shared Effect schema", () => { + const collector = createOmpTranscriptCollector([]); + collector.user("hello"); + const collected = collector.finish(); + const transcript = { + runId: "run-1", + caseId: "case-1", + repetition: 1, + model: "provider/model", + startedAt: "2026-09-11T12:00:00.000Z", + status: "completed", + messages: collected.messages, + truncated: collected.truncated, + }; + + const decoded = Effect.runSync( + Schema.decodeUnknownEffect(OmpTrialTranscriptSchema, { onExcessProperty: "error" })( + transcript, + ), + ); + assert.deepStrictEqual(decoded, transcript); + }); + + it("drops a single oversized text token instead of preserving a prefix", () => { + const collector = createOmpTranscriptCollector([]); + collector.assistant("x".repeat(OMP_TRANSCRIPT_MAX_FIELD_BYTES * 3)); + + const result = collector.finish(); + assert.deepStrictEqual(result.messages, [ + { role: "assistant", type: "text", text: "[omitted:too-large]" }, + ]); + assert.strictEqual(result.truncated, true); + }); +}); diff --git a/packages/evals/__tests__/trial-journal.test.ts b/packages/evals/__tests__/trial-journal.test.ts index eab20f7..e6197ac 100644 --- a/packages/evals/__tests__/trial-journal.test.ts +++ b/packages/evals/__tests__/trial-journal.test.ts @@ -7,6 +7,7 @@ import { Cause, Data, Deferred, + Duration, Effect, Exit, Fiber, @@ -15,6 +16,7 @@ import { Path, PlatformError, Schema, + Scope, } from "effect"; import type { PluginEvalObservation } from "../src/contracts"; @@ -25,16 +27,21 @@ import { type OpenRouterBudgetEvidence, } from "../src/openrouter-budget"; import type { OpenRouterGenerationEvidence } from "../src/provider-evidence"; +import type { OmpTrialTranscript } from "../src/omp-transcript"; +import { OMP_TRANSCRIPT_SCHEMA_VERSION } from "../src/omp-transcript"; import { ALPHA_GINA_READ_SERVER_URL } from "../src/server-url"; import { createLiveEvalJournal, + createOmpTranscriptWriter, liveEvalJournalDispatchId, LiveEvalJournalError, LIVE_EVAL_JOURNAL_MAX_CASES, LIVE_EVAL_JOURNAL_SCHEMA_VERSION, + OMP_TRANSCRIPT_MAX_RECORD_BYTES, withJournaledTrial, type LiveEvalJournal, type LiveEvalJournalOptions, + type OmpTranscriptWriterOptions, } from "../src/trial-journal"; const TestPlatformLayer = Layer.merge(BunFileSystem.layer, BunPath.layer); @@ -96,6 +103,46 @@ const journalOptions = ( const selectedCaseIds = ["simple-spot-price"] as const; +const transcriptWriterOptions = ( + outputDir: string, + fileName: string, + overrides: Partial = {}, +): OmpTranscriptWriterOptions => ({ + outputDir, + fileName, + runId: "am-spot-routing-20260910", + candidate: "sol-medium-diagnostic", + model: "openai/gpt-5.6-sol", + serverUrl: PRODUCTION_MCP_URL, + suiteId: "synthetic-model-smoke-v1", + suiteVersion: 1, + fixtureVersion: 1, + catalogSha, + reasoning: "medium", + accountClass: "local", + caseIds: selectedCaseIds, + repetitions: 3, + ...overrides, +}); + +const ompTranscript = ( + repetition: number, + overrides: Partial = {}, +): OmpTrialTranscript => ({ + runId: "am-spot-routing-20260910", + caseId: "simple-spot-price", + repetition, + model: "openai/gpt-5.6-sol", + startedAt: `2026-09-10T00:00:0${repetition}.000Z`, + status: "completed", + messages: [ + { role: "user", type: "text", text: `price check ${repetition}` }, + { role: "assistant", type: "text", text: `answer ${repetition}` }, + ], + truncated: false, + ...overrides, +}); + const v1Report = (overrides: Record = {}) => ({ schemaVersion: "v1", runId: "am-spot-routing-20260910", @@ -925,3 +972,260 @@ describe("live eval trial journal", () => { ); }); }); + +describe("OMP transcript writer", () => { + it.layer(TestPlatformLayer, { excludeTestServices: true })((it) => { + it.effect("cancels a stalled callback write and closes its scope before sync resumes", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "omp-transcript-cancel-" }); + const outputPath = path.join(directory, "run.jsonl"); + const enteredSync = yield* Deferred.make(); + const resumeSync = yield* Deferred.make(); + const writerScope = yield* Scope.make(); + let syncCount = 0; + const gated: FileSystem.FileSystem = { + ...fs, + open: (name, options) => + fs.open(name, options).pipe( + Effect.map((file) => ({ + ...file, + stat: file.stat, + writeAll: (bytes: Uint8Array) => file.writeAll(bytes), + sync: Effect.suspend(() => { + syncCount += 1; + return syncCount === 2 + ? Deferred.succeed(enteredSync, undefined).pipe( + Effect.andThen(Deferred.await(resumeSync)), + Effect.andThen(file.sync), + ) + : file.sync; + }), + })), + ), + }; + yield* Effect.gen(function* () { + const writer = yield* createOmpTranscriptWriter( + transcriptWriterOptions(directory, "run.jsonl"), + ).pipe(Effect.provideService(FileSystem.FileSystem, gated), Scope.provide(writerScope)); + const runWriter = Effect.runPromiseWith( + yield* Effect.context(), + ); + const callback = yield* Effect.forkChild( + Effect.tryPromise({ + try: (signal) => runWriter(writer.writeTrial(ompTranscript(1)), { signal }), + catch: () => new LiveEvalJournalError({ reason: "write-failed" }), + }), + ); + yield* Deferred.await(enteredSync); + yield* Fiber.interrupt(callback); + const cancelled = yield* Fiber.await(callback); + assert.isTrue(Exit.isFailure(cancelled)); + if (Exit.isFailure(cancelled)) assert.isTrue(Cause.hasInterrupts(cancelled.cause)); + + const shutdown = yield* Effect.gen(function* () { + const retry = yield* Effect.result(writer.writeTrial(ompTranscript(1))); + assert.strictEqual(reasonOf(retry), "write-failed"); + const binding = yield* Effect.result( + writer.bindReport( + `${encodeUnknownJson(v1Report({ target: "omp_harness" }))}\n`, + selectedCaseIds, + ), + ); + assert.strictEqual(binding._tag, "Failure"); + yield* Scope.close(writerScope, Exit.void); + }).pipe(Effect.timeout(Duration.seconds(2)), Effect.result); + assert.strictEqual(shutdown._tag, "Success"); + assert.isFalse( + parseRecords(yield* fs.readFileString(outputPath)).some( + (record) => record.kind === "report-bound", + ), + ); + }).pipe( + Effect.ensuring(Deferred.succeed(resumeSync, undefined)), + Effect.ensuring(Scope.close(writerScope, Exit.void)), + ); + }), + ); + }); + + it.layer(TestPlatformLayer)((it) => { + it.effect("writes ordered private trials and binds the exact report bytes", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "omp-transcript-write-" }); + const fileName = "run.transcripts-v1.jsonl"; + const outputPath = path.join(directory, fileName); + const writer = yield* createOmpTranscriptWriter( + transcriptWriterOptions(directory, fileName), + ); + assert.strictEqual((yield* fs.stat(outputPath)).mode & 0o777, 0o600); + + for (const repetition of [1, 2, 3]) { + yield* writer.writeTrial(ompTranscript(repetition)); + } + const reportContent = `${encodeUnknownJson(v1Report({ target: "omp_harness" }))}\n`; + yield* writer.bindReport(reportContent, selectedCaseIds); + + const records = parseRecords(yield* fs.readFileString(outputPath)); + assert.deepStrictEqual( + records.map((record) => record.kind), + ["run", "trial", "trial", "trial", "report-bound"], + ); + assert.deepStrictEqual(records[0], { + kind: "run", + schemaVersion: OMP_TRANSCRIPT_SCHEMA_VERSION, + runId: "am-spot-routing-20260910", + candidate: "sol-medium-diagnostic", + target: "omp_harness", + model: "openai/gpt-5.6-sol", + serverUrl: PRODUCTION_MCP_URL, + suiteId: "synthetic-model-smoke-v1", + suiteVersion: 1, + fixtureVersion: 1, + catalogSha, + reasoning: "medium", + accountClass: "local", + caseIds: selectedCaseIds, + repetitions: 3, + }); + for (const [index, repetition] of [1, 2, 3].entries()) { + assert.strictEqual( + records[index + 1]?.dispatchId, + liveEvalJournalDispatchId("am-spot-routing-20260910", "simple-spot-price", repetition), + ); + assert.deepStrictEqual(records[index + 1]?.transcript, ompTranscript(repetition)); + } + assert.deepStrictEqual(records.at(-1), { + kind: "report-bound", + sourceReportSha256: createHash("sha256").update(reportContent, "utf8").digest("hex"), + }); + }), + ); + + it.effect( + "rejects escaped, preexisting, duplicate, unplanned, model-mismatched and incomplete records", + () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "omp-transcript-plan-" }); + const escapedPath = path.join(path.dirname(directory), "escaped-transcript.jsonl"); + const escaped = yield* Effect.result( + createOmpTranscriptWriter( + transcriptWriterOptions(directory, "../escaped-transcript.jsonl"), + ), + ); + assert.strictEqual(reasonOf(escaped), "invalid-path"); + assert.isFalse(yield* fs.exists(escapedPath)); + + const preexistingPath = path.join(directory, "preexisting.jsonl"); + yield* fs.writeFileString(preexistingPath, "foreign bytes\n", { + flag: "wx", + mode: 0o600, + }); + const preexisting = yield* Effect.result( + createOmpTranscriptWriter(transcriptWriterOptions(directory, "preexisting.jsonl")), + ); + assert.strictEqual(reasonOf(preexisting), "output-exists"); + assert.strictEqual(yield* fs.readFileString(preexistingPath), "foreign bytes\n"); + + const outputPath = path.join(directory, "planned.jsonl"); + const writer = yield* createOmpTranscriptWriter( + transcriptWriterOptions(directory, path.basename(outputPath)), + ); + yield* writer.writeTrial(ompTranscript(1)); + const durablePrefix = yield* fs.readFileString(outputPath); + assert.strictEqual( + reasonOf(yield* Effect.result(writer.writeTrial(ompTranscript(1)))), + "already-started", + ); + assert.strictEqual( + reasonOf( + yield* Effect.result(writer.writeTrial(ompTranscript(2, { caseId: "unplanned" }))), + ), + "invalid-record", + ); + assert.strictEqual( + reasonOf( + yield* Effect.result( + writer.writeTrial(ompTranscript(2, { model: "anthropic/other-model" })), + ), + ), + "invalid-record", + ); + assert.strictEqual( + reasonOf(yield* Effect.result(writer.writeTrial(ompTranscript(3)))), + "invalid-record", + ); + const reportContent = `${encodeUnknownJson(v1Report({ target: "omp_harness" }))}\n`; + assert.strictEqual( + reasonOf(yield* Effect.result(writer.bindReport(reportContent, selectedCaseIds))), + "invalid-record", + ); + assert.strictEqual(yield* fs.readFileString(outputPath), durablePrefix); + yield* writer.writeTrial(ompTranscript(2)); + yield* writer.writeTrial(ompTranscript(3)); + yield* writer.bindReport(reportContent, selectedCaseIds); + }), + ); + + it.effect("rejects a schema-valid trial whose encoded UTF-8 record exceeds the cap", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "omp-transcript-cap-" }); + const fileName = "oversized.jsonl"; + const outputPath = path.join(directory, fileName); + const writer = yield* createOmpTranscriptWriter( + transcriptWriterOptions(directory, fileName, { repetitions: 1 }), + ); + const text = "😀".repeat(16_384); + const messages: OmpTrialTranscript["messages"] = Array.from({ length: 17 }, () => ({ + role: "assistant" as const, + type: "text" as const, + text, + })); + const transcript = ompTranscript(1, { messages }); + assert.isAbove( + new TextEncoder().encode(encodeUnknownJson(transcript)).byteLength, + OMP_TRANSCRIPT_MAX_RECORD_BYTES, + ); + assert.strictEqual( + reasonOf(yield* Effect.result(writer.writeTrial(transcript))), + "invalid-record", + ); + assert.deepStrictEqual( + parseRecords(yield* fs.readFileString(outputPath)).map((record) => record.kind), + ["run"], + ); + }), + ); + + it.effect("rejects a malicious file replacement without appending raw chat", () => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const directory = yield* fs.makeTempDirectoryScoped({ prefix: "omp-transcript-replaced-" }); + const fileName = "run.jsonl"; + const outputPath = path.join(directory, fileName); + const movedPath = path.join(directory, "owned.jsonl"); + const writer = yield* createOmpTranscriptWriter( + transcriptWriterOptions(directory, fileName, { repetitions: 1 }), + ); + const ownedPrefix = yield* fs.readFileString(outputPath); + yield* fs.rename(outputPath, movedPath); + yield* fs.writeFileString(outputPath, "foreign bytes\n", { flag: "wx", mode: 0o600 }); + + assert.strictEqual( + reasonOf(yield* Effect.result(writer.writeTrial(ompTranscript(1)))), + "output-conflict", + ); + assert.strictEqual(yield* fs.readFileString(outputPath), "foreign bytes\n"); + assert.strictEqual(yield* fs.readFileString(movedPath), ownedPrefix); + }), + ); + }); +}); diff --git a/packages/evals/src/bin/live.ts b/packages/evals/src/bin/live.ts index 61e07d9..7f6de1a 100755 --- a/packages/evals/src/bin/live.ts +++ b/packages/evals/src/bin/live.ts @@ -55,6 +55,7 @@ import { runLiveEvalSuite, selectCases, } from "../live"; +import type { OmpTrialTranscript } from "../omp-transcript"; import { isOmpApiKeyProvider, isOmpProviderIdentifier, @@ -86,7 +87,12 @@ import { runResponsesApiPluginEvalTrial, type PluginEvalResponsesError } from ". import { isSafePublicEvalText } from "../sanitize"; import type { PluginEvalObservation } from "../contracts"; import { OpenRouterBudgetError, preflightOpenRouterBudget } from "../openrouter-budget"; -import { createLiveEvalJournal, LiveEvalJournalError, withJournaledTrial } from "../trial-journal"; +import { + createLiveEvalJournal, + createOmpTranscriptWriter, + LiveEvalJournalError, + withJournaledTrial, +} from "../trial-journal"; const GIT_STATUS_LIMIT_BYTES = 65_536; const CODEX_PREFLIGHT_LIMIT_BYTES = 1_048_576; @@ -1227,7 +1233,7 @@ const run = (options: LiveEvalCliOptions) => Effect.gen(function* () { const root = process.cwd(); const path = yield* Path.Path; - const runGenerationEvidence = Effect.runPromiseWith( + const runFileSystemEffect = Effect.runPromiseWith( yield* Effect.context(), ); const dispatch = liveEvalTrialDispatch(options); @@ -1246,11 +1252,22 @@ const run = (options: LiveEvalCliOptions) => ? liveEvalConfigurationEvidenceOutputPath(path, outputPath, "configuration-v2") : undefined; const journalPath = outputPath.replace(/\.json$/u, ".journal-v1.jsonl"); + const transcriptPath = + options.runner === "omp" + ? outputPath.replace(/\.json$/u, ".transcripts-v1.jsonl") + : undefined; yield* assertLiveEvalDurableOutputs(outputPath, identityPath, configurationPath); yield* assertPublicEvalAttemptOutputPath(journalPath, outputPath); + if (transcriptPath !== undefined) { + yield* assertPublicEvalAttemptOutputPath(transcriptPath, outputPath); + yield* assertPublicEvalAttemptOutputPath(transcriptPath, journalPath); + } if (options.attemptsOutputPath !== undefined) { yield* assertPublicEvalAttemptOutputPath(options.attemptsOutputPath, outputPath); yield* assertPublicEvalAttemptOutputPath(options.attemptsOutputPath, journalPath); + if (transcriptPath !== undefined) { + yield* assertPublicEvalAttemptOutputPath(options.attemptsOutputPath, transcriptPath); + } if (identityPath !== undefined) { yield* assertPublicEvalAttemptOutputPath(options.attemptsOutputPath, identityPath); } @@ -1291,6 +1308,25 @@ const run = (options: LiveEvalCliOptions) => caseIds: plannedCaseIds, repetitions: options.repetitions, }); + const transcriptWriter = + transcriptPath === undefined + ? undefined + : yield* createOmpTranscriptWriter({ + outputDir: path.dirname(transcriptPath), + fileName: path.basename(transcriptPath), + runId: options.runId, + candidate: options.candidate, + model, + serverUrl: PRODUCTION_MCP_URL, + suiteId: suite.suite.id, + suiteVersion: suite.version, + fixtureVersion: 1, + catalogSha, + reasoning: options.reasoning, + accountClass: options.accountClass, + caseIds: plannedCaseIds, + repetitions: options.repetitions, + }); const credentials = yield* loadLiveEvalCredentials( options.runner === "omp" ? { runner: "omp", authMode: options.auth.mode } : options.runner, ); @@ -1431,7 +1467,7 @@ const run = (options: LiveEvalCliOptions) => endpoint: options.endpoint, expectedProvider: options.expectedProvider, onGenerationEvidence: (evidence) => - runGenerationEvidence(journal.generation(dispatchId, evidence)), + runFileSystemEffect(journal.generation(dispatchId, evidence)), reasoning: options.reasoning, runId: input.runId, repetition: input.repetition, @@ -1485,7 +1521,11 @@ const run = (options: LiveEvalCliOptions) => timeoutMs: options.timeoutMs, }); case "omp": { - if (credentials.runner !== "omp" || ompRuntimeDirectory === undefined) { + if ( + credentials.runner !== "omp" || + ompRuntimeDirectory === undefined || + transcriptWriter === undefined + ) { return Effect.fail(new LiveEvalCliError({ reason: "omp-preflight-failed" })); } const auth = @@ -1515,6 +1555,8 @@ const run = (options: LiveEvalCliOptions) => reasoning: options.reasoning, mcpAuthorization: credentials.accessToken, timeoutMs: options.timeoutMs, + onTranscript: (transcript: OmpTrialTranscript, signal: AbortSignal) => + runFileSystemEffect(transcriptWriter.writeTrial(transcript), { signal }), }).pipe( Effect.filterOrFail( (observation) => observation.model === input.model, @@ -1549,6 +1591,9 @@ const run = (options: LiveEvalCliOptions) => }); } yield* journal.bindReport(encoded, selectedCaseIds); + if (transcriptWriter !== undefined) { + yield* transcriptWriter.bindReport(encoded, selectedCaseIds); + } if ( requestedRouting !== undefined && identityPath !== undefined && diff --git a/packages/evals/src/index.ts b/packages/evals/src/index.ts index a87d48a..e0b95c0 100644 --- a/packages/evals/src/index.ts +++ b/packages/evals/src/index.ts @@ -72,12 +72,41 @@ export { } from "./openrouter-budget"; export { createLiveEvalJournal, + createOmpTranscriptWriter, withJournaledTrial, LiveEvalJournalError, + OMP_TRANSCRIPT_MAX_RECORD_BYTES, + OmpTranscriptJournalRecordSchema, type LiveEvalJournal, type LiveEvalJournalOptions, type LiveEvalJournalRecord, + type OmpTranscriptJournalRecord, + type OmpTranscriptWriter, + type OmpTranscriptWriterOptions, } from "./trial-journal"; +export { + createOmpTranscriptCollector, + OMP_TRANSCRIPT_MAX_BYTES, + OMP_TRANSCRIPT_MAX_FIELD_BYTES, + OMP_TRANSCRIPT_MAX_MESSAGES, + OMP_TRANSCRIPT_OMITTED, + OMP_TRANSCRIPT_REDACTION, + OMP_TRANSCRIPT_SCHEMA_VERSION, + OmpTranscriptMessageSchema, + OmpTranscriptTextMessageSchema, + OmpTranscriptToolCallMessageSchema, + OmpTranscriptToolResultMessageSchema, + OmpTrialTranscriptSchema, + OmpTrialTranscriptStatusSchema, + type OmpTranscriptCollector, + type OmpTranscriptCollectorResult, + type OmpTranscriptMessage, + type OmpTranscriptTextMessage, + type OmpTranscriptToolCallMessage, + type OmpTranscriptToolResultMessage, + type OmpTrialTranscript, + type OmpTrialTranscriptStatus, +} from "./omp-transcript"; export { PluginEvalResponsesDecodeError, PluginEvalResponsesHttpError, @@ -129,6 +158,7 @@ export { PluginEvalOmpHarnessMcpError, PluginEvalOmpHarnessProcessError, PluginEvalOmpHarnessTimeoutError, + PluginEvalOmpHarnessTranscriptError, type PrepareOmpHarnessRuntimeOptions, type PreparedOmpHarnessRuntime, type OmpHarnessTrialOptions, diff --git a/packages/evals/src/omp-harness.ts b/packages/evals/src/omp-harness.ts index fb07159..0917dd7 100644 --- a/packages/evals/src/omp-harness.ts +++ b/packages/evals/src/omp-harness.ts @@ -14,8 +14,15 @@ import { SKILL_NAMES, type SkillName, } from "@askgina/contracts"; -import { jsonSchema, type StepResult, type ToolSet } from "ai"; import { + jsonSchema, + type GenerateTextResult, + type StepResult, + type StreamTextResult, + type ToolSet, +} from "ai"; +import { + Cause, Clock, Data, DateTime, @@ -23,20 +30,30 @@ import { Effect, Exit, FileSystem, + Fiber, Function, Option, Path, Redacted, Scope, + Stream, } from "effect"; import type { PluginEvalCase, PluginEvalObservation, PluginEvalToolCall } from "./contracts"; import { createLocalHarnessSandbox } from "./local-harness-sandbox"; +import { + createOmpTranscriptCollector, + type OmpTranscriptCollector, + type OmpTrialTranscript, +} from "./omp-transcript"; const DEFAULT_TIMEOUT_MS = 120_000; const MAX_MCP_TOOL_PAGES = 32; const MAX_MCP_CLOSE_WAIT_MS = 1_000; const MAX_SESSION_DESTROY_WAIT_MS = 8_000; +const MAX_TRANSCRIPT_CALLBACK_WAIT_MS = 5_000; +const OMP_TRANSCRIPT_TOOL_EXECUTION_FAILED = "[tool execution failed]"; +const OMP_TRANSCRIPT_TOOL_OUTPUT_DENIED = "[tool output denied]"; const OMP_INCOMPLETE_GENERATION_ERROR = "OMP generation did not complete with a final answer"; const OMP_TOOL_EXECUTION_ERROR = "OMP tool execution failed"; const OMP_EVAL_PROVIDER_ALIAS = "omp-eval"; @@ -129,6 +146,10 @@ export interface OmpHarnessTrialOptions { readonly timeoutMs: number; readonly serverUrl?: string; readonly sandbox?: HarnessV1SandboxProvider; + readonly onTranscript?: ( + transcript: OmpTrialTranscript, + signal: AbortSignal, + ) => PromiseLike; } type ValidatedOmpAuth = @@ -182,6 +203,11 @@ interface ObservedOmpToolCalls { readonly failedNativeRead: boolean; } +type OmpGenerationEvidence = Pick< + GenerateTextResult, never>, + "text" | "finishReason" | "usage" | "steps" +>; + export class PluginEvalOmpHarnessExecutableError extends Data.TaggedError( "PluginEvalOmpHarnessExecutableError", )<{ @@ -221,13 +247,22 @@ export class PluginEvalOmpHarnessTimeoutError extends Data.TaggedError( readonly timeoutMs: number; }> {} +/** Capture failure fails a successful trial without replacing an existing trial failure. */ +export class PluginEvalOmpHarnessTranscriptError extends Data.TaggedError( + "PluginEvalOmpHarnessTranscriptError", +)<{ + readonly caseId: string; + readonly reason: "write-failed" | "write-timeout"; +}> {} + export type PluginEvalOmpHarnessError = | PluginEvalOmpHarnessExecutableError | PluginEvalOmpHarnessRequestError | PluginEvalOmpHarnessSpawnError | PluginEvalOmpHarnessMcpError | PluginEvalOmpHarnessProcessError - | PluginEvalOmpHarnessTimeoutError; + | PluginEvalOmpHarnessTimeoutError + | PluginEvalOmpHarnessTranscriptError; const catalogsMatch = (left: readonly string[], right: readonly string[]): boolean => { if (left.length !== right.length) return false; @@ -477,7 +512,138 @@ const isKnownHarnessError = (error: unknown): error is PluginEvalOmpHarnessError error instanceof PluginEvalOmpHarnessSpawnError || error instanceof PluginEvalOmpHarnessMcpError || error instanceof PluginEvalOmpHarnessProcessError || - error instanceof PluginEvalOmpHarnessTimeoutError; + error instanceof PluginEvalOmpHarnessTimeoutError || + error instanceof PluginEvalOmpHarnessTranscriptError; + +const generationError = (caseId: string, error: unknown): PluginEvalOmpHarnessError => + isKnownHarnessError(error) + ? error + : new PluginEvalOmpHarnessProcessError({ caseId, reason: "generation-failed" }); + +const consumeTranscriptStream = ( + streamed: StreamTextResult, never>, + collector: OmpTranscriptCollector, + caseId: string, + signal: AbortSignal, +): Promise => + Effect.runPromise( + Stream.fromAsyncIterable(streamed.stream, (error) => generationError(caseId, error)).pipe( + Stream.runForEach((part) => + Effect.sync(() => { + switch (part.type) { + case "text-delta": + collector.assistant(part.text); + break; + case "tool-call": + collector.toolCall(part.toolCallId, part.toolName, part.input); + break; + case "tool-result": + collector.toolResult(part.toolCallId, part.toolName, part.output, false); + break; + case "tool-error": + collector.toolResult( + part.toolCallId, + part.toolName, + OMP_TRANSCRIPT_TOOL_EXECUTION_FAILED, + true, + ); + break; + case "tool-output-denied": + collector.toolResult( + part.toolCallId, + part.toolName, + OMP_TRANSCRIPT_TOOL_OUTPUT_DENIED, + true, + ); + break; + } + }), + ), + Effect.flatMap(() => + Effect.tryPromise({ + try: () => + Promise.all([ + streamed.text, + streamed.finishReason, + streamed.usage, + streamed.steps, + ]).then(([text, finishReason, usage, steps]) => ({ text, finishReason, usage, steps })), + catch: (error) => generationError(caseId, error), + }), + ), + ), + { signal }, + ); + +const isTypedTimeoutError = (error: unknown): boolean => + Cause.isTimeoutError(error) || + (typeof error === "object" && + error !== null && + "_tag" in error && + typeof error._tag === "string" && + error._tag.endsWith("TimeoutError")); + +const transcriptStatus = ( + exit: Exit.Exit, +): OmpTrialTranscript["status"] => { + if (Exit.isSuccess(exit)) return exit.value.status; + if (Cause.hasInterrupts(exit.cause)) return "interruption"; + return isTypedTimeoutError(Option.getOrUndefined(Cause.findErrorOption(exit.cause))) + ? "timeout" + : "failed"; +}; + +const invokeTranscriptCallback = ( + callback: (transcript: OmpTrialTranscript, signal: AbortSignal) => PromiseLike, + transcript: OmpTrialTranscript, + caseId: string, +): Effect.Effect => + Effect.gen(function* () { + const fiber = yield* Effect.tryPromise({ + try: (signal) => callback(transcript, signal), + catch: () => new PluginEvalOmpHarnessTranscriptError({ caseId, reason: "write-failed" }), + }).pipe(Effect.forkChild({ startImmediately: true, uninterruptible: false })); + const outcome = yield* Effect.raceFirst( + Fiber.await(fiber).pipe(Effect.map((exit) => ({ type: "settled" as const, exit }))), + Effect.sleep(Duration.millis(MAX_TRANSCRIPT_CALLBACK_WAIT_MS)).pipe( + Effect.as({ type: "timed-out" as const }), + ), + ); + if (outcome.type === "timed-out") { + yield* Fiber.interrupt(fiber); + return yield* new PluginEvalOmpHarnessTranscriptError({ + caseId, + reason: "write-timeout", + }); + } + return yield* outcome.exit; + }); + +const finalizeTranscript = ( + collector: OmpTranscriptCollector, + callback: (transcript: OmpTrialTranscript, signal: AbortSignal) => PromiseLike, + metadata: Omit, + exit: Exit.Exit, +): Effect.Effect => + Effect.gen(function* () { + const collected = yield* Effect.try({ + try: () => collector.finish(), + catch: () => + new PluginEvalOmpHarnessTranscriptError({ + caseId: metadata.caseId, + reason: "write-failed", + }), + }); + const transcript = { + ...metadata, + status: transcriptStatus(exit), + messages: collected.messages, + truncated: collected.truncated, + } satisfies OmpTrialTranscript; + yield* invokeTranscriptCallback(callback, transcript, metadata.caseId); + }).pipe( + Exit.isSuccess(exit) ? Function.identity : Effect.catch((error) => Effect.logWarning(error)), + ); const parseProviderBaseUrl = (value: string): string | undefined => { if (value.includes("@") || value.includes("\n") || value.includes("\0") || value.includes("\\")) { @@ -1496,8 +1662,17 @@ export const runOmpHarnessPluginEvalTrial = Function.dual< const startedMillis = yield* Clock.currentTimeMillis; const startedAt = DateTime.formatIso(DateTime.makeUnsafe(startedMillis)); const deadlineMillis = startedMillis + validated.timeoutMs; + const prompt = promptText(evalCase); + const transcriptCallback = options.onTranscript; + const transcriptCollector = + transcriptCallback === undefined + ? undefined + : createOmpTranscriptCollector([ + validated.mcpAuthorization, + ...(validated.auth.mode === "api-key" ? [validated.auth.apiKey] : []), + ]); - return yield* Effect.gen(function* () { + const trial = Effect.gen(function* () { const skills = yield* withRunDeadline( loadStagedSkills(validated.runtimeDirectory, evalCase.id), evalCase.id, @@ -1629,21 +1804,30 @@ export const runOmpHarnessPluginEvalTrial = Function.dual< validated.timeoutMs, deadlineMillis, ); - const generatedText = yield* Effect.tryPromise({ - try: (signal) => - agent.generate({ - session, - prompt: promptText(evalCase), - abortSignal: signal, - }), - catch: (error) => - isKnownHarnessError(error) - ? error - : new PluginEvalOmpHarnessProcessError({ - caseId: evalCase.id, - reason: "generation-failed", - }), - }); + let generatedText: OmpGenerationEvidence; + if (transcriptCollector === undefined) { + generatedText = yield* Effect.tryPromise({ + try: (signal) => + agent.generate({ session, prompt, abortSignal: signal }), + catch: (error) => generationError(evalCase.id, error), + }); + } else { + transcriptCollector.user(prompt); + generatedText = yield* Effect.tryPromise({ + try: (signal) => + agent + .stream({ session, prompt, abortSignal: signal }) + .then((streamed) => + consumeTranscriptStream( + streamed, + transcriptCollector, + evalCase.id, + signal, + ), + ), + catch: (error) => generationError(evalCase.id, error), + }); + } yield* ensureBeforeDeadline( evalCase.id, validated.timeoutMs, @@ -1743,6 +1927,25 @@ export const runOmpHarnessPluginEvalTrial = Function.dual< orElse: () => Effect.fail(timeoutError(evalCase.id, validated.timeoutMs)), }), ); + if (transcriptCollector === undefined || transcriptCallback === undefined) { + return yield* trial; + } + return yield* trial.pipe( + Effect.onExit((exit) => + finalizeTranscript( + transcriptCollector, + transcriptCallback, + { + runId: validated.runId, + caseId: evalCase.id, + repetition: validated.repetition, + model: validated.modelIdentity, + startedAt, + }, + exit, + ), + ), + ); }).pipe( Effect.withSpan("plugin_evals.omp_harness_trial", { attributes: { diff --git a/packages/evals/src/omp-transcript.ts b/packages/evals/src/omp-transcript.ts new file mode 100644 index 0000000..4577bdf --- /dev/null +++ b/packages/evals/src/omp-transcript.ts @@ -0,0 +1,470 @@ +import { Schema } from "effect"; + +import { findPublicTextViolations } from "./sanitize"; + +export const OMP_TRANSCRIPT_SCHEMA_VERSION = "omp-transcript.v1" as const; +export const OMP_TRANSCRIPT_MAX_MESSAGES = 512; +export const OMP_TRANSCRIPT_MAX_BYTES = 1_048_576; +export const OMP_TRANSCRIPT_MAX_FIELD_BYTES = 65_536; +export const OMP_TRANSCRIPT_REDACTION = "[redacted]"; +export const OMP_TRANSCRIPT_OMITTED = "[omitted]"; + +const DELTA_BUFFER_SLACK_BYTES = OMP_TRANSCRIPT_MAX_FIELD_BYTES; +const MAX_JSON_DEPTH = 32; +const MAX_JSON_NODES = 4_096; +const MAX_LABEL_LENGTH = 256; +const MIN_SECRET_PREFIX_LENGTH = 3; +const BOUNDARY_WINDOW = 256; +const TRUNCATION_SUFFIX = "...[truncated]"; +const UTC_TIMESTAMP = /^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{3})?Z$/u; +const PRIVATE_KEY_END = "-----END"; +// Mirrors the private credential-key policy in sanitize.ts for JSON object keys. +const CREDENTIAL_KEY = + /^(?:[A-Za-z0-9]+[_-])*(?:api[_-]?key|authorization|bearer|cookie|credential|password|private[_-]?key|secret(?:[_-]?key)?|session(?:[_-]?id)?|token(?:[_-]?id)?)$/i; + +const UTF8 = new TextEncoder(); +const UTF8_DECODER = new TextDecoder(); +const UTF8_FATAL_DECODER = new TextDecoder("utf-8", { fatal: true }); + +const BoundedTextSchema = Schema.String.check( + Schema.isMaxLength(OMP_TRANSCRIPT_MAX_FIELD_BYTES + TRUNCATION_SUFFIX.length + 64), +); +const BoundedLabelSchema = Schema.String.check(Schema.isMaxLength(MAX_LABEL_LENGTH)); + +export const OmpTranscriptTextMessageSchema = Schema.Struct({ + role: Schema.Literals(["user", "assistant"]), + type: Schema.Literal("text"), + text: BoundedTextSchema, +}); + +export const OmpTranscriptToolCallMessageSchema = Schema.Struct({ + role: Schema.Literal("assistant"), + type: Schema.Literal("tool-call"), + toolCallId: BoundedLabelSchema, + toolName: Schema.NonEmptyString.check(Schema.isMaxLength(MAX_LABEL_LENGTH)), + input: Schema.Json, +}); + +export const OmpTranscriptToolResultMessageSchema = Schema.Struct({ + role: Schema.Literal("tool"), + type: Schema.Literal("tool-result"), + toolCallId: BoundedLabelSchema, + toolName: Schema.NonEmptyString.check(Schema.isMaxLength(MAX_LABEL_LENGTH)), + output: Schema.Json, + isError: Schema.Boolean, +}); + +export const OmpTranscriptMessageSchema = Schema.Union([ + OmpTranscriptTextMessageSchema, + OmpTranscriptToolCallMessageSchema, + OmpTranscriptToolResultMessageSchema, +]); + +export const OmpTrialTranscriptStatusSchema = Schema.Literals([ + "completed", + "failed", + "blocked", + "timeout", + "interruption", +]); + +export const OmpTrialTranscriptSchema = Schema.Struct({ + runId: Schema.NonEmptyString.check(Schema.isMaxLength(MAX_LABEL_LENGTH)), + caseId: Schema.NonEmptyString.check(Schema.isMaxLength(MAX_LABEL_LENGTH)), + repetition: Schema.Int.check(Schema.isGreaterThan(0)), + model: Schema.NonEmptyString.check(Schema.isMaxLength(MAX_LABEL_LENGTH)), + startedAt: Schema.NonEmptyString.check(Schema.isMaxLength(64), Schema.isPattern(UTC_TIMESTAMP)), + status: OmpTrialTranscriptStatusSchema, + messages: Schema.Array(OmpTranscriptMessageSchema).check( + Schema.isMaxLength(OMP_TRANSCRIPT_MAX_MESSAGES), + ), + truncated: Schema.Boolean, +}); + +export type OmpTranscriptTextMessage = typeof OmpTranscriptTextMessageSchema.Type; +export type OmpTranscriptToolCallMessage = typeof OmpTranscriptToolCallMessageSchema.Type; +export type OmpTranscriptToolResultMessage = typeof OmpTranscriptToolResultMessageSchema.Type; +export type OmpTranscriptMessage = typeof OmpTranscriptMessageSchema.Type; +export type OmpTrialTranscriptStatus = typeof OmpTrialTranscriptStatusSchema.Type; +export type OmpTrialTranscript = typeof OmpTrialTranscriptSchema.Type; + +export interface OmpTranscriptCollectorResult { + readonly messages: readonly OmpTranscriptMessage[]; + readonly truncated: boolean; +} + +export interface OmpTranscriptCollector { + readonly user: (text: string) => void; + readonly assistant: (delta: string) => void; + readonly toolCall: (toolCallId: string, toolName: string, input: unknown) => void; + readonly toolResult: ( + toolCallId: string, + toolName: string, + output: unknown, + isError: boolean, + ) => void; + readonly finish: () => OmpTranscriptCollectorResult; +} + +const omittedValue = (reason: string): string => `[omitted:${reason}]`; + +const isPlainJsonObject = (value: object): value is Record => { + const prototype = Object.getPrototypeOf(value); + return prototype === Object.prototype || prototype === null; +}; + +/** + * Collects a private per-trial chat transcript from OMP harness stream events. + * + * Redaction runs on coalesced terminal text and on every JSON string value + * before storage: explicit `secrets` first, then `findPublicTextViolations` + * spans (excluding `host-absolute-path`, which is not a credential), then a + * tail-prefix mask so a stream that dies mid-secret never persists a usable + * fragment. Unsupported or oversized values become explicit `[omitted:*]` + * placeholders and raise `truncated`; nothing is silently dropped. + */ +export const createOmpTranscriptCollector = ( + secrets: readonly string[], +): OmpTranscriptCollector => { + const explicitSecrets = [...new Set(secrets.filter((secret) => secret.length > 0))].sort( + (left, right) => right.length - left.length, + ); + const redactionMarker = explicitSecrets.some((secret) => + OMP_TRANSCRIPT_REDACTION.includes(secret), + ) + ? "" + : OMP_TRANSCRIPT_REDACTION; + + const messages: OmpTranscriptMessage[] = []; + let bytesUsed = 2; // JSON array brackets; commas are charged when messages are added. + let truncated = false; + let sealed = false; + let outputExhausted = false; + let finished: OmpTranscriptCollectorResult | undefined; + + let pendingRole: "user" | "assistant" | null = null; + let pendingParts: string[] = []; + let pendingBytes = 0; + let pendingDropped = false; + + const maskSecretTail = (text: string): string => { + for (const secret of explicitSecrets) { + const longest = Math.min(secret.length - 1, text.length); + for (let prefix = longest; prefix >= MIN_SECRET_PREFIX_LENGTH; prefix -= 1) { + if (text.endsWith(secret.slice(0, prefix))) { + return text.slice(0, text.length - prefix) + redactionMarker; + } + } + } + return text; + }; + + const redactText = (input: string): string => { + let text = input; + for (const secret of explicitSecrets) { + if (text.includes(secret)) text = text.split(secret).join(redactionMarker); + } + // Violations arrive sorted by start; a shorter assignment/header match on a + // private-key BEGIN line must extend, not suppress, the overlapping block + // span, so overlapping spans are unioned by their maximum end. + const spans: Array<[number, number]> = []; + for (const violation of findPublicTextViolations(text)) { + if (violation.kind === "host-absolute-path") continue; + const last = spans[spans.length - 1]; + if (last !== undefined && violation.index < last[1] && violation.kind !== "private-key") + continue; + let end: number; + if (violation.kind === "private-key") { + const marker = text.indexOf(PRIVATE_KEY_END, violation.index); + const newline = marker === -1 ? -1 : text.indexOf("\n", marker); + end = marker === -1 ? text.length : newline === -1 ? text.length : newline; + } else { + const newline = text.indexOf("\n", violation.index); + end = newline === -1 ? text.length : newline; + } + if (last !== undefined && violation.index < last[1]) { + if (end > last[1]) last[1] = end; + continue; + } + spans.push([violation.index, end]); + } + if (spans.length > 0) { + let redacted = ""; + let cursor = 0; + for (const [start, end] of spans) { + redacted += text.slice(cursor, start) + redactionMarker; + cursor = end; + } + text = redacted + text.slice(cursor); + } + return maskSecretTail(text); + }; + + // Input must already be redacted; a byte cut can only expose a fragment when + // the source text was itself truncated, so the cut retreats to whitespace and + // a single oversized token is dropped whole rather than persisted as a prefix. + const capText = (redacted: string): string => { + const bytes = UTF8.encode(redacted); + if (bytes.byteLength <= OMP_TRANSCRIPT_MAX_FIELD_BYTES) return redacted; + truncated = true; + let kept = UTF8_DECODER.decode(bytes.slice(0, OMP_TRANSCRIPT_MAX_FIELD_BYTES)); + let boundary = -1; + for ( + let index = kept.length - 1; + index >= Math.max(0, kept.length - BOUNDARY_WINDOW); + index -= 1 + ) { + if (/\s/u.test(kept[index])) { + boundary = index; + break; + } + } + if (boundary === -1) return omittedValue("too-large"); + kept = maskSecretTail(kept.slice(0, boundary)); + return kept + TRUNCATION_SUFFIX; + }; + + const capLabel = (value: string): string => { + if (value.length > MAX_LABEL_LENGTH) { + truncated = true; + return omittedValue("too-large"); + } + // Redaction can expand a label (a short secret becomes the longer marker), + // so the bound is enforced again on the redacted text. + const redacted = redactText(value); + if (redacted.length > MAX_LABEL_LENGTH) { + truncated = true; + return omittedValue("too-large"); + } + return redacted; + }; + + const sanitizeJson = (value: unknown): Schema.Json => { + let nodes = 0; + let traversalBytes = 0; + let traversalBudgetExceeded = false; + const chargeText = (text: string): boolean => { + const remaining = OMP_TRANSCRIPT_MAX_FIELD_BYTES - traversalBytes; + if (text.length > remaining) { + traversalBudgetExceeded = true; + truncated = true; + return false; + } + const size = UTF8.encode(text).byteLength; + if (size > remaining) { + traversalBudgetExceeded = true; + truncated = true; + return false; + } + traversalBytes += size; + return true; + }; + const visit = (current: unknown, depth: number, seen: Set): Schema.Json => { + nodes += 1; + if (nodes > MAX_JSON_NODES) { + truncated = true; + return omittedValue("too-large"); + } + if (current === null) return null; + switch (typeof current) { + case "string": + return chargeText(current) ? redactText(current) : omittedValue("too-large"); + case "number": + if (Number.isFinite(current)) return current; + truncated = true; + return omittedValue("non-finite-number"); + case "boolean": + return current; + case "object": { + if (seen.has(current)) { + truncated = true; + return omittedValue("circular-reference"); + } + if (depth >= MAX_JSON_DEPTH) { + truncated = true; + return omittedValue("too-deep"); + } + if (Array.isArray(current)) { + seen.add(current); + const items: Schema.Json[] = []; + for (const item of current) { + if (nodes >= MAX_JSON_NODES) { + truncated = true; + items.push(omittedValue("too-large")); + break; + } + const safeItem = visit(item, depth + 1, seen); + if (traversalBudgetExceeded) break; + items.push(safeItem); + } + seen.delete(current); + return items; + } + if (!isPlainJsonObject(current)) { + truncated = true; + return omittedValue("unsupported-type"); + } + seen.add(current); + const record: Record = {}; + for (const key in current) { + if (!Object.hasOwn(current, key)) continue; + if (nodes >= MAX_JSON_NODES) { + truncated = true; + record[omittedValue("too-large")] = OMP_TRANSCRIPT_OMITTED; + break; + } + if (!chargeText(key)) break; + const safeKey = redactText(key); + if (Object.hasOwn(record, safeKey)) { + truncated = true; + continue; + } + let safeValue: Schema.Json = redactionMarker; + if (!CREDENTIAL_KEY.test(key)) { + try { + safeValue = visit(current[key], depth + 1, seen); + } catch { + truncated = true; + safeValue = omittedValue("unsupported-type"); + } + } + if (traversalBudgetExceeded) break; + Object.defineProperty(record, safeKey, { + configurable: true, + enumerable: true, + value: safeValue, + writable: true, + }); + } + seen.delete(current); + return record; + } + default: + truncated = true; + return omittedValue("unsupported-type"); + } + }; + try { + const sanitized = visit(value, 0, new Set()); + if (traversalBudgetExceeded) return omittedValue("too-large"); + if (UTF8.encode(JSON.stringify(sanitized)).byteLength > OMP_TRANSCRIPT_MAX_FIELD_BYTES) { + truncated = true; + return omittedValue("too-large"); + } + return sanitized; + } catch { + truncated = true; + return omittedValue("unsupported-type"); + } + }; + + const canAcceptMessage = (): boolean => { + if (outputExhausted) return false; + if (messages.length >= OMP_TRANSCRIPT_MAX_MESSAGES || bytesUsed >= OMP_TRANSCRIPT_MAX_BYTES) { + truncated = true; + outputExhausted = true; + return false; + } + return true; + }; + + const pushMessage = (message: OmpTranscriptMessage): void => { + if (!canAcceptMessage()) return; + const size = UTF8.encode(JSON.stringify(message)).byteLength; + const separator = messages.length === 0 ? 0 : 1; + if (bytesUsed + separator + size > OMP_TRANSCRIPT_MAX_BYTES) { + truncated = true; + outputExhausted = true; + return; + } + messages.push(message); + bytesUsed += separator + size; + }; + + const flushPending = (): void => { + if (pendingRole === null) return; + const role = pendingRole; + const parts = pendingParts; + pendingRole = null; + pendingParts = []; + pendingBytes = 0; + if (pendingDropped) { + pendingDropped = false; + truncated = true; + } + const text = parts.join(""); + if (text.length === 0) return; + pushMessage({ role, type: "text", text: capText(redactText(text)) }); + }; + + const appendText = (role: "user" | "assistant", text: string): void => { + if (sealed || typeof text !== "string" || text.length === 0) return; + if (pendingRole !== null && pendingRole !== role) flushPending(); + if (!canAcceptMessage()) return; + pendingRole = role; + const remaining = OMP_TRANSCRIPT_MAX_FIELD_BYTES + DELTA_BUFFER_SLACK_BYTES - pendingBytes; + if (remaining <= 0) { + pendingDropped = true; + return; + } + const candidate = text.slice(0, Math.min(text.length, remaining + 1)); + const bytes = UTF8.encode(candidate); + if (bytes.byteLength <= remaining) { + pendingParts.push(candidate); + pendingBytes += bytes.byteLength; + if (candidate.length < text.length) pendingDropped = true; + return; + } + let end = remaining; + while (end > 0) { + try { + const prefix = UTF8_FATAL_DECODER.decode(bytes.slice(0, end)); + if (prefix.length > 0) { + pendingParts.push(prefix); + pendingBytes += end; + } + break; + } catch { + end -= 1; + } + } + pendingDropped = true; + }; + + return { + user: (text) => appendText("user", text), + assistant: (delta) => appendText("assistant", delta), + toolCall: (toolCallId, toolName, input) => { + if (sealed) return; + flushPending(); + if (!canAcceptMessage()) return; + pushMessage({ + role: "assistant", + type: "tool-call", + toolCallId: capLabel(toolCallId), + toolName: capLabel(toolName) || OMP_TRANSCRIPT_OMITTED, + input: sanitizeJson(input), + }); + }, + toolResult: (toolCallId, toolName, output, isError) => { + if (sealed) return; + flushPending(); + if (!canAcceptMessage()) return; + pushMessage({ + role: "tool", + type: "tool-result", + toolCallId: capLabel(toolCallId), + toolName: capLabel(toolName) || OMP_TRANSCRIPT_OMITTED, + output: sanitizeJson(output), + isError: isError === true, + }); + }, + finish: () => { + if (finished !== undefined) return finished; + flushPending(); + sealed = true; + finished = { messages: Object.freeze(messages.slice()), truncated }; + return finished; + }, + }; +}; diff --git a/packages/evals/src/trial-journal.ts b/packages/evals/src/trial-journal.ts index bbc6739..eb9d7eb 100644 --- a/packages/evals/src/trial-journal.ts +++ b/packages/evals/src/trial-journal.ts @@ -15,6 +15,12 @@ import { } from "effect"; import { sha256Hex } from "./canonical-json"; +import { + OMP_TRANSCRIPT_MAX_BYTES, + OMP_TRANSCRIPT_SCHEMA_VERSION, + OmpTrialTranscriptSchema, + type OmpTrialTranscript, +} from "./omp-transcript"; import { SanitizedEvalRunReportSchema } from "./report"; import { PluginEvalTargetSchema, @@ -36,6 +42,7 @@ export const LIVE_EVAL_JOURNAL_MAX_REPETITIONS = 5; export const LIVE_EVAL_JOURNAL_MAX_STEPS = 32; export const LIVE_EVAL_JOURNAL_MAX_TRIALS = LIVE_EVAL_JOURNAL_MAX_CASES * LIVE_EVAL_JOURNAL_MAX_REPETITIONS; +export const OMP_TRANSCRIPT_MAX_RECORD_BYTES = OMP_TRANSCRIPT_MAX_BYTES + 16_384; const MAX_LABEL_LENGTH = 128; const MAX_MODEL_LENGTH = 128; @@ -152,6 +159,40 @@ const BoundRecordSchema = Schema.Struct({ sourceReportSha256: Sha256, }); +const OmpTranscriptRunRecordSchema = Schema.Struct({ + kind: Schema.Literal("run"), + schemaVersion: Schema.Literal(OMP_TRANSCRIPT_SCHEMA_VERSION), + runId: BoundedLabel, + candidate: BoundedLabel, + target: Schema.Literal("omp_harness"), + model: BoundedModel, + serverUrl: Schema.NonEmptyString.check(Schema.isPattern(SINGLE_LINE)), + suiteId: BoundedLabel, + suiteVersion: PositiveVersion, + fixtureVersion: PositiveVersion, + catalogSha: Sha256, + reasoning: Schema.optionalKey(BoundedLabel), + accountClass: BoundedLabel, + caseIds: PlannedCaseIds, + repetitions: Repetition, +}); +const OmpTranscriptTrialRecordSchema = Schema.Struct({ + kind: Schema.Literal("trial"), + dispatchId: DispatchId, + transcript: OmpTrialTranscriptSchema, +}); +const OmpTranscriptBoundRecordSchema = Schema.Struct({ + kind: Schema.Literal("report-bound"), + sourceReportSha256: Sha256, +}); + +export const OmpTranscriptJournalRecordSchema = Schema.Union([ + OmpTranscriptRunRecordSchema, + OmpTranscriptTrialRecordSchema, + OmpTranscriptBoundRecordSchema, +]); +export type OmpTranscriptJournalRecord = typeof OmpTranscriptJournalRecordSchema.Type; + export type LiveEvalJournalRecord = | typeof RunRecordSchema.Type | typeof StartedRecordSchema.Type @@ -191,6 +232,24 @@ export interface LiveEvalJournalOptions { readonly repetitions: number; } +export interface OmpTranscriptWriterOptions extends Omit< + LiveEvalJournalOptions, + "outputPath" | "target" +> { + readonly outputDir: string; + readonly fileName: string; +} + +export interface OmpTranscriptWriter { + readonly writeTrial: ( + transcript: OmpTrialTranscript, + ) => Effect.Effect; + readonly bindReport: ( + reportContent: string, + selectedCaseIds: readonly string[], + ) => Effect.Effect; +} + export interface LiveEvalJournal { readonly startTrial: (input: { readonly caseId: string; @@ -229,6 +288,16 @@ const encodeStarted = Schema.encodeEffect(Schema.fromJsonString(StartedRecordSch const encodeGeneration = Schema.encodeEffect(Schema.fromJsonString(GenerationRecordSchema)); const encodeFinished = Schema.encodeEffect(Schema.fromJsonString(FinishedRecordSchema)); const encodeBound = Schema.encodeEffect(Schema.fromJsonString(BoundRecordSchema)); +const decodeOmpTranscript = Schema.decodeUnknownEffect(OmpTrialTranscriptSchema, DECODE_OPTIONS); +const encodeOmpTranscriptRun = Schema.encodeEffect( + Schema.fromJsonString(OmpTranscriptRunRecordSchema), +); +const encodeOmpTranscriptTrial = Schema.encodeEffect( + Schema.fromJsonString(OmpTranscriptTrialRecordSchema), +); +const encodeOmpTranscriptBound = Schema.encodeEffect( + Schema.fromJsonString(OmpTranscriptBoundRecordSchema), +); const admitLabel = (value: string): Effect.Effect => decodeLabel(value).pipe( @@ -248,6 +317,129 @@ const isUnsafePath = (outputPath: string): boolean => outputPath.includes("\0") || outputPath.split(/[\\/]/u).includes(".."); +const isUnsafeFileName = (fileName: string): boolean => + fileName.trim().length === 0 || + fileName !== fileName.trim() || + fileName.includes("\0") || + fileName.includes("/") || + fileName.includes("\\") || + fileName === "." || + fileName === ".."; + +interface ExclusiveJournalSink { + readonly append: (record: string, maxBytes?: number) => Effect.Effect; + readonly exclusive: ( + effect: Effect.Effect, + ) => Effect.Effect; +} + +/** + * Interruption policy for sink critical sections. + * - "defer": writes run uninterruptibly; interruption is deferred until the section completes. + * - "interrupt": writes may be interrupted; an interrupted or failed append poisons the sink. + */ +type JournalSinkInterruption = "defer" | "interrupt"; + +/** Opens a private append-only JSONL sink with exclusive creation, ownership checks and fsync. */ +const openExclusiveJournalSink = ( + output: string, + interruption: JournalSinkInterruption, +): Effect.Effect< + ExclusiveJournalSink, + LiveEvalJournalError, + FileSystem.FileSystem | Path.Path | Scope.Scope +> => + Effect.gen(function* () { + const fs = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + if (yield* fs.exists(output).pipe(Effect.mapError(() => fail("write-failed")))) { + return yield* fail("output-exists"); + } + let ancestor = path.dirname(output); + while (!(yield* fs.exists(ancestor).pipe(Effect.mapError(() => fail("invalid-path"))))) { + const parent = path.dirname(ancestor); + if (parent === ancestor) return yield* fail("invalid-path"); + ancestor = parent; + } + if ( + (yield* fs.realPath(ancestor).pipe(Effect.mapError(() => fail("invalid-path")))) !== ancestor + ) { + return yield* fail("invalid-path"); + } + yield* fs + .makeDirectory(path.dirname(output), { recursive: true }) + .pipe(Effect.mapError(() => fail("write-failed"))); + const file = yield* fs + .open(output, { flag: "wx", mode: 0o600 }) + .pipe( + Effect.mapError((error) => + fail(error.reason._tag === "AlreadyExists" ? "output-exists" : "write-failed"), + ), + ); + const identity = yield* file.stat.pipe(Effect.mapError(() => fail("write-failed"))); + const inode = Option.getOrUndefined(identity.ino); + let expectedSize = 0n; + let poisoned = false; + let closed = false; + const lock = Semaphore.makeUnsafe(1); + yield* Effect.addFinalizer(() => + Effect.uninterruptible( + lock.withPermit( + Effect.sync(() => { + closed = true; + }), + ), + ), + ); + const verifyOwnership = Effect.gen(function* () { + if (poisoned || closed) return yield* fail("write-failed"); + const canonical = yield* fs + .realPath(output) + .pipe(Effect.mapError(() => fail("output-conflict"))); + const current = yield* fs.stat(output).pipe(Effect.mapError(() => fail("output-conflict"))); + const held = yield* file.stat.pipe(Effect.mapError(() => fail("write-failed"))); + for (const info of [current, held]) { + if ( + canonical !== output || + inode === undefined || + info.type !== "File" || + info.dev !== identity.dev || + Option.getOrUndefined(info.ino) !== inode || + Option.getOrUndefined(info.nlink) !== 1 || + info.size !== expectedSize || + (info.mode & 0o777) !== 0o600 + ) { + return yield* fail("output-conflict"); + } + } + }); + const append = (record: string, maxBytes?: number) => { + const bytes = UTF8.encode(`${record}\n`); + if (maxBytes !== undefined && bytes.byteLength > maxBytes) { + return Effect.fail(fail("invalid-record")); + } + const operation = Effect.gen(function* () { + yield* verifyOwnership; + yield* file.writeAll(bytes).pipe(Effect.mapError(() => fail("write-failed"))); + yield* file.sync.pipe(Effect.mapError(() => fail("write-failed"))); + expectedSize += BigInt(bytes.byteLength); + yield* verifyOwnership; + }).pipe( + Effect.onError(() => + Effect.sync(() => { + poisoned = true; + }), + ), + ); + return interruption === "defer" ? Effect.uninterruptible(operation) : operation; + }; + const exclusive = (effect: Effect.Effect) => { + const operation = interruption === "defer" ? Effect.uninterruptible(effect) : effect; + return lock.withPermit(operation); + }; + return { append, exclusive }; + }); + interface TrialState { readonly caseId: string; readonly repetition: number; @@ -272,7 +464,6 @@ export const createLiveEvalJournal = ( FileSystem.FileSystem | Path.Path | Scope.Scope > => Effect.gen(function* () { - const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; if (isUnsafePath(options.outputPath)) return yield* fail("invalid-path"); const output = path.resolve(options.outputPath); @@ -345,92 +536,12 @@ export const createLiveEvalJournal = ( caseIds: plannedCaseIds, repetitions: plannedRepetitions, }).pipe(Effect.mapError(() => fail("invalid-identity"))); - if (yield* fs.exists(output).pipe(Effect.mapError(() => fail("write-failed")))) { - return yield* fail("output-exists"); - } - let ancestor = path.dirname(output); - while (!(yield* fs.exists(ancestor).pipe(Effect.mapError(() => fail("invalid-path"))))) { - const parent = path.dirname(ancestor); - if (parent === ancestor) return yield* fail("invalid-path"); - ancestor = parent; - } - if ( - (yield* fs.realPath(ancestor).pipe(Effect.mapError(() => fail("invalid-path")))) !== ancestor - ) { - return yield* fail("invalid-path"); - } - yield* fs - .makeDirectory(path.dirname(output), { recursive: true }) - .pipe(Effect.mapError(() => fail("write-failed"))); - const file = yield* fs - .open(output, { flag: "wx", mode: 0o600 }) - .pipe( - Effect.mapError((error) => - fail(error.reason._tag === "AlreadyExists" ? "output-exists" : "write-failed"), - ), - ); - const identity = yield* file.stat.pipe(Effect.mapError(() => fail("write-failed"))); - const inode = Option.getOrUndefined(identity.ino); - let expectedSize = 0n; - let poisoned = false; - let closed = false; - const lock = Semaphore.makeUnsafe(1); - yield* Effect.addFinalizer(() => - Effect.uninterruptible( - lock.withPermit( - Effect.sync(() => { - closed = true; - }), - ), - ), - ); - const verifyOwnership = Effect.gen(function* () { - if (poisoned || closed) return yield* fail("write-failed"); - const canonical = yield* fs - .realPath(output) - .pipe(Effect.mapError(() => fail("output-conflict"))); - const current = yield* fs.stat(output).pipe(Effect.mapError(() => fail("output-conflict"))); - const held = yield* file.stat.pipe(Effect.mapError(() => fail("write-failed"))); - for (const info of [current, held]) { - if ( - canonical !== output || - inode === undefined || - info.type !== "File" || - info.dev !== identity.dev || - Option.getOrUndefined(info.ino) !== inode || - Option.getOrUndefined(info.nlink) !== 1 || - info.size !== expectedSize || - (info.mode & 0o777) !== 0o600 - ) { - return yield* fail("output-conflict"); - } - } - }); - const append = (record: string) => - Effect.uninterruptible( - Effect.gen(function* () { - yield* verifyOwnership; - const bytes = UTF8.encode(`${record}\n`); - yield* file.writeAll(bytes).pipe(Effect.mapError(() => fail("write-failed"))); - yield* file.sync.pipe(Effect.mapError(() => fail("write-failed"))); - expectedSize += BigInt(bytes.byteLength); - yield* verifyOwnership; - }).pipe( - Effect.tapError(() => - Effect.sync(() => { - poisoned = true; - }), - ), - ), - ); + const { append, exclusive } = yield* openExclusiveJournalSink(output, "defer"); yield* append(encoded); const trials = new Map(); let bound = false; - const exclusive = (effect: Effect.Effect) => - lock.withPermit(Effect.uninterruptible(effect)); - const requireOpen = (): Effect.Effect => bound ? Effect.fail(fail("already-bound")) : Effect.void; @@ -604,6 +715,193 @@ export const createLiveEvalJournal = ( return { startTrial, generation, finishTrial, bindReport }; }); +export const createOmpTranscriptWriter = ( + options: OmpTranscriptWriterOptions, +): Effect.Effect< + OmpTranscriptWriter, + LiveEvalJournalError, + FileSystem.FileSystem | Path.Path | Scope.Scope +> => + Effect.gen(function* () { + const path = yield* Path.Path; + if (isUnsafePath(options.outputDir) || isUnsafeFileName(options.fileName)) { + return yield* fail("invalid-path"); + } + const outputDir = path.resolve(options.outputDir); + const output = path.resolve(outputDir, options.fileName); + if ( + outputDir !== path.normalize(outputDir) || + path.isAbsolute(options.fileName) || + path.basename(options.fileName) !== options.fileName || + path.dirname(output) !== outputDir + ) { + return yield* fail("invalid-path"); + } + + const runId = yield* admitLabel(options.runId); + const candidate = yield* admitLabel(options.candidate); + const model = yield* decodeModel(options.model).pipe( + Effect.mapError(() => fail("invalid-identity")), + Effect.filterOrFail(isSafePublicEvalText, () => fail("invalid-identity")), + ); + if ( + !isAllowedGinaReadServerUrl(options.serverUrl) || + !isSafePublicEvalText(options.serverUrl) + ) { + return yield* fail("invalid-identity"); + } + const suiteId = yield* admitLabel(options.suiteId); + const suiteVersion = yield* decodePositiveVersion(options.suiteVersion).pipe( + Effect.mapError(() => fail("invalid-identity")), + ); + const fixtureVersion = yield* decodePositiveVersion(options.fixtureVersion).pipe( + Effect.mapError(() => fail("invalid-identity")), + ); + const catalogSha = yield* decodeSha256(options.catalogSha).pipe( + Effect.mapError(() => fail("invalid-identity")), + ); + const reasoning = + options.reasoning === undefined ? undefined : yield* admitLabel(options.reasoning); + const accountClass = yield* admitLabel(options.accountClass); + if ( + !Array.isArray(options.caseIds) || + options.caseIds.length < 1 || + options.caseIds.length > LIVE_EVAL_JOURNAL_MAX_CASES + ) { + return yield* fail("invalid-identity"); + } + const plannedCaseIds: string[] = []; + const plannedCaseIdSet = new Set(); + for (const value of options.caseIds) { + const caseId = yield* admitLabel(value); + if (plannedCaseIdSet.has(caseId)) return yield* fail("invalid-identity"); + plannedCaseIdSet.add(caseId); + plannedCaseIds.push(caseId); + } + if ( + !Number.isSafeInteger(options.repetitions) || + options.repetitions < 1 || + options.repetitions > LIVE_EVAL_JOURNAL_MAX_REPETITIONS + ) { + return yield* fail("invalid-identity"); + } + const plannedRepetitions = options.repetitions; + const expectedDispatchIds: string[] = []; + for (let repetition = 1; repetition <= plannedRepetitions; repetition += 1) { + for (const caseId of plannedCaseIds) { + expectedDispatchIds.push(liveEvalJournalDispatchId(runId, caseId, repetition)); + } + } + const encodedRun = yield* encodeOmpTranscriptRun({ + kind: "run", + schemaVersion: OMP_TRANSCRIPT_SCHEMA_VERSION, + runId, + candidate, + target: "omp_harness", + model, + serverUrl: options.serverUrl, + suiteId, + suiteVersion, + fixtureVersion, + catalogSha, + ...(reasoning === undefined ? {} : { reasoning }), + accountClass, + caseIds: plannedCaseIds, + repetitions: plannedRepetitions, + }).pipe(Effect.mapError(() => fail("invalid-identity"))); + const { append, exclusive } = yield* openExclusiveJournalSink(output, "interrupt"); + yield* append(encodedRun); + + const trials = new Set(); + let bound = false; + + const writeTrial: OmpTranscriptWriter["writeTrial"] = (transcript) => + Effect.gen(function* () { + const admitted = yield* decodeOmpTranscript(transcript).pipe( + Effect.mapError(() => fail("invalid-record")), + ); + const caseId = yield* admitCaseId(admitted.caseId); + if ( + admitted.runId !== runId || + admitted.model !== model || + !Number.isSafeInteger(admitted.repetition) || + admitted.repetition > plannedRepetitions || + !plannedCaseIdSet.has(caseId) + ) { + return yield* fail("invalid-record"); + } + const dispatchId = liveEvalJournalDispatchId(runId, caseId, admitted.repetition); + const encodedTrial = yield* encodeOmpTranscriptTrial({ + kind: "trial", + dispatchId, + transcript: admitted, + }).pipe(Effect.mapError(() => fail("invalid-record"))); + yield* exclusive( + Effect.gen(function* () { + if (bound) return yield* fail("already-bound"); + if (trials.has(dispatchId)) return yield* fail("already-started"); + if (trials.size >= LIVE_EVAL_JOURNAL_MAX_TRIALS) { + return yield* fail("capacity-exceeded"); + } + if (expectedDispatchIds[trials.size] !== dispatchId) { + return yield* fail("invalid-record"); + } + yield* append(encodedTrial, OMP_TRANSCRIPT_MAX_RECORD_BYTES); + trials.add(dispatchId); + }), + ); + }); + + const bindReport: OmpTranscriptWriter["bindReport"] = (reportContent, selectedCaseIds) => + Effect.gen(function* () { + if ( + typeof reportContent !== "string" || + UTF8.encode(reportContent).byteLength > 1_048_576 || + !Array.isArray(selectedCaseIds) || + selectedCaseIds.length !== plannedCaseIds.length || + selectedCaseIds.some((caseId, index) => caseId !== plannedCaseIds[index]) + ) { + return yield* fail("invalid-record"); + } + const report = yield* Schema.decodeEffect( + Schema.fromJsonString(SanitizedEvalRunReportSchema), + DECODE_OPTIONS, + )(reportContent).pipe(Effect.mapError(() => fail("invalid-record"))); + if ( + report.runId !== runId || + report.candidate !== candidate || + report.target !== "omp_harness" || + report.model !== model || + report.accountClass !== accountClass || + report.repetitions !== plannedRepetitions || + (report.reasoning ?? undefined) !== (reasoning ?? undefined) || + report.aggregate.suiteId !== suiteId || + report.aggregate.suiteVersion !== suiteVersion || + report.aggregate.fixtureVersion !== fixtureVersion || + report.aggregate.catalogSha !== catalogSha || + report.aggregate.overall.total !== expectedDispatchIds.length + ) { + return yield* fail("invalid-record"); + } + const encodedBound = yield* encodeOmpTranscriptBound({ + kind: "report-bound", + sourceReportSha256: sha256Hex(reportContent), + }).pipe(Effect.mapError(() => fail("invalid-record"))); + yield* exclusive( + Effect.gen(function* () { + if (bound) return yield* fail("already-bound"); + if (trials.size !== expectedDispatchIds.length) { + return yield* fail("invalid-record"); + } + yield* append(encodedBound); + bound = true; + }), + ); + }); + + return { writeTrial, bindReport }; + }); + const isTypedTimeoutError = (error: unknown): boolean => Cause.isTimeoutError(error) || (typeof error === "object" &&