Compare commits

...

3 Commits

Author SHA1 Message Date
Kit Langton fd229f7edf fix(core): separate interruption from finish 2026-06-26 11:04:08 -04:00
Kit Langton db664db5f3 fix(core): enforce one step settlement 2026-06-25 23:25:08 -04:00
Kit Langton 59e8dbb066 feat(core): publish interrupted session steps 2026-06-25 23:20:44 -04:00
16 changed files with 305 additions and 31 deletions
+10
View File
@@ -572,6 +572,7 @@ export type SessionsContextOutput = {
>
readonly snapshot?: { readonly start?: string; readonly end?: string; readonly files?: ReadonlyArray<string> }
readonly finish?: string
readonly settlement?: "completed" | "failed" | "interrupted"
readonly cost?: number
readonly tokens?: {
readonly input: number
@@ -795,6 +796,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 }
@@ -1189,6 +1198,7 @@ export type SessionsMessageOutput = {
>
readonly snapshot?: { readonly start?: string; readonly end?: string; readonly files?: ReadonlyArray<string> }
readonly finish?: string
readonly settlement?: "completed" | "failed" | "interrupted"
readonly cost?: number
readonly tokens?: {
readonly input: number
@@ -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,9 +225,16 @@ 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.settlement = "interrupted"
})
},
"session.next.text.started": (event) => {
return updateOwnedAssistant(event.data.assistantMessageID, (draft) => {
draft.content.push(
+1
View File
@@ -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))
+3 -3
View File
@@ -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)
@@ -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
@@ -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 assistantFailed = false
let assistantSettled = false
let providerFailed = false
let stepSettlement: { readonly finish: string; readonly tokens: ReturnType<typeof tokens> } | undefined
let stepSettlement: { readonly finish: SessionMessage.Finish; readonly tokens: ReturnType<typeof tokens> } | 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,20 +194,39 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
yield* flushFragments()
})
const failAssistant = Effect.fnUntraced(function* (message: string) {
if (assistantFailed) return
const settleAssistant = Effect.fnUntraced(function* (
publish: (assistantMessageID: SessionMessage.ID) => Effect.Effect<void>,
) {
if (assistantSettled) return
yield* flush()
const assistantMessageID = yield* startAssistant()
assistantActive = false
assistantFailed = true
yield* events.publish(SessionEvent.Step.Failed, {
sessionID: input.sessionID,
timestamp: yield* timestamp,
assistantMessageID,
error: { type: "unknown", message },
})
assistantSettled = true
yield* publish(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,
hostedOnly = false,
@@ -395,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
@@ -412,9 +428,10 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
publish,
flush,
failAssistant,
interruptAssistant,
failUnsettledTools,
hasActiveAssistant: () => assistantActive,
hasAssistantStarted: () => assistantMessageID !== undefined,
hasAssistantSettled: () => assistantSettled,
hasProviderError: () => providerFailed,
stepSettlement: () => stepSettlement,
startAssistant,
@@ -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.settlement !== "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" },
settlement: "interrupted",
time: { created, completed: created },
}),
],
+21 -2
View File
@@ -519,6 +519,7 @@ const verifyPartialFlushOnFailure = (kind: FragmentKind) =>
{
type: "assistant",
finish: "error",
settlement: "failed",
error: { type: "unknown", message: "Provider unavailable" },
content: [fixture.expectedContent],
},
@@ -542,12 +543,28 @@ 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 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.aggregate_id, sessionID))
.all()
.pipe(Effect.orDie)
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 },
{
type: "assistant",
finish: "error",
error: { type: "unknown", message: "Provider turn interrupted" },
settlement: "interrupted",
content: [
kind === "tool input"
? { type: "tool", id: fragmentID(kind, "interrupted"), state: { status: "error" } }
@@ -2653,6 +2670,7 @@ describe("SessionRunnerLLM", () => {
state: { status: "error", error: { type: "unknown", message: "Tool execution interrupted" } },
},
],
settlement: "interrupted",
},
])
}),
@@ -2797,6 +2815,7 @@ describe("SessionRunnerLLM", () => {
state: { status: "error", error: { type: "unknown", message: "Tool execution interrupted" } },
},
],
settlement: "interrupted",
},
])
}),
@@ -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)
+13
View File
@@ -165,6 +165,7 @@ export namespace Step {
schema: {
...Base,
assistantMessageID: SessionMessage.ID,
// Step.Ended v2 was originally persisted with an open string schema.
finish: Schema.String,
cost: Schema.Finite,
tokens: Schema.Struct({
@@ -192,6 +193,16 @@ export namespace Step {
},
})
export type Failed = typeof Failed.Type
export const Interrupted = Event.define({
type: "session.next.step.interrupted",
...stepSettlementOptions,
schema: {
...Base,
assistantMessageID: SessionMessage.ID,
},
})
export type Interrupted = typeof Interrupted.Type
}
export namespace Text {
@@ -458,6 +469,7 @@ export const DurableDefinitions = Event.inventory(
Step.Started,
Step.Ended,
Step.Failed,
Step.Interrupted,
Text.Started,
Text.Ended,
Tool.Input.Started,
@@ -489,6 +501,7 @@ export const Definitions = Event.inventory(
Step.Started,
Step.Ended,
Step.Failed,
Step.Interrupted,
Text.Started,
Text.Delta,
Text.Ended,
+15
View File
@@ -21,6 +21,19 @@ 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",
])
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),
@@ -169,7 +182,9 @@ export const Assistant = Schema.Struct({
end: Schema.String.pipe(optional),
files: Schema.Array(RelativePath).pipe(optional),
}).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,
+22 -5
View File
@@ -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"
@@ -9,8 +10,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 +24,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,12 +43,28 @@ 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([
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)
})
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")
})
})
@@ -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" })
})
+77
View File
@@ -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.2"
id: string
seq: number
aggregateID: string
data: {
timestamp: number
sessionID: string
assistantMessageID: string
}
}
}
export type SyncEventSessionNextTextStarted = {
type: "sync"
id: string
@@ -3995,6 +4023,7 @@ export type SessionMessageAssistant = {
files?: Array<string>
}
finish?: string
settlement?: "completed" | "failed" | "interrupted"
cost?: number
tokens?: {
input: number
@@ -4289,6 +4318,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 +5392,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 +6998,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"
+10
View File
@@ -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({
+66
View File
@@ -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<typeof useData>
let ready!: () => void
const mounted = new Promise<void>((resolve) => {
ready = resolve
})
function Probe() {
sync = useData()
onMount(ready)
return <box />
}
const app = await testRender(() => (
<TestTuiContexts>
<SDKProvider url="http://test" directory={directory} events={events.source} fetch={calls.fetch}>
<ProjectProvider>
<DataProvider>
<Probe />
</DataProvider>
</ProjectProvider>
</SDKProvider>
</TestTuiContexts>
))
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)