diff --git a/CHANGESET.md b/CHANGESET.md index 0f8ac4551851..12f04206c563 100644 --- a/CHANGESET.md +++ b/CHANGESET.md @@ -69,6 +69,11 @@ Newest first. Each entry names the upstream range brought in. ### Runtime fixes +- **#201** Log SQLite write-lock holds and lock-stop call sites, diagnostics only (firstmate task + opencode-stop-instrument-180, follows #194): an immediate transaction holding the write lock over 250 ms logs its + pid, purpose and duration; a lock failure leaving the retry gate logs one summary with the statement head, attempts, + elapsed time and native code; and halt's error line names the failed write's event, stream case and call site. No + behaviour changes. _Fork-only._ - **#194** Fix dropped accepted prompts under concurrent SQLite writers (swxtchio/swx-opencode#162, #165): queue writes begin with BEGIN IMMEDIATE, so a write waits on the lock instead of failing on a snapshot pinned by an earlier read, and the busy-retry gate has one owner in effect-sqlite-node/src/retry.ts. Queue writes on the consumption path use the same lock, which is beyond #167's scope. _Fork-only._ - **#163** Revert #155 (swxtchio/swx-opencode#153): under concurrent writers it dropped more accepted prompts than its parent #150, so the busy waits return to the pre-#155 schedule. The fix-forward is tracked in swxtchio/swx-opencode#162. _Fork-only._ - **#144** Make `opencode db path` print the selected path without initializing `AppRuntime` or SQLite, defer global directory creation to runtime initialization, and preserve explicit and channel-specific database path selection. _Upstreamable._ diff --git a/packages/core/src/database/sqlite.bun.ts b/packages/core/src/database/sqlite.bun.ts index 70f646fb4880..614321833bb1 100644 --- a/packages/core/src/database/sqlite.bun.ts +++ b/packages/core/src/database/sqlite.bun.ts @@ -83,7 +83,7 @@ const make = (options: Config) => try: () => (statement.all(...(params as any)) ?? []) as Array>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, query) }) const runValues = (query: string, params: ReadonlyArray = []) => @@ -95,7 +95,7 @@ const make = (options: Config) => try: () => (statement.values(...(params as any)) ?? []) as Array, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, query) }) const connection = identity({ diff --git a/packages/core/src/database/sqlite.node.ts b/packages/core/src/database/sqlite.node.ts index 17ca0b66bd8f..3ddd0d3947e4 100644 --- a/packages/core/src/database/sqlite.node.ts +++ b/packages/core/src/database/sqlite.node.ts @@ -91,7 +91,7 @@ const make = (options: Config) => try: () => statement.all(...(params as SQLInputValue[])) as Array>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, query) }) const runValues = (query: string, params: ReadonlyArray = []) => @@ -103,7 +103,7 @@ const make = (options: Config) => try: () => statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, query) }) const connection = identity({ diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index f0b92fd16b89..7faa855cbeb7 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -5,6 +5,8 @@ import { Event } from "@opencode-ai/schema/event" import type { Data, Definition, Payload } from "@opencode-ai/schema/event" import { and, asc, eq, gt, inArray } from "drizzle-orm" import { Database } from "./database/database" +import { TransactionPurpose } from "@opencode-ai/effect-drizzle-sqlite" +import { FailureOrigin } from "./observability/failure-origin" import { EventRetentionTable, EventSequenceTable, EventTable } from "./event/sql" import { Location } from "./location" import { makeGlobalNode } from "./effect/app-node" @@ -466,7 +468,11 @@ export const layerWith = (options?: LayerOptions) => }), { behavior: "immediate" }, ) - .pipe(Effect.orDie) + .pipe( + Effect.provideService(TransactionPurpose, definition.type), + Effect.tapCause((cause) => FailureOrigin.record(cause, { event: definition.type })), + Effect.orDie, + ) if (committed) { yield* Effect.forEach( pubsub.durable.get(committed.aggregateID) ?? [], diff --git a/packages/core/src/observability/failure-origin.ts b/packages/core/src/observability/failure-origin.ts new file mode 100644 index 000000000000..8bd8b2a1d387 --- /dev/null +++ b/packages/core/src/observability/failure-origin.ts @@ -0,0 +1,41 @@ +import { Cause, Context, Effect } from "effect" + +// Where a failure came from, for diagnostics only. Keyed by the failure object, so the attribution follows that +// failure through Cause squashing and later work cannot overwrite it; the failure itself is never changed. +export type Origin = { + readonly event?: string + readonly stage?: "handler" | "stream" + readonly case?: string + readonly callSite?: string +} + +const origins = new WeakMap() + +// The innermost recording wins each field. The call site is the stack frame chain the runtime captured where the +// failure was raised, so it names the failing callable even when it is recorded after those scopes have unwound or +// across a stream boundary. +export const record = (cause: Cause.Cause, origin: Omit) => + Effect.sync(() => { + if (Cause.hasInterruptsOnly(cause)) return + const failure = Cause.squash(cause) + if (typeof failure !== "object" || failure === null) return + const callSite = frameChain(cause) + origins.set(failure, { ...origin, ...(callSite && { callSite }), ...origins.get(failure) }) + }) + +export const get = (failure: unknown) => + typeof failure === "object" && failure !== null ? origins.get(failure) : undefined + +function frameChain(cause: Cause.Cause) { + const names: string[] = [] + const captured = Context.getOption(Cause.annotations(cause), Cause.StackTrace) + let frame = captured._tag === "Some" ? captured.value : undefined + while (frame && names.length < 8) { + // Effect.fn records a definition-site frame beside each call frame; the call frames name the path. + if (!frame.name.endsWith(" (definition)")) names.push(frame.name) + frame = frame.parent + } + return names.length ? names.join(" < ") : undefined +} + +export * as FailureOrigin from "./failure-origin" diff --git a/packages/core/test/sqlite-lock-diagnostics.test.ts b/packages/core/test/sqlite-lock-diagnostics.test.ts new file mode 100644 index 000000000000..cf0de7397bc1 --- /dev/null +++ b/packages/core/test/sqlite-lock-diagnostics.test.ts @@ -0,0 +1,557 @@ +import { describe, expect, test } from "bun:test" +import { Database } from "@opencode-ai/core/database/database" +import { TransactionPurpose } from "@opencode-ai/effect-drizzle-sqlite" +import { Cause, Context, Effect, Exit, Fiber, Layer, Logger } from "effect" +import { SqlClient } from "effect/unstable/sql/SqlClient" +import { isSqlError } from "effect/unstable/sql/SqlError" +import { Sqlite } from "../src/database/sqlite" +import { + countStatement, + newStatementAttempts, + patchBunQuery, + type StatementAttempts, +} from "../src/database/sqlite-attempts" +import path from "path" +import { tmpdir } from "./fixture/tmpdir" + +const nodeSqliteDrivers = await import("node:sqlite").then( + async (nodeSqlite) => ({ + core: await import("../src/database/sqlite.node"), + standalone: await import("@opencode-ai/effect-sqlite-node"), + DatabaseSync: nodeSqlite.DatabaseSync, + }), + () => undefined, +) +// Above the lock-hold threshold by a margin that scheduling noise cannot close. +const longHoldMs = 400 +// Outlasts the driver's own retry window, so it only catches a retry that never stops. +const lockBackstop = "10 seconds" + +type LogLine = { readonly text: unknown; readonly fields: Record } + +// Collects what the instrumented code sends to the Effect logger, in place of the file sink. +const captured = (effect: (lines: LogLine[]) => Effect.Effect) => + Effect.suspend(() => { + const lines: LogLine[] = [] + const capture = Logger.make((options) => { + const [text, fields] = Array.isArray(options.message) ? options.message : [options.message] + lines.push({ text, fields: typeof fields === "object" && fields !== null ? fields : {} }) + }) + return effect(lines).pipe(Effect.provideService(Logger.CurrentLoggers, new Set([capture]))) + }) + +const holds = (lines: LogLine[]) => lines.filter((line) => line.text === "sqlite write lock held") +const exhausted = (lines: LogLine[]) => lines.filter((line) => line.text === "sqlite lock retries exhausted") +const unretried = (lines: LogLine[]) => lines.filter((line) => line.text === "sqlite lock failed without retry") + +const run = async (effect: Effect.Effect) => { + const exit = await Effect.runPromiseExit(Effect.scoped(effect)) + if (Exit.isFailure(exit)) throw new Error(Cause.pretty(exit.cause)) + return exit.value +} + +const productionDatabase = (filename: string) => + Layer.build(Database.layerFromPath(filename)).pipe(Effect.map((context) => Context.get(context, Database.Service).db)) + +// Holds the write lock from a second connection in this process until the returned release runs. +const inProcessHolder = Effect.fn("inProcessHolder")(function* (filename: string) { + const sqlite = yield* Effect.promise(() => import("bun:sqlite")) + const holder = new sqlite.Database(filename) + yield* Effect.addFinalizer(() => + Effect.sync(() => { + if (holder.inTransaction) holder.run("ROLLBACK") + holder.close() + }), + ) + holder.run("BEGIN IMMEDIATE") + return () => { + if (holder.inTransaction) holder.run("COMMIT") + } +}) + +// A holder in another process takes a real file lock, which every SQLite library in this process must respect. +const childHolder = Effect.fn("childHolder")(function* (filename: string) { + const child = Bun.spawn( + [ + "bun", + "-e", + `import { Database } from "bun:sqlite" +const native = new Database(process.env.SQLITE_LOCK_DB) +native.run("BEGIN IMMEDIATE") +console.log("ready") +await new Promise((resolve) => process.stdin.once("data", resolve)) +native.run("COMMIT") +native.close()`, + ], + { env: { ...process.env, SQLITE_LOCK_DB: filename }, stdin: "pipe", stdout: "pipe", stderr: "inherit" }, + ) + yield* Effect.addFinalizer(() => + Effect.promise(async () => { + child.stdin.write("release\n") + child.stdin.end() + await child.exited + }), + ) + yield* Effect.promise(async () => { + const reader = child.stdout.getReader() + const decoder = new TextDecoder() + let output = "" + while (!output.includes("ready\n")) { + const chunk = await reader.read() + if (chunk.done) throw new Error(`SQLite holder exited before ready: ${output}`) + output += decoder.decode(chunk.value) + } + reader.releaseLock() + }) +}) + +const nodeClientWithAttemptCount = (input: { + clientType: "core-node" | "standalone-node" + filename: string + statement: string + attempts: StatementAttempts +}) => + Effect.gen(function* () { + const drivers = nodeSqliteDrivers + if (!drivers) return yield* Effect.die(new Error("node:sqlite tests ran without node:sqlite")) + const counting = (prepare: (sql: string) => object) => + function (this: unknown, sql: string) { + const statement = prepare.call(this, sql) + return sql === input.statement ? countStatement(statement, input.attempts) : statement + } + if (input.clientType === "core-node") { + const context = yield* Layer.build(drivers.core.layer({ filename: input.filename })) + const native = Context.get(context, Sqlite.Native) as InstanceType + Object.defineProperty(native, "prepare", { configurable: true, writable: true, value: counting(native.prepare) }) + return Context.get(context, SqlClient) + } + const context = yield* Layer.build(drivers.standalone.NodeSqliteClient.layer({ filename: input.filename })) + const prototype = drivers.DatabaseSync.prototype + const descriptor = Object.getOwnPropertyDescriptor(prototype, "prepare") + if (!descriptor) return yield* Effect.die(new Error("node:sqlite prepare method was not found")) + Object.defineProperty(prototype, "prepare", { ...descriptor, value: counting(prototype.prepare) }) + yield* Effect.addFinalizer(() => Effect.sync(() => Object.defineProperty(prototype, "prepare", descriptor))) + return Context.get(context, SqlClient) + }) + +describe("SQLite write-lock hold diagnostics", () => { + test("logs an immediate transaction held past the threshold, by commit and by rollback, and not a short one", () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const db = yield* productionDatabase(path.join(tmp.path, "holds.sqlite")) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + + const committedStartedAt = performance.now() + yield* db + .transaction( + (tx) => + tx.run("INSERT INTO hold_test (id) VALUES (1)").pipe(Effect.andThen(Effect.sleep(`${longHoldMs} millis`))), + { behavior: "immediate" }, + ) + .pipe(Effect.provideService(TransactionPurpose, "test.commit")) + const committedElapsedMs = performance.now() - committedStartedAt + + const rolledBack = yield* db + .transaction( + (tx) => + tx.run("INSERT INTO hold_test (id) VALUES (2)").pipe( + Effect.andThen(Effect.sleep(`${longHoldMs} millis`)), + Effect.andThen(Effect.fail("abandoned" as const)), + ), + { behavior: "immediate" }, + ) + .pipe(Effect.provideService(TransactionPurpose, "test.rollback"), Effect.flip) + expect(rolledBack).toBe("abandoned") + + yield* db.transaction((tx) => tx.run("INSERT INTO hold_test (id) VALUES (3)"), { behavior: "immediate" }).pipe( + Effect.provideService(TransactionPurpose, "test.short"), + ) + // A deferred transaction does not take the write lock at BEGIN, so it is not a hold. + yield* db.transaction(() => Effect.sleep(`${longHoldMs} millis`)) + + const logged = holds(lines) + expect(logged.map((line) => [line.fields.purpose, line.fields.outcome])).toEqual([ + ["test.commit", "commit"], + ["test.rollback", "rollback"], + ]) + for (const line of logged) { + expect(line.fields.pid).toBe(process.pid) + expect(line.fields.durationMs).toBeGreaterThanOrEqual(longHoldMs) + } + expect(logged[0].fields.durationMs).toBeLessThanOrEqual(Math.ceil(committedElapsedMs)) + expect(yield* db.values("SELECT id FROM hold_test ORDER BY id")).toEqual([[1], [3]]) + }), + ), + ), 20_000) + + test("names an unlabelled hold by its span and keeps nested savepoints inside the outer hold", () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const db = yield* productionDatabase(path.join(tmp.path, "nested.sqlite")) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + + // The savepoint alone outlasts the threshold, so only its exclusion keeps it from logging a hold of its own. + const stepMs = 150 + yield* db + .transaction( + (tx) => + Effect.gen(function* () { + yield* Effect.sleep(`${stepMs} millis`) + // A nested immediate transaction, as a durable-event commit inside an open one makes, is a savepoint. + yield* db.transaction( + (inner) => + inner + .run("INSERT INTO hold_test (id) VALUES (1)") + .pipe(Effect.andThen(Effect.sleep(`${longHoldMs} millis`))), + { behavior: "immediate" }, + ) + yield* Effect.sleep(`${stepMs} millis`) + }), + { behavior: "immediate" }, + ) + .pipe(Effect.withSpan("test.nestedCaller")) + + const logged = holds(lines) + expect(logged).toHaveLength(1) + expect(logged[0].fields).toMatchObject({ purpose: "test.nestedCaller", span: "test.nestedCaller", outcome: "commit" }) + expect(logged[0].fields.durationMs).toBeGreaterThanOrEqual(stepMs * 2 + longHoldMs) + }), + ), + )) + + test("attributes a failed commit and its rollback to the same hold", () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const db = yield* productionDatabase(path.join(tmp.path, "commit-failure.sqlite")) + yield* db.run("CREATE TABLE parent (id INTEGER PRIMARY KEY)") + yield* db.run( + "CREATE TABLE child (id INTEGER PRIMARY KEY, parent_id INTEGER REFERENCES parent(id) DEFERRABLE INITIALLY DEFERRED)", + ) + + // The deferred foreign key is checked at COMMIT, which fails and is followed by the rollback. + const failure = yield* db + .transaction( + (tx) => + tx + .run("INSERT INTO child (id, parent_id) VALUES (1, 99)") + .pipe(Effect.andThen(Effect.sleep(`${longHoldMs} millis`))), + { behavior: "immediate" }, + ) + .pipe(Effect.provideService(TransactionPurpose, "test.failedCommit"), Effect.flip) + expect(isSqlError(failure) && failure.reason.cause).toMatchObject({ code: "SQLITE_CONSTRAINT_FOREIGNKEY" }) + + const logged = holds(lines) + expect(logged).toHaveLength(1) + expect(logged[0].fields).toMatchObject({ purpose: "test.failedCommit", outcome: "rollback" }) + expect(logged[0].fields.durationMs).toBeGreaterThanOrEqual(longHoldMs) + expect(yield* db.values("SELECT COUNT(*) FROM child")).toEqual([[0]]) + }), + ), + )) + + test("does not count a long wait for the lock as a hold, nor log a statement that recovers", () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, "wait.sqlite") + const attempts = newStatementAttempts() + yield* patchBunQuery((sql) => sql === "begin immediate", attempts) + const db = yield* productionDatabase(filename) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + const release = yield* inProcessHolder(filename) + + const waitStartedAt = performance.now() + const writer = yield* db + .transaction((tx) => tx.run("INSERT INTO hold_test (id) VALUES (1)"), { behavior: "immediate" }) + .pipe(Effect.provideService(TransactionPurpose, "test.waiter"), Effect.forkChild({ startImmediately: true })) + // The BEGIN has met the held lock; keep holding well past the hold threshold before releasing it. + while (attempts.busy.length === 0 || performance.now() - waitStartedAt < longHoldMs) + yield* Effect.sleep("10 millis") + release() + yield* Fiber.join(writer).pipe( + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("writer never finished")) }), + ) + const waitedMs = performance.now() - waitStartedAt + + expect(waitedMs).toBeGreaterThan(longHoldMs) + expect(attempts.busy.length).toBeGreaterThan(0) + expect(attempts.count).toBeGreaterThan(attempts.busy.length) + expect(holds(lines)).toEqual([]) + expect(exhausted(lines)).toEqual([]) + expect(unretried(lines)).toEqual([]) + expect(yield* db.values("SELECT id FROM hold_test")).toEqual([[1]]) + }), + ), + )) +}) + +describe("SQLite lock retry exhaustion diagnostics", () => { + test( + "summarizes an exhausted BEGIN IMMEDIATE once, with its observed attempts and native code, and logs no hold", + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, "begin-exhausted.sqlite") + const attempts = newStatementAttempts() + yield* patchBunQuery((sql) => sql === "begin immediate", attempts) + const db = yield* productionDatabase(filename) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + yield* inProcessHolder(filename) + + const startedAt = performance.now() + yield* db + .transaction((tx) => tx.run("INSERT INTO hold_test (id) VALUES (1)"), { behavior: "immediate" }) + .pipe( + Effect.provideService(TransactionPurpose, "test.exhausted"), + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("retries never stopped")) }), + Effect.flip, + ) + const elapsedMs = performance.now() - startedAt + + const summaries = exhausted(lines) + expect(summaries).toHaveLength(1) + expect(summaries[0].fields).toMatchObject({ + pid: process.pid, + statement: "begin immediate", + attempts: attempts.count, + "sqlite.code": "SQLITE_BUSY", + "sqlite.errcode": 5, + }) + expect(attempts.count).toBeGreaterThan(1) + expect(summaries[0].fields.elapsedMs).toBeGreaterThanOrEqual(5_000) + expect(summaries[0].fields.elapsedMs).toBeLessThanOrEqual(Math.ceil(elapsedMs)) + expect(holds(lines)).toEqual([]) + expect(unretried(lines)).toEqual([]) + }), + ), + ), + 20_000, + ) + + test( + "keeps attempt counts local to each execution of one reused statement effect", + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, "reuse.sqlite") + const writerSql = "INSERT INTO hold_test (id) VALUES (1)" + const attempts = newStatementAttempts() + yield* patchBunQuery((sql) => sql === writerSql, attempts) + const db = yield* productionDatabase(filename) + yield* db.run("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)") + yield* inProcessHolder(filename) + + // One statement effect, run twice at once; the connection serializes them, so each meets its own window. + const write = db.$client.unsafe(writerSql).raw.pipe(Effect.flip) + yield* Effect.all([write, write], { concurrency: 2 }).pipe( + Effect.timeoutOrElse({ duration: "20 seconds", orElse: () => Effect.die(new Error("retries never stopped")) }), + ) + + const summaries = exhausted(lines) + expect(summaries).toHaveLength(2) + const counts = summaries.map((line) => line.fields.attempts as number) + expect(counts.every((count) => count > 1)).toBe(true) + expect(counts[0] + counts[1]).toBe(attempts.count) + for (const line of summaries) { + expect(line.fields.statement).toBe("insert") + expect(line.fields.elapsedMs).toBeLessThan(10_000) + } + }), + ), + ), + 30_000, + ) + + test( + "names an exhausted statement by its operation, never its comments, literals, identifiers or parameters", + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, "redaction.sqlite") + const setup = yield* productionDatabase(filename) + yield* setup.run("CREATE TABLE redact_test (id INTEGER PRIMARY KEY, v TEXT)") + yield* setup.run(`CREATE TABLE "secret_identifier" (id INTEGER PRIMARY KEY)`) + yield* setup.run("INSERT INTO redact_test (id, v) VALUES (1, NULL)") + const statements = [ + { sql: "/* secret_comment */ INSERT INTO redact_test (id) VALUES (10)", params: [], operation: "insert" }, + { sql: "-- secret_line_comment\nINSERT INTO redact_test (id) VALUES (11)", params: [], operation: "insert" }, + { + sql: "INSERT INTO redact_test (id, v) VALUES (12, 'secret_literal')", + params: [], + operation: "insert", + }, + { sql: `INSERT INTO "secret_identifier" (id) VALUES (13)`, params: [], operation: "insert" }, + { sql: "\n\t bEgIn\r\n ImMeDiAtE ", params: [], operation: "begin immediate" }, + { + sql: "UPDATE redact_test SET v = 'secret_multi' WHERE id = 1; DELETE FROM redact_test", + params: [], + operation: "update", + }, + { + sql: "INSERT INTO redact_test (id, v) VALUES (?, ?)", + params: [14, "secret_parameter"], + operation: "insert", + }, + ] + // One connection per statement, so every statement meets the held lock at once. + const clients = yield* Effect.forEach(statements, () => + productionDatabase(filename).pipe(Effect.map((db) => db.$client)), + ) + yield* inProcessHolder(filename) + + const failures = yield* Effect.forEach( + statements, + (statement, index) => clients[index].unsafe(statement.sql, statement.params).raw.pipe(Effect.flip), + { concurrency: "unbounded" }, + ).pipe( + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("retries never stopped")) }), + ) + + // The returned errors keep their existing lock classification and message. + for (const failure of failures) { + expect(isSqlError(failure) && failure.reason._tag).toBe("LockTimeoutError") + expect(failure.message).toBe("Failed to execute statement: database is locked (SQLITE_BUSY)") + } + const summaries = exhausted(lines) + expect(summaries.map((line) => line.fields.statement).toSorted()).toEqual( + statements.map((statement) => statement.operation).toSorted(), + ) + expect(JSON.stringify(summaries)).not.toContain("secret") + }), + ), + ), + 20_000, + ) + + for (const clientType of ["core-node", "standalone-node"] as const) { + test.skipIf(!nodeSqliteDrivers)( + `${clientType} summarizes an exhausted statement once with its full native result code`, + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, `${clientType}-exhausted.sqlite`) + const writerSql = "INSERT INTO hold_test (id) VALUES (1)" + const attempts = newStatementAttempts() + const client = yield* nodeClientWithAttemptCount({ clientType, filename, statement: writerSql, attempts }) + yield* client.unsafe("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)").raw + yield* childHolder(filename) + + yield* client.unsafe(writerSql).raw.pipe( + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("retries never stopped")) }), + Effect.flip, + ) + + const summaries = exhausted(lines) + expect(summaries).toHaveLength(1) + expect(summaries[0].fields).toMatchObject({ + statement: "insert", + attempts: attempts.count, + "sqlite.code": "ERR_SQLITE_ERROR", + "sqlite.errcode": 5, + }) + expect(attempts.count).toBeGreaterThan(1) + expect(unretried(lines)).toEqual([]) + }), + ), + ), + 20_000, + ) + } + + for (const clientType of ["bun", "core-node", "standalone-node"] as const) { + test.skipIf(clientType !== "bun" && !nodeSqliteDrivers)( + `${clientType} logs a stale-snapshot write once by its own native code, apart from plain SQLITE_BUSY`, + () => + run( + captured((lines) => + Effect.gen(function* () { + const tmp = yield* Effect.acquireRelease( + Effect.promise(() => tmpdir()), + (tmp) => Effect.promise(() => tmp[Symbol.asyncDispose]()), + ) + const filename = path.join(tmp.path, `${clientType}-snapshot.sqlite`) + const writerSql = "INSERT INTO hold_test (id) VALUES (1)" + const attempts = newStatementAttempts() + const client: SqlClient = + clientType === "bun" + ? yield* patchBunQuery((sql) => sql === writerSql, attempts).pipe( + Effect.andThen(productionDatabase(filename)), + Effect.map((db) => db.$client), + ) + : yield* nodeClientWithAttemptCount({ clientType, filename, statement: writerSql, attempts }) + yield* client.unsafe("CREATE TABLE hold_test (id INTEGER PRIMARY KEY)").raw + yield* client.unsafe("PRAGMA journal_mode = WAL").raw + const sqlite = yield* Effect.promise(() => import("bun:sqlite")) + const other = new sqlite.Database(filename) + yield* Effect.addFinalizer(() => Effect.sync(() => other.close())) + + yield* client + .withTransaction( + Effect.gen(function* () { + yield* client.unsafe("SELECT COUNT(*) FROM hold_test").values + // Another connection commits after this transaction's read, so its snapshot is stale. + other.run("INSERT INTO hold_test (id) VALUES (2)") + yield* client.unsafe(writerSql).raw + }), + ) + .pipe( + Effect.timeoutOrElse({ duration: lockBackstop, orElse: () => Effect.die(new Error("retries never stopped")) }), + Effect.flip, + ) + + const summaries = unretried(lines) + expect(summaries).toHaveLength(1) + expect(summaries[0].fields).toMatchObject({ statement: "insert", attempts: 1 }) + expect(attempts.count).toBe(1) + expect(summaries[0].fields).toMatchObject( + clientType === "bun" + ? { "sqlite.code": "SQLITE_BUSY_SNAPSHOT", "sqlite.errcode": 517 } + : { "sqlite.code": "ERR_SQLITE_ERROR", "sqlite.errcode": 517 }, + ) + expect(exhausted(lines)).toEqual([]) + }), + ), + ), + 20_000, + ) + } +}) diff --git a/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts b/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts index cba4c11bd424..36449c53fa8e 100644 --- a/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts +++ b/packages/effect-drizzle-sqlite/src/effect-sqlite/session.ts @@ -3,6 +3,7 @@ import * as Context from "effect/Context" import * as Effect from "effect/Effect" import * as Exit from "effect/Exit" import * as Scope from "effect/Scope" +import * as Tracer from "effect/Tracer" import type { SqlClient } from "effect/unstable/sql/SqlClient" import type { SqlError } from "effect/unstable/sql/SqlError" import type { EffectCacheShape } from "drizzle-orm/cache/core/cache-effect" @@ -26,6 +27,15 @@ export interface EffectSQLiteQueryEffectHKT extends QueryEffectHKTBase { export type EffectSQLiteRunResult = readonly never[] +// Names what a write transaction is for in its lock-hold diagnostic. Unset, the diagnostic names the current span. +export const TransactionPurpose = Context.Reference( + "@opencode-ai/effect-drizzle-sqlite/TransactionPurpose", + { defaultValue: () => undefined }, +) + +// A write lock held longer than this is logged as a likely cause of another connection's lock wait. +const longWriteLockHoldMs = 250 + export interface EffectSQLiteSessionOptions { logger: EffectLoggerShape cache: EffectCacheShape @@ -137,15 +147,34 @@ export class EffectSQLiteSession extends SQLite ), ) const id = connectionOption._tag === "Some" ? connectionOption.value[1] + 1 : 0 + const behavior = config?.behavior ?? "deferred" return connection.pipe( Effect.flatMap(([scope, connection]) => { const transaction = this.executeTransactionStatement( connection, - id === 0 ? `begin ${config?.behavior ?? "deferred"}` : `savepoint effect_sql_${id}`, + id === 0 ? `begin ${behavior}` : `savepoint effect_sql_${id}`, ).pipe( - Effect.flatMap(() => - Effect.provideContext( + Effect.flatMap(() => { + // A top-level immediate or exclusive BEGIN owns the write lock from here until COMMIT or ROLLBACK + // succeeds; time spent waiting for the lock is the retry gate's, not this hold's. Only those top-level + // statements report a release, so a savepoint neither starts nor ends a hold. + const acquiredAt = performance.now() + const released = (outcome: "commit" | "rollback") => + Effect.suspend(() => { + const heldMs = performance.now() - acquiredAt + if (behavior === "deferred" || heldMs <= longWriteLockHoldMs) return Effect.void + const parent = Context.getOption(services, Tracer.ParentSpan) + const span = parent._tag === "Some" && parent.value._tag === "Span" ? parent.value.name : undefined + return Effect.logWarning("sqlite write lock held", { + pid: process.pid, + purpose: Context.get(services, TransactionPurpose) ?? span ?? "unknown", + ...(span && { span }), + outcome, + durationMs: Math.round(heldMs), + }) + }) + return Effect.provideContext( restore(effect), Context.add(services, this.client.transactionService, [connection, id]), ).pipe( @@ -154,9 +183,11 @@ export class EffectSQLiteSession extends SQLite const finalize = Exit.isSuccess(exit) ? id === 0 ? this.executeTransactionStatement(connection, "commit").pipe( + Effect.tap(() => released("commit")), // SQLite keeps the transaction open after deferred constraint commit failures. Effect.catch((error) => this.executeTransactionStatement(connection, "rollback").pipe( + Effect.tap(() => released("rollback")), Effect.catch(() => Effect.void), Effect.andThen(Effect.fail(error)), ), @@ -164,7 +195,9 @@ export class EffectSQLiteSession extends SQLite ) : this.executeTransactionStatement(connection, `release savepoint effect_sql_${id}`) : id === 0 - ? this.executeTransactionStatement(connection, "rollback") + ? this.executeTransactionStatement(connection, "rollback").pipe( + Effect.tap(() => released("rollback")), + ) : this.executeTransactionStatement(connection, `rollback to savepoint effect_sql_${id}`).pipe( Effect.andThen( this.executeTransactionStatement(connection, `release savepoint effect_sql_${id}`), @@ -173,8 +206,8 @@ export class EffectSQLiteSession extends SQLite return finalize.pipe(Effect.flatMap(() => exit)) }), - ), - ), + ) + }), ) return scope === undefined diff --git a/packages/effect-sqlite-node/src/index.ts b/packages/effect-sqlite-node/src/index.ts index 937332af8776..339125972f6c 100644 --- a/packages/effect-sqlite-node/src/index.ts +++ b/packages/effect-sqlite-node/src/index.ts @@ -107,7 +107,7 @@ export const make = ( try: () => statement.all(...(params as SQLInputValue[])) as Array>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, sql) }) const runValues = (sql: string, params: ReadonlyArray = []) => @@ -120,7 +120,7 @@ export const make = ( statement.all(...(params as SQLInputValue[])) as unknown as ReadonlyArray>, catch: statementError, }) - return retryLocked(execute) + return retryLocked(execute, sql) }) return identity({ diff --git a/packages/effect-sqlite-node/src/retry.ts b/packages/effect-sqlite-node/src/retry.ts index f42f9be841a2..01d40ef1575b 100644 --- a/packages/effect-sqlite-node/src/retry.ts +++ b/packages/effect-sqlite-node/src/retry.ts @@ -22,11 +22,44 @@ const isStaleSnapshot = (cause: unknown) => (("code" in cause && cause.code === "SQLITE_BUSY_SNAPSHOT") || ("errcode" in cause && cause.errcode === sqliteBusySnapshot)) -// Every SQLite client routes its statements through this one gate. -export const retryLocked = (execute: Effect.Effect) => { - const retryable = (error: SqlError) => error.reason.isRetryable && !isStaleSnapshot(error.reason.cause) - // Only a statement that met the lock builds the retry schedule, which re-attempts it at once before backing off. - return execute.pipe( - Effect.catchIf(retryable, () => Effect.retry(execute, { schedule: retrySchedule, while: retryable })), - ) +// Every SQLite client routes its statements through this one gate. A lock failure that leaves it logs one summary, +// never one line per attempt, so a contended statement cannot flood the log. +export const retryLocked = (execute: Effect.Effect, sql: string) => + Effect.suspend(() => { + const startedAt = performance.now() + let attempts = 0 + const attempt = Effect.suspend(() => { + attempts++ + return execute + }) + const retryable = (error: SqlError) => error.reason.isRetryable && !isStaleSnapshot(error.reason.cause) + // Only a statement that met the lock builds the retry schedule, which re-attempts it at once before backing off. + return attempt.pipe( + Effect.catchIf(retryable, () => Effect.retry(attempt, { schedule: retrySchedule, while: retryable })), + Effect.tapError((error) => { + if (error.reason._tag !== "LockTimeoutError") return Effect.void + const cause: { code?: unknown; errcode?: unknown; errno?: unknown } = + typeof error.reason.cause === "object" && error.reason.cause !== null ? error.reason.cause : {} + return Effect.logWarning(retryable(error) ? "sqlite lock retries exhausted" : "sqlite lock failed without retry", { + pid: process.pid, + statement: statementOperation(sql), + attempts, + elapsedMs: Math.round(performance.now() - startedAt), + // node:sqlite's code is generic, so its full extended result code tells BUSY from BUSY_SNAPSHOT. + "sqlite.code": cause.code, + "sqlite.errcode": cause.errcode ?? cause.errno, + }) + }), + ) + }) + +// Names a statement by its operation alone: the leading keyword, and a BEGIN's transaction behaviour. Leading comments +// are skipped and nothing after the operation is read, so comments, literals, identifiers and parameters never reach +// the log. +const statementOperation = (sql: string) => { + const statement = sql.replace(/^(?:\s|--[^\n]*|\/\*[\s\S]*?\*\/)*/, "") + const operation = statement.match(/^[a-z]+\b/i)?.[0].toLowerCase() + if (operation !== "begin") return operation ?? "unknown" + const behavior = statement.match(/^begin\s+(deferred|immediate|exclusive)\b/i)?.[1] + return behavior ? `begin ${behavior.toLowerCase()}` : "begin" } diff --git a/packages/opencode/src/session/processor.ts b/packages/opencode/src/session/processor.ts index 23972f669fdd..ac6aa7ee2726 100644 --- a/packages/opencode/src/session/processor.ts +++ b/packages/opencode/src/session/processor.ts @@ -23,6 +23,7 @@ import { ModelV2 } from "@opencode-ai/core/model" import { Question } from "@/question" import { errorMessage } from "@/util/error" import { sqliteLockMessage } from "@/util/sqlite-error" +import { FailureOrigin } from "@opencode-ai/core/observability/failure-origin" import { isRecord } from "@/util/record" import { EventV2Bridge } from "@/event-v2-bridge" import { Database } from "@opencode-ai/core/database/database" @@ -837,11 +838,14 @@ const layer = Layer.effect( }) const halt = Effect.fn("SessionProcessor.halt")(function* (e: unknown) { + // Names the failed operation's event, stream case and call site, so a stop is traceable past its stack. + const origin = FailureOrigin.get(e) yield* Effect.logError("process", { "session.id": input.sessionID, messageID: input.assistantMessage.id, error: errorMessage(e), stack: e instanceof Error ? e.stack : undefined, + ...(origin && { failure: origin }), }) if (ctx.assistantMessage.error) { yield* status.set(ctx.sessionID, { type: "idle" }) @@ -884,10 +888,16 @@ const layer = Layer.effect( type: "busy", activeAssistantMessageID: ctx.assistantMessage.id, }) - const stream = llm.stream(streamInput) + const stream = llm.stream(streamInput).pipe( + Stream.tapCause((cause) => FailureOrigin.record(cause, { stage: "stream" })), + ) yield* stream.pipe( - Stream.tap((event) => handleEvent(event)), + Stream.tap((event) => + handleEvent(event).pipe( + Effect.tapCause((cause) => FailureOrigin.record(cause, { stage: "handler", case: event.type })), + ), + ), Stream.takeUntil(() => ctx.needsCompaction), Stream.runDrain, ) diff --git a/packages/opencode/test/session/processor-effect.test.ts b/packages/opencode/test/session/processor-effect.test.ts index 7e49dbf58757..9d8111fa15c3 100644 --- a/packages/opencode/test/session/processor-effect.test.ts +++ b/packages/opencode/test/session/processor-effect.test.ts @@ -419,6 +419,7 @@ const sqliteTerminalDbPaths = ([2, 3] as const).map((failures) => ({ path: path.join(import.meta.dir, `.opencode-sqlite-terminal-${failures}-${crypto.randomUUID()}.db`), })) const heldLockDbPath = path.join(import.meta.dir, `.opencode-sqlite-held-lock-${crypto.randomUUID()}.db`) +const streamLockDbPath = path.join(import.meta.dir, `.opencode-sqlite-stream-lock-${crypto.randomUUID()}.db`) const routedEnv = LayerNode.compile(root, [ ...replacements, [LLM.node, routedLLM], @@ -445,6 +446,9 @@ afterAll(async () => { heldLockDbPath, `${heldLockDbPath}-wal`, `${heldLockDbPath}-shm`, + streamLockDbPath, + `${streamLockDbPath}-wal`, + `${streamLockDbPath}-shm`, routerLabelDbPath, `${routerLabelDbPath}-wal`, `${routerLabelDbPath}-shm`, @@ -552,6 +556,8 @@ const lockTerminalLLM = Layer.succeed( ) type HeldLock = { messageID: MessageID + // The write the second connection takes the lock just before: the reasoning-end part or the step-finish message. + trigger: "reasoning-end" | "step-finish" holder?: Sqlite streaming: ReturnType> resume: ReturnType> @@ -588,14 +594,26 @@ const heldLockSession = LayerNode.make({ Effect.suspend(() => { const lock = lockFor(part.sessionID, part.messageID) // A second connection takes the write lock just before the turn's reasoning-end write. - if (lock && !lock.holder && part.type === "reasoning" && part.time.end !== undefined) { + if ( + lock?.trigger === "reasoning-end" && + !lock.holder && + part.type === "reasoning" && + part.time.end !== undefined + ) { lock.holder = new Sqlite(heldLockDbPath) lock.holder.run("BEGIN IMMEDIATE") } return heldLockWrite(lock, real.updatePart(part)) }), updateMessage: (msg: T) => - Effect.suspend(() => heldLockWrite(lockFor(msg.sessionID, msg.id), real.updateMessage(msg))), + Effect.suspend(() => { + const lock = lockFor(msg.sessionID, msg.id) + if (lock?.trigger === "step-finish" && !lock.holder && msg.role === "assistant" && msg.finish) { + lock.holder = new Sqlite(heldLockDbPath) + lock.holder.run("BEGIN IMMEDIATE") + } + return heldLockWrite(lock, real.updateMessage(msg)) + }), updateExistingMessage: (msg: SessionV1.Info) => Effect.suspend(() => { const lock = lockFor(msg.sessionID, msg.id) @@ -659,6 +677,35 @@ const sqliteLockEnv = LayerNode.compile(root, [ [Database.node, Database.layerFromPath(sqliteLockDbPath)], ]) const itSqliteLock = testEffect(sqliteLockEnv) +// The provider stream dies with the session's lock failure after its handled events all succeed. +const streamLockFailures = new Map() +// A named callable inside the provider stream, so the halt line must name it rather than the processor around it. +const failProviderStream = Effect.fn("test.failProviderStream")(function* (sessionID: SessionID) { + return yield* Effect.die(streamLockFailures.get(sessionID)) +}) +const itStreamLock = testEffect( + LayerNode.compile(root, [ + ...replacements, + [ + LLM.node, + Layer.succeed( + LLM.Service, + LLM.Service.of({ + stream: (input) => + Stream.concat( + Stream.make( + LLMEvent.stepStart({ index: 0 }), + LLMEvent.reasoningStart({ id: "reasoning-1" }), + LLMEvent.reasoningEnd({ id: "reasoning-1" }), + ), + Stream.unwrap(failProviderStream(input.sessionID as SessionID)), + ), + }), + ), + ], + [Database.node, Database.layerFromPath(streamLockDbPath)], + ]), +) const boot = Effect.fn("test.boot")(function* () { const processors = yield* SessionProcessor.Service @@ -1669,7 +1716,10 @@ const itHeldLock = testEffect( ) // Runs a turn whose reasoning-end write takes a real write lock that stays held after the processor exits. -const failTurnUnderHeldLock = Effect.fn("test.failTurnUnderHeldLock")(function* (dir: string) { +const failTurnUnderHeldLock = Effect.fn("test.failTurnUnderHeldLock")(function* ( + dir: string, + trigger: HeldLock["trigger"] = "reasoning-end", +) { const { processors, session, provider } = yield* boot() const events = yield* EventV2Bridge.Service const statuses = yield* SessionStatus.Service @@ -1677,7 +1727,7 @@ const failTurnUnderHeldLock = Effect.fn("test.failTurnUnderHeldLock")(function* const parent = yield* user(chat.id, "hold the write lock") const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) const model = yield* provider.getModel(ref.providerID, ref.modelID) - const lock: HeldLock = { messageID: msg.id, streaming: defer(), resume: defer(), replayed: 0 } + const lock: HeldLock = { messageID: msg.id, trigger, streaming: defer(), resume: defer(), replayed: 0 } heldLocks.set(chat.id, lock) yield* Effect.addFinalizer(() => Effect.sync(() => { @@ -1861,6 +1911,121 @@ itHeldLock.live( 30_000, ) +// Captures halt's error line, which carries the failed operation's recorded origin. +const haltLines = () => { + const lines: Record[] = [] + const logger = Logger.make((options) => { + const [text, fields] = Array.isArray(options.message) ? options.message : [options.message] + if (options.logLevel === "Error" && text === "process" && typeof fields === "object" && fields !== null) + lines.push(fields) + }) + return { lines, logger } +} + +for (const scenario of [ + { + trigger: "reasoning-end", + event: SessionV1.Event.PartUpdated.type, + callSite: ["Session.updatePart", "SessionProcessor.finishReasoning"], + }, + { trigger: "step-finish", event: SessionV1.Event.MessageUpdated.type, callSite: ["Session.updateMessage"] }, +] as const) { + itHeldLock.live( + `processor halt names the ${scenario.trigger} write that met the held lock`, + () => + provideTmpdirInstance( + (dir) => + Effect.gen(function* () { + const halt = haltLines() + const loggers = yield* Effect.service(Logger.CurrentLoggers) + const { lock, exit, terminal } = yield* failTurnUnderHeldLock(dir, scenario.trigger).pipe( + Effect.provideService(Logger.CurrentLoggers, new Set([...loggers, halt.logger])), + ) + expect(lock.produced && sqliteLockMessage(lock.produced)).toBe("Database is locked (SQLITE_BUSY)") + + expect(halt.lines).toHaveLength(1) + expect(String(halt.lines[0].error).toLowerCase()).toContain("database is locked (sqlite_busy)") + const failure = halt.lines[0].failure as Record | undefined + expect(failure).toMatchObject({ event: scenario.event, stage: "handler", case: scenario.trigger }) + // The span chain runs from the failed commit outward, through the helper the case called. + const frames = String(failure?.callSite).split(" < ") + expect(frames[0]).toBe(scenario.callSite[0]) + const positions = scenario.callSite.map((name) => frames.indexOf(name)) + expect(positions.every((position) => position >= 0)).toBe(true) + expect(positions).toEqual(positions.toSorted((a, b) => a - b)) + + // The diagnostics leave the stop, its idle state and the eventual terminal write as they were. + expect(Exit.isSuccess(exit) && exit.value).toBe("stop") + lock.holder?.run("ROLLBACK") + yield* Deferred.await(terminal).pipe( + Effect.timeoutOrElse({ + duration: "10 seconds", + orElse: () => Effect.die(new Error("the failed turn was not persisted after the lock was released")), + }), + ) + }), + { config: cfg }, + ), + 40_000, + ) +} + +itStreamLock.live( + "processor halt attributes a provider stream lock failure to the stream, not the last handled case", + () => + provideTmpdirInstance( + (dir) => + Effect.gen(function* () { + const { processors, session, provider } = yield* boot() + const database = yield* Database.Service + const chat = yield* session.create({}) + const parent = yield* user(chat.id, "stream lock origin") + const msg = yield* assistant(chat.id, parent.id, path.resolve(dir)) + const model = yield* provider.getModel(ref.providerID, ref.modelID) + const error = yield* produceSqliteBusyError(database.db, streamLockDbPath) + streamLockFailures.set(chat.id, error) + yield* Effect.addFinalizer(() => Effect.sync(() => streamLockFailures.delete(chat.id))) + + const halt = haltLines() + const loggers = yield* Effect.service(Logger.CurrentLoggers) + const handle = yield* processors.create({ assistantMessage: msg, sessionID: chat.id, model }) + const result = yield* handle + .process({ + user: { + id: parent.id, + sessionID: chat.id, + role: "user", + time: parent.time, + agent: parent.agent, + model: { providerID: ref.providerID, modelID: ref.modelID }, + } satisfies SessionV1.User, + sessionID: chat.id, + model, + agent: agent(), + system: [], + messages: [{ role: "user", content: "stream lock origin" }], + tools: {}, + }) + .pipe(Effect.provideService(Logger.CurrentLoggers, new Set([...loggers, halt.logger]))) + + expect(result).toBe("stop") + expect(halt.lines).toHaveLength(1) + expect(String(halt.lines[0].error).toLowerCase()).toContain("database is locked (sqlite_busy)") + const failure = halt.lines[0].failure as Record | undefined + expect(failure).toMatchObject({ stage: "stream" }) + // The call site is the failing callable inside the stream, not the processor that observed the failure. + expect(String(failure?.callSite).split(" < ")[0]).toBe("test.failProviderStream") + expect(failure).not.toHaveProperty("event") + expect(failure).not.toHaveProperty("case") + // The handled reasoning case wrote its part before the stream failed. + const stored = yield* MessageV2.get({ sessionID: chat.id, messageID: msg.id }) + expect(stored.parts.some((part) => part.type === "reasoning")).toBe(true) + }), + { config: cfg }, + ), + 30_000, +) + itPreludeInterrupt.live("session.processor classifies a prelude interrupt as an aborted assistant", () => provideTmpdirInstance( (dir) =>