From 59e8dbb066b71e37610f4777fe0b59c15f4ac4a6 Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 25 Jun 2026 23:20:44 -0400 Subject: [PATCH 1/3] 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" From db664db5f3da9263e48b764c35c5f99a5980e55e Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 25 Jun 2026 23:25:08 -0400 Subject: [PATCH 2/3] fix(core): enforce one step settlement --- packages/client/src/generated/types.ts | 6 +-- packages/core/src/session/runner/llm.ts | 2 +- .../src/session/runner/publish-llm-event.ts | 53 ++++++++++--------- packages/core/test/session-runner.test.ts | 17 ++++-- packages/schema/src/session-event.ts | 4 +- packages/schema/src/session-message.ts | 13 ++++- packages/schema/test/event-manifest.test.ts | 5 +- packages/sdk/js/src/v2/gen/types.gen.ts | 14 ++--- 8 files changed, 71 insertions(+), 43 deletions(-) diff --git a/packages/client/src/generated/types.ts b/packages/client/src/generated/types.ts index 2f3baf7123..3f580b0dc1 100644 --- a/packages/client/src/generated/types.ts +++ b/packages/client/src/generated/types.ts @@ -571,7 +571,7 @@ export type SessionsContextOutput = { } > readonly snapshot?: { readonly start?: string; readonly end?: string; readonly files?: ReadonlyArray } - readonly finish?: string + readonly finish?: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" readonly cost?: number readonly tokens?: { readonly input: number @@ -770,7 +770,7 @@ export type SessionsEventsOutput = readonly timestamp: number readonly sessionID: string readonly assistantMessageID: string - readonly finish: string + readonly finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" readonly cost: number readonly tokens: { readonly input: number @@ -1196,7 +1196,7 @@ export type SessionsMessageOutput = { } > readonly snapshot?: { readonly start?: string; readonly end?: string; readonly files?: ReadonlyArray } - readonly finish?: string + readonly finish?: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" readonly cost?: number readonly tokens?: { readonly input: number diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 3ff580e666..d0b6d6d729 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -307,7 +307,7 @@ export const layer = Layer.effect( yield* withPublication(publisher.failUnsettledTools(`Tool execution failed: ${message}`)) } const stepSettlement = publisher.stepSettlement() - if (stepSettlement && !publisher.hasProviderError()) { + if (stepSettlement && !publisher.hasProviderError() && !publisher.hasAssistantSettled()) { const endSnapshot = yield* snapshots.capture() const files = startSnapshot && endSnapshot diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 471950cbff..0fd87c109a 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -66,15 +66,13 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) >() const timestamp = DateTime.now let assistantMessageID: SessionMessage.ID | undefined - let assistantActive = false let assistantSettled = false let providerFailed = false - let stepSettlement: { readonly finish: string; readonly tokens: ReturnType } | undefined + let stepSettlement: { readonly finish: SessionMessage.Finish; readonly tokens: ReturnType } | undefined const startAssistant = Effect.fnUntraced(function* () { if (assistantMessageID !== undefined) return assistantMessageID assistantMessageID = SessionMessage.ID.create() - assistantActive = true yield* events.publish(SessionEvent.Step.Started, { ...input, assistantMessageID, @@ -196,32 +194,38 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) yield* flushFragments() }) - const failAssistant = Effect.fnUntraced(function* (message: string) { + const settleAssistant = Effect.fnUntraced(function* ( + publish: (assistantMessageID: SessionMessage.ID) => Effect.Effect, + ) { if (assistantSettled) return yield* flush() const assistantMessageID = yield* startAssistant() - assistantActive = false assistantSettled = true - yield* events.publish(SessionEvent.Step.Failed, { - sessionID: input.sessionID, - timestamp: yield* timestamp, - assistantMessageID, - error: { type: "unknown", message }, - }) + yield* publish(assistantMessageID) }) - 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 failAssistant = (message: string) => + settleAssistant((assistantMessageID) => + Effect.gen(function* () { + yield* events.publish(SessionEvent.Step.Failed, { + sessionID: input.sessionID, + timestamp: yield* timestamp, + assistantMessageID, + error: { type: "unknown", message }, + }) + }), + ) + + const interruptAssistant = () => + settleAssistant((assistantMessageID) => + Effect.gen(function* () { + yield* events.publish(SessionEvent.Step.Interrupted, { + sessionID: input.sessionID, + timestamp: yield* timestamp, + assistantMessageID, + }) + }), + ) const failUnsettledTools = Effect.fn("SessionRunner.failUnsettledTools")(function* ( message: string, @@ -408,7 +412,6 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) } case "step-finish": yield* flush() - assistantActive = false if (stepSettlement) return yield* Effect.die("Duplicate step finish") stepSettlement = { finish: event.reason, tokens: tokens(event.usage) } return @@ -427,8 +430,8 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input) failAssistant, interruptAssistant, failUnsettledTools, - hasActiveAssistant: () => assistantActive, hasAssistantStarted: () => assistantMessageID !== undefined, + hasAssistantSettled: () => assistantSettled, hasProviderError: () => providerFailed, stepSettlement: () => stepSettlement, startAssistant, diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 706659d371..edc1607e2a 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -543,13 +543,22 @@ const verifyPartialFlushOnInterruption = (kind: FragmentKind) => yield* Deferred.await(streamed) yield* Fiber.interrupt(fiber) const { db } = yield* Database.Service - const interrupted = yield* db + const interruptedVersion = SessionEvent.Step.Interrupted.durable?.version + expect(interruptedVersion).toBe(2) + if (interruptedVersion === undefined) return yield* Effect.die("Step.Interrupted must be durable") + const settlements = yield* db .select({ type: EventTable.type }) .from(EventTable) - .where(eq(EventTable.type, EventV2.versionedType(SessionEvent.Step.Interrupted.type, 1))) + .where(eq(EventTable.aggregate_id, sessionID)) .all() .pipe(Effect.orDie) - expect(interrupted).toHaveLength(1) + expect( + settlements.filter(({ type }) => + [SessionEvent.Step.Ended.type, SessionEvent.Step.Failed.type, SessionEvent.Step.Interrupted.type].some((settled) => + type.startsWith(settled), + ), + ), + ).toEqual([{ type: EventV2.versionedType(SessionEvent.Step.Interrupted.type, interruptedVersion) }]) expect(yield* session.context(sessionID)).toMatchObject([ { type: "user", text: prompt }, { @@ -2660,6 +2669,7 @@ describe("SessionRunnerLLM", () => { state: { status: "error", error: { type: "unknown", message: "Tool execution interrupted" } }, }, ], + finish: "interrupted", }, ]) }), @@ -2804,6 +2814,7 @@ describe("SessionRunnerLLM", () => { state: { status: "error", error: { type: "unknown", message: "Tool execution interrupted" } }, }, ], + finish: "interrupted", }, ]) }), diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index b84ada4511..8b40e296c2 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -165,7 +165,7 @@ export namespace Step { schema: { ...Base, assistantMessageID: SessionMessage.ID, - finish: Schema.String, + finish: SessionMessage.Finish, cost: Schema.Finite, tokens: Schema.Struct({ input: Schema.Finite, @@ -195,7 +195,7 @@ export namespace Step { export const Interrupted = Event.define({ type: "session.next.step.interrupted", - ...options, + ...stepSettlementOptions, schema: { ...Base, assistantMessageID: SessionMessage.ID, diff --git a/packages/schema/src/session-message.ts b/packages/schema/src/session-message.ts index f32bb0582e..90356e60fe 100644 --- a/packages/schema/src/session-message.ts +++ b/packages/schema/src/session-message.ts @@ -21,6 +21,17 @@ export const UnknownError = Schema.Struct({ message: Schema.String, }).annotate({ identifier: "Session.Error.Unknown" }) +export const Finish = Schema.Literals([ + "stop", + "length", + "tool-calls", + "content-filter", + "error", + "unknown", + "interrupted", +]) +export type Finish = typeof Finish.Type + const Base = { id: ID, metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(optional), @@ -169,7 +180,7 @@ export const Assistant = Schema.Struct({ end: Schema.String.pipe(optional), files: Schema.Array(RelativePath).pipe(optional), }).pipe(optional), - finish: Schema.String.pipe(optional), + finish: Finish.pipe(optional), cost: Schema.Finite.pipe(optional), tokens: Schema.Struct({ input: Schema.Finite, diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index dbd1227c66..bc81467d36 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -42,11 +42,14 @@ 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(44, 47)).toEqual([ + const partDelta = EventManifest.Definitions.indexOf(SessionV1.Event.PartDelta) + expect(partDelta).toBeGreaterThanOrEqual(0) + expect(EventManifest.Definitions.slice(partDelta, partDelta + 3)).toEqual([ SessionV1.Event.PartDelta, SessionV1.Event.Diff, SessionV1.Event.Error, ]) + expect(EventManifest.Latest.get("session.next.step.interrupted")).toBe(SessionEvent.Step.Interrupted) expect(EventManifest.Durable.has("session.next.step.ended.1")).toBe(false) expect(EventManifest.Durable.get("session.next.step.ended.2")).toBe(SessionEvent.Step.Ended) }) diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index ce70001ae4..5cc34c1c65 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -931,7 +931,7 @@ export type GlobalEvent = { timestamp: number sessionID: string assistantMessageID: string - finish: string + finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" cost: number tokens: { input: number @@ -3378,7 +3378,7 @@ export type SyncEventSessionNextStepEnded = { timestamp: number sessionID: string assistantMessageID: string - finish: string + finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" cost: number tokens: { input: number @@ -3416,7 +3416,7 @@ export type SyncEventSessionNextStepInterrupted = { type: "sync" id: string syncEvent: { - type: "session.next.step.interrupted.1" + type: "session.next.step.interrupted.2" id: string seq: number aggregateID: string @@ -4022,7 +4022,7 @@ export type SessionMessageAssistant = { end?: string files?: Array } - finish?: string + finish?: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" cost?: number tokens?: { input: number @@ -4281,7 +4281,7 @@ export type SessionNextStepEnded = { timestamp: number sessionID: string assistantMessageID: string - finish: string + finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" cost: number tokens: { input: number @@ -5355,7 +5355,7 @@ export type V2EventSessionNextStepEnded = { timestamp: number sessionID: string assistantMessageID: string - finish: string + finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" cost: number tokens: { input: number @@ -6970,7 +6970,7 @@ export type EventSessionNextStepEnded = { timestamp: number sessionID: string assistantMessageID: string - finish: string + finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" cost: number tokens: { input: number From fd229f7edfc986f28a78013da0997130e6f2a38c Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Fri, 26 Jun 2026 11:04:08 -0400 Subject: [PATCH 3/3] fix(core): separate interruption from finish --- packages/client/src/generated/types.ts | 8 ++- packages/core/src/session/message-updater.ts | 4 +- .../core/src/session/runner/to-llm-message.ts | 2 +- .../core/test/session-runner-message.test.ts | 2 +- packages/core/test/session-runner.test.ts | 7 +- packages/schema/src/session-event.ts | 3 +- packages/schema/src/session-message.ts | 8 ++- packages/schema/test/event-manifest.test.ts | 14 ++++ packages/schema/test/session-message.test.ts | 22 +++++++ packages/sdk/js/src/v2/gen/types.gen.ts | 13 ++-- packages/tui/src/context/data.tsx | 10 +++ packages/tui/test/cli/tui/data.test.tsx | 66 +++++++++++++++++++ 12 files changed, 141 insertions(+), 18 deletions(-) create mode 100644 packages/schema/test/session-message.test.ts diff --git a/packages/client/src/generated/types.ts b/packages/client/src/generated/types.ts index 3f580b0dc1..9943b03fd8 100644 --- a/packages/client/src/generated/types.ts +++ b/packages/client/src/generated/types.ts @@ -571,7 +571,8 @@ export type SessionsContextOutput = { } > readonly snapshot?: { readonly start?: string; readonly end?: string; readonly files?: ReadonlyArray } - readonly finish?: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" + readonly finish?: string + readonly settlement?: "completed" | "failed" | "interrupted" readonly cost?: number readonly tokens?: { readonly input: number @@ -770,7 +771,7 @@ export type SessionsEventsOutput = readonly timestamp: number readonly sessionID: string readonly assistantMessageID: string - readonly finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" + readonly finish: string readonly cost: number readonly tokens: { readonly input: number @@ -1196,7 +1197,8 @@ export type SessionsMessageOutput = { } > readonly snapshot?: { readonly start?: string; readonly end?: string; readonly files?: ReadonlyArray } - readonly finish?: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" + readonly finish?: string + readonly settlement?: "completed" | "failed" | "interrupted" readonly cost?: number readonly tokens?: { readonly input: number diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 2c92af8583..006dda7e6e 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -210,6 +210,7 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { return updateOwnedAssistant(event.data.assistantMessageID, (draft) => { draft.time.completed = event.data.timestamp draft.finish = event.data.finish + draft.settlement = "completed" draft.cost = event.data.cost draft.tokens = event.data.tokens if (event.data.snapshot || event.data.files) @@ -224,13 +225,14 @@ export function update(adapter: Adapter, event: SessionEvent.Event) { return updateOwnedAssistant(event.data.assistantMessageID, (draft) => { draft.time.completed = event.data.timestamp draft.finish = "error" + draft.settlement = "failed" 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" + draft.settlement = "interrupted" }) }, "session.next.text.started": (event) => { diff --git a/packages/core/src/session/runner/to-llm-message.ts b/packages/core/src/session/runner/to-llm-message.ts index dcc6cd99a7..4fdb5cef33 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 && message.finish !== "interrupted" + const reuseProviderMetadata = sameModel && message.error === undefined && message.settlement !== "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 1c33dfc26e..0b0a793366 100644 --- a/packages/core/test/session-runner-message.test.ts +++ b/packages/core/test/session-runner-message.test.ts @@ -361,7 +361,7 @@ Recent work time: { created, completed: created }, }), ], - finish: "interrupted", + settlement: "interrupted", time: { created, completed: created }, }), ], diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index edc1607e2a..b1d02aa7bc 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -519,6 +519,7 @@ const verifyPartialFlushOnFailure = (kind: FragmentKind) => { type: "assistant", finish: "error", + settlement: "failed", error: { type: "unknown", message: "Provider unavailable" }, content: [fixture.expectedContent], }, @@ -563,7 +564,7 @@ const verifyPartialFlushOnInterruption = (kind: FragmentKind) => { type: "user", text: prompt }, { type: "assistant", - finish: "interrupted", + settlement: "interrupted", content: [ kind === "tool input" ? { type: "tool", id: fragmentID(kind, "interrupted"), state: { status: "error" } } @@ -2669,7 +2670,7 @@ describe("SessionRunnerLLM", () => { state: { status: "error", error: { type: "unknown", message: "Tool execution interrupted" } }, }, ], - finish: "interrupted", + settlement: "interrupted", }, ]) }), @@ -2814,7 +2815,7 @@ describe("SessionRunnerLLM", () => { state: { status: "error", error: { type: "unknown", message: "Tool execution interrupted" } }, }, ], - finish: "interrupted", + settlement: "interrupted", }, ]) }), diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index 8b40e296c2..2f92291325 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -165,7 +165,8 @@ export namespace Step { schema: { ...Base, assistantMessageID: SessionMessage.ID, - finish: SessionMessage.Finish, + // Step.Ended v2 was originally persisted with an open string schema. + finish: Schema.String, cost: Schema.Finite, tokens: Schema.Struct({ input: Schema.Finite, diff --git a/packages/schema/src/session-message.ts b/packages/schema/src/session-message.ts index 90356e60fe..0c50c41d9e 100644 --- a/packages/schema/src/session-message.ts +++ b/packages/schema/src/session-message.ts @@ -28,10 +28,12 @@ export const Finish = Schema.Literals([ "content-filter", "error", "unknown", - "interrupted", ]) export type Finish = typeof Finish.Type +export const Settlement = Schema.Literals(["completed", "failed", "interrupted"]) +export type Settlement = typeof Settlement.Type + const Base = { id: ID, metadata: Schema.Record(Schema.String, Schema.Unknown).pipe(optional), @@ -180,7 +182,9 @@ export const Assistant = Schema.Struct({ end: Schema.String.pipe(optional), files: Schema.Array(RelativePath).pipe(optional), }).pipe(optional), - finish: Finish.pipe(optional), + // Projected histories predate the typed provider finish model and may contain arbitrary values. + finish: Schema.String.pipe(optional), + settlement: Settlement.pipe(optional), cost: Schema.Finite.pipe(optional), tokens: Schema.Struct({ input: Schema.Finite, diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index bc81467d36..94c3de67e2 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -1,4 +1,5 @@ import { describe, expect, test } from "bun:test" +import { Schema } from "effect" import { FileSystem, Integration, Permission, Project, Reference, Session, Workspace } from "../src" import { EventManifest } from "../src/event-manifest" import { IdeEvent } from "../src/ide-event" @@ -53,4 +54,17 @@ describe("public event manifest", () => { expect(EventManifest.Durable.has("session.next.step.ended.1")).toBe(false) expect(EventManifest.Durable.get("session.next.step.ended.2")).toBe(SessionEvent.Step.Ended) }) + + test("decodes legacy Step.Ended v2 finish strings", () => { + const event = Schema.decodeUnknownSync(SessionEvent.Step.Ended.data)({ + sessionID: "ses_legacy", + timestamp: 0, + assistantMessageID: "msg_legacy", + finish: "legacy-provider-reason", + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + }) + + expect(event.finish).toBe("legacy-provider-reason") + }) }) diff --git a/packages/schema/test/session-message.test.ts b/packages/schema/test/session-message.test.ts new file mode 100644 index 0000000000..df0ef623f1 --- /dev/null +++ b/packages/schema/test/session-message.test.ts @@ -0,0 +1,22 @@ +import { expect, test } from "bun:test" +import { Schema } from "effect" +import { SessionMessage } from "../src/session-message" + +test("does not model interruption as a provider finish reason", () => { + expect(() => Schema.decodeUnknownSync(SessionMessage.Finish)("interrupted")).toThrow() + expect(Schema.decodeUnknownSync(SessionMessage.Finish)("error")).toBe("error") +}) + +test("decodes projected assistant histories with arbitrary finish strings", () => { + const message = Schema.decodeUnknownSync(SessionMessage.Message)({ + id: "msg_legacy", + type: "assistant", + agent: "build", + model: { id: "model", providerID: "provider" }, + content: [], + finish: "legacy-provider-reason", + time: { created: 0, completed: 1 }, + }) + + expect(message).toMatchObject({ type: "assistant", finish: "legacy-provider-reason" }) +}) diff --git a/packages/sdk/js/src/v2/gen/types.gen.ts b/packages/sdk/js/src/v2/gen/types.gen.ts index 5cc34c1c65..2902fc85c2 100644 --- a/packages/sdk/js/src/v2/gen/types.gen.ts +++ b/packages/sdk/js/src/v2/gen/types.gen.ts @@ -931,7 +931,7 @@ export type GlobalEvent = { timestamp: number sessionID: string assistantMessageID: string - finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" + finish: string cost: number tokens: { input: number @@ -3378,7 +3378,7 @@ export type SyncEventSessionNextStepEnded = { timestamp: number sessionID: string assistantMessageID: string - finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" + finish: string cost: number tokens: { input: number @@ -4022,7 +4022,8 @@ export type SessionMessageAssistant = { end?: string files?: Array } - finish?: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" + finish?: string + settlement?: "completed" | "failed" | "interrupted" cost?: number tokens?: { input: number @@ -4281,7 +4282,7 @@ export type SessionNextStepEnded = { timestamp: number sessionID: string assistantMessageID: string - finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" + finish: string cost: number tokens: { input: number @@ -5355,7 +5356,7 @@ export type V2EventSessionNextStepEnded = { timestamp: number sessionID: string assistantMessageID: string - finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" + finish: string cost: number tokens: { input: number @@ -6970,7 +6971,7 @@ export type EventSessionNextStepEnded = { timestamp: number sessionID: string assistantMessageID: string - finish: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted" + finish: string cost: number tokens: { input: number diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index 9b2e58907a..7c3197c49f 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -227,6 +227,7 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ if (!currentAssistant) return currentAssistant.time.completed = event.data.timestamp currentAssistant.finish = event.data.finish + currentAssistant.settlement = "completed" currentAssistant.cost = event.data.cost currentAssistant.tokens = event.data.tokens if (event.data.snapshot) @@ -239,9 +240,18 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ if (!currentAssistant) return currentAssistant.time.completed = event.data.timestamp currentAssistant.finish = "error" + currentAssistant.settlement = "failed" currentAssistant.error = event.data.error }) break + case "session.next.step.interrupted": + message.update(event.data.sessionID, (draft) => { + const currentAssistant = message.assistant(draft, event.data.assistantMessageID) + if (!currentAssistant) return + currentAssistant.time.completed = event.data.timestamp + currentAssistant.settlement = "interrupted" + }) + break case "session.next.text.started": message.update(event.data.sessionID, (draft) => { message.assistant(draft, event.data.assistantMessageID)?.content.push({ diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 279ba4d065..49dcfd3be5 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -370,6 +370,72 @@ test("settles pending tools when a live failure arrives", async () => { } }) +test("marks an interrupted assistant inactive without a finish reason", async () => { + const events = createEventSource() + const calls = createFetch(undefined, events) + let sync!: ReturnType + let ready!: () => void + const mounted = new Promise((resolve) => { + ready = resolve + }) + + function Probe() { + sync = useData() + onMount(ready) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await mounted + emitEvent(events, { + id: "evt_step_started_interrupted", + type: "session.next.step.started", + properties: { + sessionID: "session-interrupted", + assistantMessageID: "msg_interrupted", + timestamp: 1, + agent: "build", + model: { id: "model-1", providerID: "provider-1" }, + }, + }) + emitEvent(events, { + id: "evt_step_interrupted", + type: "session.next.step.interrupted", + properties: { + sessionID: "session-interrupted", + assistantMessageID: "msg_interrupted", + timestamp: 2, + }, + }) + + await wait(() => { + const message = sync.session.message.list("session-interrupted")?.[0] + return message?.type === "assistant" && message.time.completed === 2 + }) + const assistant = sync.session.message.list("session-interrupted")?.[0] + expect(assistant).toMatchObject({ + type: "assistant", + settlement: "interrupted", + time: { completed: 2 }, + }) + expect(assistant).not.toHaveProperty("finish") + } finally { + app.renderer.destroy() + } +}) + test("renders admitted prompts only after they become model-visible", async () => { const events = createEventSource() const calls = createFetch(undefined, events)