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