From 59e8dbb066b71e37610f4777fe0b59c15f4ac4a6 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 25 Jun 2026 23:20:44 -0400 Subject: [PATCH] feat(core): publish interrupted session steps --- packages/client/src/generated/types.ts | 8 ++ packages/core/src/session/message-updater.ts | 6 ++ packages/core/src/session/projector.ts | 1 + packages/core/src/session/runner/llm.ts | 4 +- .../src/session/runner/publish-llm-event.ts | 20 ++++- .../core/src/session/runner/to-llm-message.ts | 2 +- .../core/test/session-runner-message.test.ts | 5 +- packages/core/test/session-runner.test.ts | 11 ++- packages/opencode/test/event-manifest.test.ts | 2 +- packages/schema/src/session-event.ts | 12 +++ packages/schema/test/event-manifest.test.ts | 10 +-- packages/sdk/js/src/v2/gen/types.gen.ts | 76 +++++++++++++++++++ 12 files changed, 140 insertions(+), 17 deletions(-) diff --git a/packages/client/src/generated/types.ts b/packages/client/src/generated/types.ts index 858756e2a4..2f3baf7123 100644 --- a/packages/client/src/generated/types.ts +++ b/packages/client/src/generated/types.ts @@ -795,6 +795,14 @@ export type SessionsEventsOutput = readonly error: { readonly type: "unknown"; readonly message: string } } } + | { + readonly id: string + readonly metadata?: { readonly [x: string]: unknown } + readonly type: "session.next.step.interrupted" + readonly durable?: { readonly aggregateID: string; readonly seq: number; readonly version: number } + readonly location?: { readonly directory: string; readonly workspaceID?: string } + readonly data: { readonly timestamp: number; readonly sessionID: string; readonly assistantMessageID: string } + } | { readonly id: string readonly metadata?: { readonly [x: string]: unknown } diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index a97afb58b6..2c92af8583 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -227,6 +227,12 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { draft.error = event.data.error }) }, + "session.next.step.interrupted": (event) => { + return updateOwnedAssistant(event.data.assistantMessageID, (draft) => { + draft.time.completed = event.data.timestamp + draft.finish = "interrupted" + }) + }, "session.next.text.started": (event) => { return updateOwnedAssistant(event.data.assistantMessageID, (draft) => { draft.content.push( diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 4d8b9a05b4..4db0775248 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -381,6 +381,7 @@ export const layer = Layer.effectDiscard( yield* events.project(SessionEvent.Step.Started, (event) => run(db, event)) yield* events.project(SessionEvent.Step.Ended, (event) => run(db, event)) yield* events.project(SessionEvent.Step.Failed, (event) => run(db, event)) + yield* events.project(SessionEvent.Step.Interrupted, (event) => run(db, event)) yield* events.project(SessionEvent.Text.Started, (event) => run(db, event)) yield* events.project(SessionEvent.Text.Ended, (event) => run(db, event)) yield* events.project(SessionEvent.Tool.Input.Started, (event) => run(db, event)) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 9e29ae71ee..3ff580e666 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -290,6 +290,7 @@ export const layer = Layer.effect( if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { yield* FiberSet.clear(toolFibers) yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) + yield* withPublication(publisher.interruptAssistant()) return yield* Effect.interrupt } if ( @@ -298,8 +299,7 @@ export const layer = Layer.effect( ) { yield* FiberSet.clear(toolFibers) yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) - if (publisher.hasActiveAssistant()) - yield* withPublication(publisher.failAssistant("Provider turn interrupted")) + yield* withPublication(publisher.interruptAssistant()) } if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) { const failure = Cause.squash(settled.cause) diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 33652a618c..471950cbff 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -67,7 +67,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) const timestamp = DateTime.now let assistantMessageID: SessionMessage.ID | undefined let assistantActive = false - let assistantFailed = false + let assistantSettled = false let providerFailed = false let stepSettlement: { readonly finish: string; readonly tokens: ReturnType } | undefined @@ -197,11 +197,11 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) }) const failAssistant = Effect.fnUntraced(function* (message: string) { - if (assistantFailed) return + if (assistantSettled) return yield* flush() const assistantMessageID = yield* startAssistant() assistantActive = false - assistantFailed = true + assistantSettled = true yield* events.publish(SessionEvent.Step.Failed, { sessionID: input.sessionID, timestamp: yield* timestamp, @@ -210,6 +210,19 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) }) }) + const interruptAssistant = Effect.fnUntraced(function* () { + if (assistantSettled) return + yield* flush() + const assistantMessageID = yield* startAssistant() + assistantActive = false + assistantSettled = true + yield* events.publish(SessionEvent.Step.Interrupted, { + sessionID: input.sessionID, + timestamp: yield* timestamp, + assistantMessageID, + }) + }) + const failUnsettledTools = Effect.fn("SessionRunner.failUnsettledTools")(function* ( message: string, hostedOnly = false, @@ -412,6 +425,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) publish, flush, failAssistant, + interruptAssistant, failUnsettledTools, hasActiveAssistant: () => assistantActive, hasAssistantStarted: () => assistantMessageID !== undefined, diff --git a/packages/core/src/session/runner/to-llm-message.ts b/packages/core/src/session/runner/to-llm-message.ts index b2b1af5d30..dcc6cd99a7 100644 --- a/packages/core/src/session/runner/to-llm-message.ts +++ b/packages/core/src/session/runner/to-llm-message.ts @@ -70,7 +70,7 @@ const toolResult = (tool: SessionMessage.AssistantTool, providerMetadata: Provid const assistant = (message: SessionMessage.Assistant, model: Model) => { const sameModel = String(message.model.providerID) === String(model.provider) && String(message.model.id) === String(model.id) - const reuseProviderMetadata = sameModel && message.error === undefined + const reuseProviderMetadata = sameModel && message.error === undefined && message.finish !== "interrupted" const content = message.content.flatMap((item): ContentPart[] => { if (item.type === "text") return [{ type: "text", text: item.text }] if (item.type === "reasoning") diff --git a/packages/core/test/session-runner-message.test.ts b/packages/core/test/session-runner-message.test.ts index 5798b665a8..1c33dfc26e 100644 --- a/packages/core/test/session-runner-message.test.ts +++ b/packages/core/test/session-runner-message.test.ts @@ -327,7 +327,7 @@ Recent work ]) }) - test("drops provider-native continuation metadata from failed assistant turns", () => { + test("drops provider-native continuation metadata from interrupted assistant turns", () => { const messages = toLLMMessages( [ SessionMessage.Assistant.make({ @@ -361,8 +361,7 @@ Recent work time: { created, completed: created }, }), ], - finish: "error", - error: { type: "unknown", message: "Provider turn interrupted" }, + finish: "interrupted", time: { created, completed: created }, }), ], diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 6df50ac575..706659d371 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -542,12 +542,19 @@ const verifyPartialFlushOnInterruption = (kind: FragmentKind) => const fiber = yield* runner.run({ sessionID, force: true }).pipe(Effect.forkChild) yield* Deferred.await(streamed) yield* Fiber.interrupt(fiber) + const { db } = yield* Database.Service + const interrupted = yield* db + .select({ type: EventTable.type }) + .from(EventTable) + .where(eq(EventTable.type, EventV2.versionedType(SessionEvent.Step.Interrupted.type, 1))) + .all() + .pipe(Effect.orDie) + expect(interrupted).toHaveLength(1) expect(yield* session.context(sessionID)).toMatchObject([ { type: "user", text: prompt }, { type: "assistant", - finish: "error", - error: { type: "unknown", message: "Provider turn interrupted" }, + finish: "interrupted", content: [ kind === "tool input" ? { type: "tool", id: fragmentID(kind, "interrupted"), state: { status: "error" } } diff --git a/packages/opencode/test/event-manifest.test.ts b/packages/opencode/test/event-manifest.test.ts index da0f80cee5..86f690e5aa 100644 --- a/packages/opencode/test/event-manifest.test.ts +++ b/packages/opencode/test/event-manifest.test.ts @@ -9,7 +9,7 @@ describe("public event manifest", () => { expect(EventManifest.Definitions).toBe(SchemaEventManifest.Definitions) expect(EventManifest.Latest).toBe(SchemaEventManifest.Latest) expect(EventManifest.Durable).toBe(SchemaEventManifest.Durable) - expect(EventManifest.Latest.size).toBe(88) + expect(EventManifest.Latest.size).toBe(89) expect(EventManifest.Latest.get("session.next.step.ended")).toBe(SessionEvent.Step.Ended) expect(EventManifest.Latest.get("todo.updated")).toBe(Todo.Event.Updated) expect(EventManifest.Latest.has("ide.installed")).toBe(false) diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index f58c3dcce0..b84ada4511 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -192,6 +192,16 @@ export namespace Step { }, }) export type Failed = typeof Failed.Type + + export const Interrupted = Event.define({ + type: "session.next.step.interrupted", + ...options, + schema: { + ...Base, + assistantMessageID: SessionMessage.ID, + }, + }) + export type Interrupted = typeof Interrupted.Type } export namespace Text { @@ -458,6 +468,7 @@ export const DurableDefinitions = Event.inventory( Step.Started, Step.Ended, Step.Failed, + Step.Interrupted, Text.Started, Text.Ended, Tool.Input.Started, @@ -489,6 +500,7 @@ export const Definitions = Event.inventory( Step.Started, Step.Ended, Step.Failed, + Step.Interrupted, Text.Started, Text.Delta, Text.Ended, diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index 5694afdd30..dbd1227c66 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -9,8 +9,8 @@ import { WorkspaceEvent } from "../src/workspace-event" describe("public event manifest", () => { test("owns the complete public event surface", () => { - expect(EventManifest.ServerDefinitions.length).toBe(55) - expect(EventManifest.Definitions.length).toBe(85) + expect(EventManifest.ServerDefinitions.length).toBe(59) + expect(EventManifest.Definitions.length).toBe(89) expect(SessionV1.Event.Definitions).toEqual([ SessionV1.Event.Created, SessionV1.Event.Updated, @@ -23,8 +23,8 @@ describe("public event manifest", () => { SessionV1.Event.Diff, SessionV1.Event.Error, ]) - expect(EventManifest.Latest.size).toBe(85) - expect(EventManifest.Durable.size).toBe(32) + expect(EventManifest.Latest.size).toBe(89) + expect(EventManifest.Durable.size).toBe(36) }) test("uses canonical definitions for current public events", () => { @@ -42,7 +42,7 @@ describe("public event manifest", () => { expect(Reference.Event.Definitions).toEqual([Reference.Event.Updated]) expect(EventManifest.Latest.has("ide.installed")).toBe(false) expect(IdeEvent.Definitions).toEqual([IdeEvent.Installed]) - expect(EventManifest.Definitions.slice(40, 43)).toEqual([ + expect(EventManifest.Definitions.slice(44, 47)).toEqual([ SessionV1.Event.PartDelta, SessionV1.Event.Diff, SessionV1.Event.Error, diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 9fde0f4e5e..ce70001ae4 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -28,6 +28,7 @@ export type Event = | EventSessionNextStepStarted | EventSessionNextStepEnded | EventSessionNextStepFailed + | EventSessionNextStepInterrupted | EventSessionNextTextStarted | EventSessionNextTextDelta | EventSessionNextTextEnded @@ -955,6 +956,15 @@ export type GlobalEvent = { error: SessionErrorUnknown } } + | { + id: string + type: "session.next.step.interrupted" + properties: { + timestamp: number + sessionID: string + assistantMessageID: string + } + } | { id: string type: "session.next.text.started" @@ -1620,6 +1630,7 @@ export type GlobalEvent = { | SyncEventSessionNextStepStarted | SyncEventSessionNextStepEnded | SyncEventSessionNextStepFailed + | SyncEventSessionNextStepInterrupted | SyncEventSessionNextTextStarted | SyncEventSessionNextTextEnded | SyncEventSessionNextReasoningStarted @@ -2759,6 +2770,7 @@ export type V2Event = | V2EventSessionNextStepStarted | V2EventSessionNextStepEnded | V2EventSessionNextStepFailed + | V2EventSessionNextStepInterrupted | V2EventSessionNextTextStarted | V2EventSessionNextTextDelta | V2EventSessionNextTextEnded @@ -3400,6 +3412,22 @@ export type SyncEventSessionNextStepFailed = { } } +export type SyncEventSessionNextStepInterrupted = { + type: "sync" + id: string + syncEvent: { + type: "session.next.step.interrupted.1" + id: string + seq: number + aggregateID: string + data: { + timestamp: number + sessionID: string + assistantMessageID: string + } + } +} + export type SyncEventSessionNextTextStarted = { type: "sync" id: string @@ -4289,6 +4317,25 @@ export type SessionNextStepFailed = { } } +export type SessionNextStepInterrupted = { + id: string + metadata?: { + [key: string]: unknown + } + type: "session.next.step.interrupted" + durable?: { + aggregateID: string + seq: number | "NaN" | "Infinity" | "-Infinity" + version: number | "NaN" | "Infinity" | "-Infinity" + } + location?: LocationRef + data: { + timestamp: number + sessionID: string + assistantMessageID: string + } +} + export type SessionNextTextStarted = { id: string metadata?: { @@ -5344,6 +5391,25 @@ export type V2EventSessionNextStepFailed = { } } +export type V2EventSessionNextStepInterrupted = { + id: string + metadata?: { + [key: string]: unknown + } + durable?: { + aggregateID: string + seq: number + version: number + } + location?: LocationRef + type: "session.next.step.interrupted" + data: { + timestamp: number + sessionID: string + assistantMessageID: string + } +} + export type V2EventSessionNextTextStarted = { id: string metadata?: { @@ -6931,6 +6997,16 @@ export type EventSessionNextStepFailed = { } } +export type EventSessionNextStepInterrupted = { + id: string + type: "session.next.step.interrupted" + properties: { + timestamp: number + sessionID: string + assistantMessageID: string + } +} + export type EventSessionNextTextStarted = { id: string type: "session.next.text.started"