ceos/opencode
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(core): enforce one step settlement

Browse source
This commit is contained in:
Kit Langton 2026-06-25 23:25:08 -04:00
commit db664db5f3
8 changed files with 71 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

6
packages/client/src/generated/types.ts
View file

@ -571,7 +571,7 @@ export type SessionsContextOutput = {
}
>
readonly snapshot?: { readonly start?: string; readonly end?: string; readonly files?: ReadonlyArray<string> }
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<string> }
readonly finish?: string
readonly finish?: "stop" | "length" | "tool-calls" | "content-filter" | "error" | "unknown" | "interrupted"
readonly cost?: number
readonly tokens?: {
readonly input: number

2
packages/core/src/session/runner/llm.ts
View file

@ -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

53
packages/core/src/session/runner/publish-llm-event.ts
View file

@ -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<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,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<void>,
) {
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,

17
packages/core/test/session-runner.test.ts
View file

@ -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",
},
])
}),

4
packages/schema/src/session-event.ts
View file

@ -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,

13
packages/schema/src/session-message.ts
View file

@ -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,

5
packages/schema/test/event-manifest.test.ts
View file

@ -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)
})

14
packages/sdk/js/src/v2/gen/types.gen.ts
View file

@ -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<string>
}
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