From 9abb8227ab4d42a6f3aba2b6d36b2141a1d8045a Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Thu, 10 Sep 2026 15:59:01 -0700 Subject: [PATCH 1/3] fix(opencode): finish restart-orphaned turns at instance boot Only one opencode process serves a sandbox and a turn's run loop lives in that process, so at instance boot every assistant message without a completion time is provably dead. Finish it the way an abort would (tool parts interrupted, SessionInterruptedError, time.completed), then publish session.error and idle, so clients end the turn instead of spinning until the coordinator's 30-minute sweep. Co-Authored-By: Claude Fable 5.1 --- packages/core/src/v1/session.ts | 1 + packages/opencode/src/project/bootstrap.ts | 16 ++- packages/opencode/src/session/recover.ts | 93 ++++++++++++++ .../opencode/test/session/recover.test.ts | 114 ++++++++++++++++++ packages/schema/src/v1/session.ts | 2 + packages/sdk/js/src/v2/gen/types.gen.ts | 12 ++ packages/sdk/openapi.json | 36 ++++++ 7 files changed, 272 insertions(+), 2 deletions(-) create mode 100644 packages/opencode/src/session/recover.ts create mode 100644 packages/opencode/test/session/recover.test.ts diff --git a/packages/core/src/v1/session.ts b/packages/core/src/v1/session.ts index b4ce31e5f09a..56164774f187 100644 --- a/packages/core/src/v1/session.ts +++ b/packages/core/src/v1/session.ts @@ -66,3 +66,4 @@ export const ContextOverflowError = NamedError.create("ContextOverflowError", { responseBody: Schema.optional(Schema.String), }) export const ContentFilterError = NamedError.create("ContentFilterError", { message: Schema.String }) +export const InterruptedError = NamedError.create("SessionInterruptedError", { message: Schema.String }) diff --git a/packages/opencode/src/project/bootstrap.ts b/packages/opencode/src/project/bootstrap.ts index acf0caed3cdf..13bebe259f64 100644 --- a/packages/opencode/src/project/bootstrap.ts +++ b/packages/opencode/src/project/bootstrap.ts @@ -9,6 +9,7 @@ import { InstanceState } from "@/effect/instance-state" import { ShareNext } from "@/share/share-next" import { Effect, Layer } from "effect" import { Config } from "@/config/config" +import { SessionRecovery } from "@/session/recover" import { Service } from "./bootstrap-service" export { Service } from "./bootstrap-service" @@ -28,6 +29,7 @@ const layer = Layer.effect( const shareNext = yield* ShareNext.Service const snapshot = yield* Snapshot.Service const vcs = yield* Vcs.Service + const recovery = yield* SessionRecovery.Service const run = Effect.gen(function* () { const ctx = yield* InstanceState.context @@ -39,7 +41,7 @@ const layer = Layer.effect( // Each service self-manages its own slow work via Effect.forkScoped against // its per-instance state scope. We just await materialization here. yield* Effect.forEach( - [lsp, shareNext, format, vcs, snapshot, project], + [lsp, shareNext, format, vcs, snapshot, project, recovery], (s) => s.init().pipe(Effect.catchCause((cause) => Effect.logWarning("init failed", { cause }))), { concurrency: "unbounded", discard: true }, ).pipe(Effect.withSpan("InstanceBootstrap.init")) @@ -52,7 +54,17 @@ const layer = Layer.effect( export const node = makeGlobalNode({ service: Service, layer: layer, - deps: [Config.node, Format.node, LSP.node, Plugin.node, Project.node, ShareNext.node, Snapshot.node, Vcs.node], + deps: [ + Config.node, + Format.node, + LSP.node, + Plugin.node, + Project.node, + SessionRecovery.node, + ShareNext.node, + Snapshot.node, + Vcs.node, + ], }) export * as InstanceBootstrap from "./bootstrap" diff --git a/packages/opencode/src/session/recover.ts b/packages/opencode/src/session/recover.ts new file mode 100644 index 000000000000..fad1d909ad02 --- /dev/null +++ b/packages/opencode/src/session/recover.ts @@ -0,0 +1,93 @@ +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { Database } from "@opencode-ai/core/database/database" +import { MessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SessionV1 } from "@opencode-ai/core/v1/session" +import { and, eq, sql } from "drizzle-orm" +import { Context, Effect, Layer } from "effect" +import { InstanceState } from "@/effect/instance-state" +import { EventV2Bridge } from "@/event-v2-bridge" +import { MessageV2 } from "./message-v2" +import { Session } from "./session" +import { SessionStatus } from "./status" + +export interface Interface { + readonly init: () => Effect.Effect +} + +export class Service extends Context.Service()("@opencode/SessionRecovery") {} + +// Only one opencode process serves a sandbox and a turn's run loop lives in that process, so at +// instance boot every assistant message without a completion time is provably dead. Finish it the +// way an abort would, so clients get a terminal session.error instead of a spinner. +const layer = Layer.effect( + Service, + Effect.gen(function* () { + const database = yield* Database.Service + const sessions = yield* Session.Service + const status = yield* SessionStatus.Service + const events = yield* EventV2Bridge.Service + + const state = yield* InstanceState.make( + Effect.fn("SessionRecovery.state")(function* (ctx) { + const rows = yield* database.db + .select({ id: MessageTable.id, sessionID: MessageTable.session_id }) + .from(MessageTable) + .innerJoin(SessionTable, eq(SessionTable.id, MessageTable.session_id)) + .where( + and( + eq(SessionTable.project_id, ctx.project.id), + sql`json_extract(${MessageTable.data}, '$.role') = 'assistant'`, + sql`json_extract(${MessageTable.data}, '$.time.completed') IS NULL`, + ), + ) + .all() + .pipe(Effect.orDie) + + for (const row of rows) { + const { info, parts } = yield* MessageV2.get({ sessionID: row.sessionID, messageID: row.id }).pipe( + Effect.provideService(Database.Service, database), + Effect.orDie, + ) + if (info.role !== "assistant") continue + const end = Date.now() + for (const part of parts) { + if (part.type !== "tool" || part.state.status === "completed" || part.state.status === "error") continue + const metadata = "metadata" in part.state && part.state.metadata ? part.state.metadata : {} + yield* sessions.updatePart({ + ...part, + state: { + ...part.state, + status: "error", + error: "Tool execution interrupted by a restart", + metadata: { ...metadata, interrupted: true }, + time: { start: "time" in part.state ? part.state.time.start : end, end }, + }, + }) + } + // message.updated before session.error: clients that key "no reply yet" off the first + // assistant message must see this one land before the turn is declared failed. + info.error = new SessionV1.InterruptedError({ + message: "The agent restarted before it could finish this turn. Send your message again.", + }).toObject() + info.time.completed = end + yield* sessions.updateMessage(info) + yield* events.publish(Session.Event.Error, { sessionID: info.sessionID, error: info.error }) + yield* status.set(info.sessionID, { type: "idle" }) + yield* Effect.logInfo("recovered interrupted turn", { sessionID: info.sessionID, messageID: info.id }) + } + }), + ) + + return Service.of({ + init: () => InstanceState.get(state).pipe(Effect.asVoid), + }) + }), +) + +export const node = LayerNode.make({ + service: Service, + layer, + deps: [Database.node, Session.node, SessionStatus.node, EventV2Bridge.node, MessageV2.node], +}) + +export * as SessionRecovery from "./recover" diff --git a/packages/opencode/test/session/recover.test.ts b/packages/opencode/test/session/recover.test.ts new file mode 100644 index 000000000000..e028e34dc5c1 --- /dev/null +++ b/packages/opencode/test/session/recover.test.ts @@ -0,0 +1,114 @@ +import { expect } from "bun:test" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { SessionV1 } from "@opencode-ai/core/v1/session" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { ModelV2 } from "@opencode-ai/core/model" +import { Effect } from "effect" +import { EventV2Bridge } from "@/event-v2-bridge" +import { Session } from "@/session/session" +import { SessionStatus } from "@/session/status" +import { SessionRecovery } from "@/session/recover" +import { MessageV2 } from "../../src/session/message-v2" +import { MessageID, PartID, type SessionID } from "../../src/session/schema" +import { testEffect } from "../lib/effect" + +const it = testEffect( + LayerNode.compile( + LayerNode.group([ + Session.node, + MessageV2.node, + SessionProjector.node, + SessionStatus.node, + EventV2Bridge.node, + SessionRecovery.node, + ]), + ), +) + +const seedTurn = Effect.fn("Test.seedTurn")(function* (sessionID: SessionID, opts?: { completed?: number }) { + const sessions = yield* Session.Service + const userID = MessageID.ascending() + yield* sessions.updateMessage({ + id: userID, + sessionID, + role: "user", + time: { created: Date.now() }, + agent: "test", + model: { providerID: "test", modelID: "test" }, + tools: {}, + mode: "", + } as unknown as SessionV1.Info) + const assistantID = MessageID.ascending() + yield* sessions.updateMessage({ + id: assistantID, + sessionID, + role: "assistant", + time: { created: Date.now(), completed: opts?.completed }, + parentID: userID, + modelID: ModelV2.ID.make("test"), + providerID: ProviderV2.ID.make("test"), + mode: "", + agent: "default", + path: { cwd: "/", root: "/" }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + } as unknown as SessionV1.Info) + yield* sessions.updatePart({ + id: PartID.ascending(), + sessionID, + messageID: assistantID, + type: "tool", + callID: "call-1", + tool: "bash", + state: { status: "running", input: { command: "sleep 120" }, time: { start: 1 } }, + } as unknown as SessionV1.Part) + return assistantID +}) + +it.instance("finishes a restart-orphaned assistant message with an interrupted error", () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const events = yield* EventV2Bridge.Service + const recovery = yield* SessionRecovery.Service + const chat = yield* sessions.create({}) + const assistantID = yield* seedTurn(chat.id) + const seen: string[] = [] + const off = yield* events.listen((event) => { + seen.push(event.type) + return Effect.void + }) + + yield* recovery.init() + yield* off + + const stored = yield* MessageV2.get({ sessionID: chat.id, messageID: assistantID }) + expect(stored.info.role).toBe("assistant") + if (stored.info.role === "assistant") { + expect(stored.info.error?.name).toBe("SessionInterruptedError") + expect(stored.info.time.completed).toBeNumber() + } + const tool = stored.parts.find((part) => part.type === "tool") + expect(tool?.type === "tool" && tool.state.status === "error" ? tool.state.metadata?.interrupted : undefined).toBe( + true, + ) + expect(seen.indexOf(MessageV2.Event.Updated.type)).toBeLessThan(seen.indexOf(Session.Event.Error.type)) + expect(seen).toContain(SessionStatus.Event.Idle.type) + }), +) + +it.instance("leaves a completed assistant message alone", () => + Effect.gen(function* () { + const sessions = yield* Session.Service + const recovery = yield* SessionRecovery.Service + const chat = yield* sessions.create({}) + const assistantID = yield* seedTurn(chat.id, { completed: Date.now() }) + + yield* recovery.init() + + const stored = yield* MessageV2.get({ sessionID: chat.id, messageID: assistantID }) + expect(stored.info.role === "assistant" ? stored.info.error : "wrong role").toBeUndefined() + const tool = stored.parts.find((part) => part.type === "tool") + expect(tool?.type === "tool" ? tool.state.status : "missing").toBe("running") + }), +) diff --git a/packages/schema/src/v1/session.ts b/packages/schema/src/v1/session.ts index 75e9282f117c..9583baa154eb 100644 --- a/packages/schema/src/v1/session.ts +++ b/packages/schema/src/v1/session.ts @@ -61,6 +61,7 @@ export const ContextOverflowError = namedError("ContextOverflowError", { export const ContentFilterError = namedError("ContentFilterError", { message: Schema.String, }) +export const InterruptedError = namedError("SessionInterruptedError", { message: Schema.String }) export class OutputFormatText extends Schema.Class("OutputFormatText")({ type: Schema.Literal("text"), @@ -390,6 +391,7 @@ const AssistantErrorSchema = Schema.Union([ StructuredOutputError.EffectSchema, ContextOverflowError.EffectSchema, ContentFilterError.EffectSchema, + InterruptedError.EffectSchema, APIError.EffectSchema, ]).annotate({ discriminator: "name" }) type AssistantError = Schema.Schema.Type diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 5e067f3afb23..2d4e5888b039 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -314,6 +314,13 @@ export type ContentFilterError = { } } +export type SessionInterruptedError = { + name: "SessionInterruptedError" + data: { + message: string + } +} + export type ApiError = { name: "APIError" data: { @@ -346,6 +353,7 @@ export type AssistantMessage = { | StructuredOutputError | ContextOverflowError | ContentFilterError + | SessionInterruptedError | ApiError parentID: string modelID: string @@ -1222,6 +1230,7 @@ export type GlobalEvent = { | StructuredOutputError | ContextOverflowError | ContentFilterError + | SessionInterruptedError | ApiError } } @@ -1746,6 +1755,7 @@ export type ProviderConfig = { baseURL?: string enterpriseUrl?: string setCacheKey?: boolean + promptCacheKey?: string /** * Timeout in milliseconds for full requests to this provider. Set to false to disable timeout. */ @@ -5356,6 +5366,7 @@ export type SessionError = { | StructuredOutputError | ContextOverflowError | ContentFilterError + | SessionInterruptedError | ApiError } } @@ -6681,6 +6692,7 @@ export type EventSessionError = { | StructuredOutputError | ContextOverflowError | ContentFilterError + | SessionInterruptedError | ApiError } } diff --git a/packages/sdk/openapi.json b/packages/sdk/openapi.json index d97885b55390..6522a3f0c1c1 100644 --- a/packages/sdk/openapi.json +++ b/packages/sdk/openapi.json @@ -16187,6 +16187,27 @@ "required": ["name", "data"], "additionalProperties": false }, + "SessionInterruptedError": { + "type": "object", + "properties": { + "name": { + "type": "string", + "enum": ["SessionInterruptedError"] + }, + "data": { + "type": "object", + "properties": { + "message": { + "type": "string" + } + }, + "required": ["message"], + "additionalProperties": false + } + }, + "required": ["name", "data"], + "additionalProperties": false + }, "APIError": { "type": "object", "properties": { @@ -16283,6 +16304,9 @@ { "$ref": "#/components/schemas/ContentFilterError" }, + { + "$ref": "#/components/schemas/SessionInterruptedError" + }, { "$ref": "#/components/schemas/APIError" } @@ -19130,6 +19154,9 @@ { "$ref": "#/components/schemas/ContentFilterError" }, + { + "$ref": "#/components/schemas/SessionInterruptedError" + }, { "$ref": "#/components/schemas/APIError" } @@ -20755,6 +20782,9 @@ "setCacheKey": { "type": "boolean" }, + "promptCacheKey": { + "type": "string" + }, "timeout": { "anyOf": [ { @@ -31628,6 +31658,9 @@ { "$ref": "#/components/schemas/ContentFilterError" }, + { + "$ref": "#/components/schemas/SessionInterruptedError" + }, { "$ref": "#/components/schemas/APIError" } @@ -35509,6 +35542,9 @@ { "$ref": "#/components/schemas/ContentFilterError" }, + { + "$ref": "#/components/schemas/SessionInterruptedError" + }, { "$ref": "#/components/schemas/APIError" } From f3200428362ce5decbe299e327affe1b42d02e28 Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Fri, 11 Sep 2026 13:46:07 -0700 Subject: [PATCH 2/3] Review: reuse NamedError.Unknown, survive a bad row, assert both events Per PierrotAWB: the error name is invisible to every consumer, so drop the schema/SDK surface and use the prompt.ts pattern; catch per-row failures so one unreadable message cannot strand the other orphans; the ordering test now requires both events to be present before comparing their order. Co-Authored-By: Claude Fable 5.1 --- packages/core/src/v1/session.ts | 1 - packages/opencode/src/session/recover.ts | 65 ++++++++++--------- .../opencode/test/session/recover.test.ts | 4 +- packages/schema/src/v1/session.ts | 2 - packages/sdk/js/src/v2/gen/types.gen.ts | 12 ---- packages/sdk/openapi.json | 36 ---------- 6 files changed, 38 insertions(+), 82 deletions(-) diff --git a/packages/core/src/v1/session.ts b/packages/core/src/v1/session.ts index 56164774f187..b4ce31e5f09a 100644 --- a/packages/core/src/v1/session.ts +++ b/packages/core/src/v1/session.ts @@ -66,4 +66,3 @@ export const ContextOverflowError = NamedError.create("ContextOverflowError", { responseBody: Schema.optional(Schema.String), }) export const ContentFilterError = NamedError.create("ContentFilterError", { message: Schema.String }) -export const InterruptedError = NamedError.create("SessionInterruptedError", { message: Schema.String }) diff --git a/packages/opencode/src/session/recover.ts b/packages/opencode/src/session/recover.ts index fad1d909ad02..1f2f3763ebf9 100644 --- a/packages/opencode/src/session/recover.ts +++ b/packages/opencode/src/session/recover.ts @@ -1,12 +1,13 @@ import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { Database } from "@opencode-ai/core/database/database" import { MessageTable, SessionTable } from "@opencode-ai/core/session/sql" -import { SessionV1 } from "@opencode-ai/core/v1/session" +import { NamedError } from "@opencode-ai/core/util/error" import { and, eq, sql } from "drizzle-orm" import { Context, Effect, Layer } from "effect" import { InstanceState } from "@/effect/instance-state" import { EventV2Bridge } from "@/event-v2-bridge" import { MessageV2 } from "./message-v2" +import type { MessageID, SessionID } from "./schema" import { Session } from "./session" import { SessionStatus } from "./status" @@ -44,40 +45,44 @@ const layer = Layer.effect( .pipe(Effect.orDie) for (const row of rows) { - const { info, parts } = yield* MessageV2.get({ sessionID: row.sessionID, messageID: row.id }).pipe( - Effect.provideService(Database.Service, database), - Effect.orDie, + // One unreadable row must not strand every other orphan (REPL-29772 saw session-DB corruption). + yield* recover(row).pipe( + Effect.catchCause((cause) => Effect.logWarning("recovery failed", { messageID: row.id, cause })), ) - if (info.role !== "assistant") continue - const end = Date.now() - for (const part of parts) { - if (part.type !== "tool" || part.state.status === "completed" || part.state.status === "error") continue - const metadata = "metadata" in part.state && part.state.metadata ? part.state.metadata : {} - yield* sessions.updatePart({ - ...part, - state: { - ...part.state, - status: "error", - error: "Tool execution interrupted by a restart", - metadata: { ...metadata, interrupted: true }, - time: { start: "time" in part.state ? part.state.time.start : end, end }, - }, - }) - } - // message.updated before session.error: clients that key "no reply yet" off the first - // assistant message must see this one land before the turn is declared failed. - info.error = new SessionV1.InterruptedError({ - message: "The agent restarted before it could finish this turn. Send your message again.", - }).toObject() - info.time.completed = end - yield* sessions.updateMessage(info) - yield* events.publish(Session.Event.Error, { sessionID: info.sessionID, error: info.error }) - yield* status.set(info.sessionID, { type: "idle" }) - yield* Effect.logInfo("recovered interrupted turn", { sessionID: info.sessionID, messageID: info.id }) } }), ) + const recover = Effect.fn("SessionRecovery.recover")(function* (row: { id: MessageID; sessionID: SessionID }) { + const { info, parts } = yield* MessageV2.get({ sessionID: row.sessionID, messageID: row.id }).pipe( + Effect.provideService(Database.Service, database), + ) + if (info.role !== "assistant") return + const end = Date.now() + for (const part of parts) { + if (part.type !== "tool" || part.state.status === "completed" || part.state.status === "error") continue + const metadata = "metadata" in part.state && part.state.metadata ? part.state.metadata : {} + yield* sessions.updatePart({ + ...part, + state: { + ...part.state, + status: "error", + error: "Tool execution interrupted by a restart", + metadata: { ...metadata, interrupted: true }, + time: { start: "time" in part.state ? part.state.time.start : end, end }, + }, + }) + } + info.error = new NamedError.Unknown({ + message: "The agent restarted before it could finish this turn. Send your message again.", + }).toObject() + info.time.completed = end + yield* sessions.updateMessage(info) + yield* events.publish(Session.Event.Error, { sessionID: info.sessionID, error: info.error }) + yield* status.set(info.sessionID, { type: "idle" }) + yield* Effect.logInfo("recovered interrupted turn", { sessionID: info.sessionID, messageID: info.id }) + }) + return Service.of({ init: () => InstanceState.get(state).pipe(Effect.asVoid), }) diff --git a/packages/opencode/test/session/recover.test.ts b/packages/opencode/test/session/recover.test.ts index e028e34dc5c1..9a280ad5d96e 100644 --- a/packages/opencode/test/session/recover.test.ts +++ b/packages/opencode/test/session/recover.test.ts @@ -85,13 +85,15 @@ it.instance("finishes a restart-orphaned assistant message with an interrupted e const stored = yield* MessageV2.get({ sessionID: chat.id, messageID: assistantID }) expect(stored.info.role).toBe("assistant") if (stored.info.role === "assistant") { - expect(stored.info.error?.name).toBe("SessionInterruptedError") + expect(stored.info.error?.name).toBe("UnknownError") expect(stored.info.time.completed).toBeNumber() } const tool = stored.parts.find((part) => part.type === "tool") expect(tool?.type === "tool" && tool.state.status === "error" ? tool.state.metadata?.interrupted : undefined).toBe( true, ) + expect(seen).toContain(MessageV2.Event.Updated.type) + expect(seen).toContain(Session.Event.Error.type) expect(seen.indexOf(MessageV2.Event.Updated.type)).toBeLessThan(seen.indexOf(Session.Event.Error.type)) expect(seen).toContain(SessionStatus.Event.Idle.type) }), diff --git a/packages/schema/src/v1/session.ts b/packages/schema/src/v1/session.ts index 9583baa154eb..75e9282f117c 100644 --- a/packages/schema/src/v1/session.ts +++ b/packages/schema/src/v1/session.ts @@ -61,7 +61,6 @@ export const ContextOverflowError = namedError("ContextOverflowError", { export const ContentFilterError = namedError("ContentFilterError", { message: Schema.String, }) -export const InterruptedError = namedError("SessionInterruptedError", { message: Schema.String }) export class OutputFormatText extends Schema.Class("OutputFormatText")({ type: Schema.Literal("text"), @@ -391,7 +390,6 @@ const AssistantErrorSchema = Schema.Union([ StructuredOutputError.EffectSchema, ContextOverflowError.EffectSchema, ContentFilterError.EffectSchema, - InterruptedError.EffectSchema, APIError.EffectSchema, ]).annotate({ discriminator: "name" }) type AssistantError = Schema.Schema.Type diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 2d4e5888b039..5e067f3afb23 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -314,13 +314,6 @@ export type ContentFilterError = { } } -export type SessionInterruptedError = { - name: "SessionInterruptedError" - data: { - message: string - } -} - export type ApiError = { name: "APIError" data: { @@ -353,7 +346,6 @@ export type AssistantMessage = { | StructuredOutputError | ContextOverflowError | ContentFilterError - | SessionInterruptedError | ApiError parentID: string modelID: string @@ -1230,7 +1222,6 @@ export type GlobalEvent = { | StructuredOutputError | ContextOverflowError | ContentFilterError - | SessionInterruptedError | ApiError } } @@ -1755,7 +1746,6 @@ export type ProviderConfig = { baseURL?: string enterpriseUrl?: string setCacheKey?: boolean - promptCacheKey?: string /** * Timeout in milliseconds for full requests to this provider. Set to false to disable timeout. */ @@ -5366,7 +5356,6 @@ export type SessionError = { | StructuredOutputError | ContextOverflowError | ContentFilterError - | SessionInterruptedError | ApiError } } @@ -6692,7 +6681,6 @@ export type EventSessionError = { | StructuredOutputError | ContextOverflowError | ContentFilterError - | SessionInterruptedError | ApiError } } diff --git a/packages/sdk/openapi.json b/packages/sdk/openapi.json index 6522a3f0c1c1..d97885b55390 100644 --- a/packages/sdk/openapi.json +++ b/packages/sdk/openapi.json @@ -16187,27 +16187,6 @@ "required": ["name", "data"], "additionalProperties": false }, - "SessionInterruptedError": { - "type": "object", - "properties": { - "name": { - "type": "string", - "enum": ["SessionInterruptedError"] - }, - "data": { - "type": "object", - "properties": { - "message": { - "type": "string" - } - }, - "required": ["message"], - "additionalProperties": false - } - }, - "required": ["name", "data"], - "additionalProperties": false - }, "APIError": { "type": "object", "properties": { @@ -16304,9 +16283,6 @@ { "$ref": "#/components/schemas/ContentFilterError" }, - { - "$ref": "#/components/schemas/SessionInterruptedError" - }, { "$ref": "#/components/schemas/APIError" } @@ -19154,9 +19130,6 @@ { "$ref": "#/components/schemas/ContentFilterError" }, - { - "$ref": "#/components/schemas/SessionInterruptedError" - }, { "$ref": "#/components/schemas/APIError" } @@ -20782,9 +20755,6 @@ "setCacheKey": { "type": "boolean" }, - "promptCacheKey": { - "type": "string" - }, "timeout": { "anyOf": [ { @@ -31658,9 +31628,6 @@ { "$ref": "#/components/schemas/ContentFilterError" }, - { - "$ref": "#/components/schemas/SessionInterruptedError" - }, { "$ref": "#/components/schemas/APIError" } @@ -35542,9 +35509,6 @@ { "$ref": "#/components/schemas/ContentFilterError" }, - { - "$ref": "#/components/schemas/SessionInterruptedError" - }, { "$ref": "#/components/schemas/APIError" } From 6e796db619abb273375b7ee14e84f0e9dcc35820 Mon Sep 17 00:00:00 2001 From: Kevin O'Connell Date: Fri, 11 Sep 2026 14:21:53 -0700 Subject: [PATCH 3/3] Review: repair rows only, let the coordinator derive the terminal event Publishing session.error from the sweep races a client that clears its no-reply marker on the first assistant message.updated; the coordinator's post-restart replay already synthesizes the terminal event from a completed errored message, so the sweep now publishes only the message and part updates. Co-Authored-By: Claude Fable 5.1 --- packages/opencode/src/session/recover.ts | 12 ++++-------- packages/opencode/test/session/recover.test.ts | 8 +++++--- 2 files changed, 9 insertions(+), 11 deletions(-) diff --git a/packages/opencode/src/session/recover.ts b/packages/opencode/src/session/recover.ts index 1f2f3763ebf9..f1bd1286fdf7 100644 --- a/packages/opencode/src/session/recover.ts +++ b/packages/opencode/src/session/recover.ts @@ -5,11 +5,9 @@ import { NamedError } from "@opencode-ai/core/util/error" import { and, eq, sql } from "drizzle-orm" import { Context, Effect, Layer } from "effect" import { InstanceState } from "@/effect/instance-state" -import { EventV2Bridge } from "@/event-v2-bridge" import { MessageV2 } from "./message-v2" import type { MessageID, SessionID } from "./schema" import { Session } from "./session" -import { SessionStatus } from "./status" export interface Interface { readonly init: () => Effect.Effect @@ -19,14 +17,14 @@ export class Service extends Context.Service()("@opencode/Se // Only one opencode process serves a sandbox and a turn's run loop lives in that process, so at // instance boot every assistant message without a completion time is provably dead. Finish it the -// way an abort would, so clients get a terminal session.error instead of a spinner. +// way an abort would. No session.error is published here: a client that keys "no reply yet" off +// the first assistant message can race it, and the coordinator already derives the terminal event +// from a completed errored message when it replays the session after a restart. const layer = Layer.effect( Service, Effect.gen(function* () { const database = yield* Database.Service const sessions = yield* Session.Service - const status = yield* SessionStatus.Service - const events = yield* EventV2Bridge.Service const state = yield* InstanceState.make( Effect.fn("SessionRecovery.state")(function* (ctx) { @@ -78,8 +76,6 @@ const layer = Layer.effect( }).toObject() info.time.completed = end yield* sessions.updateMessage(info) - yield* events.publish(Session.Event.Error, { sessionID: info.sessionID, error: info.error }) - yield* status.set(info.sessionID, { type: "idle" }) yield* Effect.logInfo("recovered interrupted turn", { sessionID: info.sessionID, messageID: info.id }) }) @@ -92,7 +88,7 @@ const layer = Layer.effect( export const node = LayerNode.make({ service: Service, layer, - deps: [Database.node, Session.node, SessionStatus.node, EventV2Bridge.node, MessageV2.node], + deps: [Database.node, Session.node, MessageV2.node], }) export * as SessionRecovery from "./recover" diff --git a/packages/opencode/test/session/recover.test.ts b/packages/opencode/test/session/recover.test.ts index 9a280ad5d96e..b091eebf7858 100644 --- a/packages/opencode/test/session/recover.test.ts +++ b/packages/opencode/test/session/recover.test.ts @@ -87,15 +87,17 @@ it.instance("finishes a restart-orphaned assistant message with an interrupted e if (stored.info.role === "assistant") { expect(stored.info.error?.name).toBe("UnknownError") expect(stored.info.time.completed).toBeNumber() + expect(stored.info.finish).toBeUndefined() } const tool = stored.parts.find((part) => part.type === "tool") expect(tool?.type === "tool" && tool.state.status === "error" ? tool.state.metadata?.interrupted : undefined).toBe( true, ) + expect(seen).toContain(MessageV2.Event.PartUpdated.type) expect(seen).toContain(MessageV2.Event.Updated.type) - expect(seen).toContain(Session.Event.Error.type) - expect(seen.indexOf(MessageV2.Event.Updated.type)).toBeLessThan(seen.indexOf(Session.Event.Error.type)) - expect(seen).toContain(SessionStatus.Event.Idle.type) + // The coordinator derives the terminal event from the repaired row; publishing one here would race its marker. + expect(seen).not.toContain(Session.Event.Error.type) + expect(seen).not.toContain(SessionStatus.Event.Idle.type) }), )