feat(core): publish interrupted session steps
This commit is contained in:
parent
40ecfceec2
commit
59e8dbb066
12 changed files with 140 additions and 17 deletions
|
|
@ -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 }
|
||||
|
|
|
|||
|
|
@ -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(
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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<typeof tokens> } | 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,
|
||||
|
|
|
|||
|
|
@ -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")
|
||||
|
|
|
|||
|
|
@ -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 },
|
||||
}),
|
||||
],
|
||||
|
|
|
|||
|
|
@ -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" } }
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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,
|
||||
|
|
|
|||
|
|
@ -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"
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue