Compare commits

...

1 Commits

Author SHA1 Message Date
Kit Langton 3323db2319 fix(core): enforce step settlement ordering 2026-07-04 00:51:53 -04:00
11 changed files with 545 additions and 138 deletions
+8 -8
View File
@@ -18,32 +18,32 @@ const TOOL_OUTPUT_MAX_CHARS = 2_000
const SUMMARY_OUTPUT_TOKENS = 4_096
const SUMMARY_TEMPLATE = `Output exactly the Markdown structure shown inside <template> and keep the section order unchanged. Do not include the <template> tags in your response.
<template>
## Goal
## Continuation Goal
- [single-sentence task summary]
## Constraints & Preferences
## Operating Constraints
- [user constraints, preferences, specs, or "(none)"]
## Progress
### Done
### Completed
- [completed work or "(none)"]
### In Progress
### In Flight
- [current work or "(none)"]
### Blocked
- [blockers or "(none)"]
## Key Decisions
## Decisions To Preserve
- [decision and why, or "(none)"]
## Next Steps
## Resume From Here
- [ordered next actions or "(none)"]
## Critical Context
## Context To Preserve
- [important technical facts, errors, open questions, or "(none)"]
## Relevant Files
## Working Files
- [file or directory path: why it matters, or "(none)"]
</template>
+86 -62
View File
@@ -11,7 +11,7 @@ import {
type ProviderErrorEvent,
} from "@opencode-ai/llm"
import { SessionError } from "@opencode-ai/schema/session-error"
import { Cause, Effect, Exit, FiberSet, Layer, Option, Semaphore, Stream } from "effect"
import { Cause, Effect, Exit, Fiber, FiberSet, Layer, Option, Semaphore, Stream } from "effect"
import { AgentV2 } from "../../agent"
import { Config } from "../../config"
import { Database } from "../../database/database"
@@ -148,9 +148,6 @@ const layer = Layer.effect(
}
})
const awaitToolFibers = (fibers: FiberSet.FiberSet<void, ToolOutputStore.Error | UserInterruptedError>) =>
Effect.raceFirst(FiberSet.join(fibers), FiberSet.awaitEmpty(fibers))
// Match V1: dismissing a question halts the loop instead of becoming model-facing tool output.
const isQuestionRejected = (cause: Cause.Cause<unknown>) =>
cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionV2.RejectedError)
@@ -188,6 +185,7 @@ const layer = Layer.effect(
session.id,
)
const toolFibers = yield* FiberSet.make<void, ToolOutputStore.Error | UserInterruptedError>()
const ownedToolFibers: Array<Fiber.Fiber<void, ToolOutputStore.Error | UserInterruptedError>> = []
let needsContinuation = false
let currentStep = step
if (promotion) {
@@ -265,37 +263,39 @@ const layer = Layer.effect(
}
needsContinuation = true
const assistantMessageID = yield* publisher.assistantMessageID(event.id)
yield* Effect.uninterruptibleMask((restore) =>
restore(
toolMaterialization.settle({
sessionID: session.id,
agent: agent.id,
assistantMessageID,
call: event,
}),
).pipe(
Effect.flatMap((settlement) =>
publish(
LLMEvent.toolResult({
id: event.id,
name: event.name,
result: settlement.result,
output: settlement.output,
}),
settlement.outputPaths ?? [],
settlement.error,
).pipe(
Effect.andThen(
settlement.error?.type === "permission.rejected"
? serialized(publisher.failAssistant(settlement.error)).pipe(
Effect.andThen(Effect.fail(new UserInterruptedError())),
)
: Effect.void,
ownedToolFibers.push(
yield* Effect.uninterruptibleMask((restore) =>
restore(
toolMaterialization.settle({
sessionID: session.id,
agent: agent.id,
assistantMessageID,
call: event,
}),
).pipe(
Effect.flatMap((settlement) =>
publish(
LLMEvent.toolResult({
id: event.id,
name: event.name,
result: settlement.result,
output: settlement.output,
}),
settlement.outputPaths ?? [],
settlement.error,
).pipe(
Effect.andThen(
settlement.error?.type === "permission.rejected"
? serialized(publisher.failAssistant(settlement.error)).pipe(
Effect.andThen(Effect.fail(new UserInterruptedError())),
)
: Effect.void,
),
),
),
),
),
).pipe(FiberSet.run(toolFibers))
).pipe(FiberSet.run(toolFibers)),
)
}),
),
Effect.ensuring(serialized(publisher.flush())),
@@ -345,8 +345,8 @@ const layer = Layer.effect(
return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const
// An unrecovered held-back overflow becomes the step's durable provider error. A
// thrown LLM failure fails hosted tool calls and the assistant unless a provider
// error was already recorded from the stream.
// thrown LLM failure records the assistant failure unless a provider error was
// already recorded from the stream. Terminal publication waits for owned tools.
if (overflowFailure) yield* publish(overflowFailure)
const llmFailure = streamFailure instanceof LLMError ? streamFailure : undefined
if (llmFailure && !publisher.hasProviderError()) {
@@ -363,39 +363,39 @@ const layer = Layer.effect(
step: currentStep,
})
}
yield* serialized(
publisher.failUnsettledTools(
{ type: "tool.result-missing", message: "Provider did not return a tool result" },
true,
),
)
yield* serialized(publisher.failAssistant(error))
}
// Provider error events only arrive from the stream, so the flag is final here.
const providerFailed = publisher.hasProviderError()
// Settle tool fibers: an interrupted stream abandons unstarted tool work first.
// Settle every owned tool fiber. FiberSet.join returns on the first failure, so retain
// the individual fibers and await all exits before publishing the terminal step event.
if (streamInterrupted) yield* FiberSet.clear(toolFibers)
const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit)
const toolsInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)
const questionDismissed = settled._tag === "Failure" && isQuestionRejected(settled.cause)
const settledError =
settled._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(settled.cause)) : undefined
const permissionRejected = settledError instanceof UserInterruptedError
const settled = yield* restore(
Effect.forEach(ownedToolFibers, Fiber.await, { concurrency: "unbounded" }),
).pipe(Effect.exit)
const settledCauses =
settled._tag === "Failure"
? [settled.cause]
: settled.value.flatMap((exit) => (exit._tag === "Failure" ? [exit.cause] : []))
const toolsInterrupted = settledCauses.some(Cause.hasInterrupts)
const questionDismissed = settledCauses.some(isQuestionRejected)
const permissionRejected = settledCauses.some(
(cause) => Option.getOrUndefined(Cause.findErrorOption(cause)) instanceof UserInterruptedError,
)
if (questionDismissed || permissionRejected || streamInterrupted || toolsInterrupted) {
yield* FiberSet.clear(toolFibers)
yield* serialized(publisher.failUnsettledTools({ type: "aborted", message: "Tool execution interrupted" }))
yield* serialized(publisher.failAssistant({ type: "aborted", message: "Step interrupted" }))
// Match V1: dismissing a question halts the loop like an interruption.
if (questionDismissed || permissionRejected) return yield* new UserInterruptedError()
}
// A settled tool fiber failure is one of two things. A defect from a tool
// implementation becomes a failed tool call the model can read, and the step still
// settles so the model may recover. A typed infrastructure failure (tool output
// could not be persisted) also fails the assistant and then fails the drain.
const settledFailure =
settled._tag === "Failure" && !toolsInterrupted && !permissionRejected ? settled.cause : undefined
const settledFailure = settledCauses.find(
(cause) => !Cause.hasInterrupts(cause) && !isQuestionRejected(cause) && !permissionRejected,
)
const infraError =
settledFailure === undefined ? undefined : Option.getOrUndefined(Cause.findErrorOption(settledFailure))
if (settledFailure !== undefined) {
@@ -405,26 +405,50 @@ const layer = Layer.effect(
if (infraError !== undefined) yield* serialized(publisher.failAssistant(error))
}
const stepSettlement = publisher.stepSettlement()
const stepEndedCleanly =
!streamInterrupted && !toolsInterrupted && infraError === undefined && !providerFailed
if (stepSettlement && stepEndedCleanly) yield* publishStepEnd(stepSettlement)
// A provider error orphans recorded local calls; a clean stream can still leave
// hosted calls without results.
// Fail unresolved calls before the terminal step event. Local calls have joined, so
// these sweeps only close calls that could not produce a truthful settlement.
if (providerFailed)
yield* serialized(publisher.failUnsettledTools({ type: "aborted", message: "Tool execution interrupted" }))
if (stream._tag === "Success" && !providerFailed)
if (llmFailure && !providerFailed)
yield* serialized(
publisher.failUnsettledTools(
{ type: "tool.result-missing", message: "Provider did not return a tool result" },
{
type: "tool.result-missing",
message: "Provider did not return a tool result",
},
true,
),
)
const hostedResultMissing =
stream._tag === "Success" && !providerFailed
? yield* serialized(
publisher.failUnsettledTools(
{ type: "tool.result-missing", message: "Provider did not return a tool result" },
true,
),
)
: false
if (hostedResultMissing && !publisher.stepSettlement())
yield* serialized(
publisher.failAssistant({
type: "tool.result-missing",
message: "Provider did not return a tool result",
}),
)
const stepFailure = publisher.stepFailure()
const stepSettlement = publisher.stepSettlement()
const stepEndedCleanly =
!streamInterrupted && !toolsInterrupted && infraError === undefined && !providerFailed && !stepFailure
if (stepSettlement && stepEndedCleanly) yield* publishStepEnd(stepSettlement)
if (stepFailure) yield* serialized(publisher.publishStepFailure())
if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause)
if (settled._tag === "Failure" && (toolsInterrupted || infraError !== undefined))
return yield* Effect.failCause(settled.cause)
const stepFailure = publisher.stepFailure()
// Match V1: dismissing a question halts the loop like an interruption.
if (questionDismissed || permissionRejected) return yield* new UserInterruptedError()
if ((toolsInterrupted || infraError !== undefined) && settledFailure)
return yield* Effect.failCause(settledFailure)
if (toolsInterrupted && settled._tag === "Failure") return yield* Effect.failCause(settled.cause)
if (stepFailure) return yield* new StepFailedError({ error: stepFailure })
return {
_tag: "Completed",
@@ -68,7 +68,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
>()
let assistantMessageID = input.assistantMessageID
let stepStarted = false
let assistantFailed = false
let stepFailed = false
let providerFailed = false
let retryEvidence = false
let stepFailure: SessionError.Error | undefined
@@ -206,16 +206,20 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
yield* flushFragments()
})
const failAssistant = Effect.fnUntraced(function* (error: SessionError.Error) {
if (assistantFailed) return
const failAssistant = Effect.fnUntraced(function* (error: SessionError.Error, replace = false) {
yield* flush()
yield* startAssistant()
if (replace || stepFailure === undefined) stepFailure = error
})
const publishStepFailure = Effect.fnUntraced(function* () {
if (stepFailed || stepFailure === undefined) return
const assistantMessageID = yield* startAssistant()
assistantFailed = true
stepFailure = error
stepFailed = true
yield* events.publish(SessionEvent.Step.Failed, {
sessionID: input.sessionID,
assistantMessageID,
error,
error: stepFailure,
})
})
@@ -223,9 +227,11 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
error: SessionError.Error,
hostedOnly = false,
) {
let failed = false
for (const [callID, tool] of tools) {
if (tool.settled || (hostedOnly && !tool.providerExecuted)) continue
tool.settled = true
failed = true
yield* events.publish(SessionEvent.Tool.Failed, {
sessionID: input.sessionID,
assistantMessageID: tool.assistantMessageID,
@@ -234,6 +240,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
executed: tool.providerExecuted,
})
}
return failed
})
const assistantMessageIDForTool = (callID: string) => {
@@ -396,7 +403,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
if (stepSettlement) return yield* Effect.die(new Error("Duplicate step finish"))
if (event.reason === "content-filter") {
providerFailed = true
yield* failAssistant({ type: "provider.content-filter", message: "Provider blocked the response" })
yield* failAssistant({ type: "provider.content-filter", message: "Provider blocked the response" }, true)
return
}
stepSettlement = { finish: event.reason, tokens: tokens(event.usage) }
@@ -405,7 +412,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
return
case "provider-error":
providerFailed = true
yield* failAssistant({ type: "provider.unknown", message: event.message })
yield* failAssistant({ type: "provider.unknown", message: event.message }, true)
return
}
})
@@ -414,6 +421,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
publish,
flush,
failAssistant,
publishStepFailure,
failUnsettledTools,
hasProviderError: () => providerFailed,
hasRetryEvidence: () => retryEvidence,
@@ -68,6 +68,32 @@ test("compaction describes tool media without embedding base64", () => {
expect(serialized).not.toContain(base64)
})
test("compaction prompt requires the continuation checkpoint headings in order", () => {
const prompt = SessionCompaction.buildPrompt({ context: ["Conversation history"] })
expect(prompt.match(/^#{2,3} .+$/gm)).toEqual([
"## Continuation Goal",
"## Operating Constraints",
"## Progress",
"### Completed",
"### In Flight",
"### Blocked",
"## Decisions To Preserve",
"## Resume From Here",
"## Context To Preserve",
"## Working Files",
])
expect(prompt).toContain("single-sentence task summary")
expect(prompt).toContain("user constraints, preferences, specs")
expect(prompt).toContain("completed work")
expect(prompt).toContain("current work")
expect(prompt).toContain("blockers")
expect(prompt).toContain("decision and why")
expect(prompt).toContain("ordered next actions")
expect(prompt).toContain("important technical facts, errors, open questions")
expect(prompt).toContain("file or directory path: why it matters")
expect(prompt).toContain("Keep every section, even when empty.")
})
it.effect("manual compaction summarizes short context instead of no-op", () =>
Effect.gen(function* () {
requests = []
@@ -150,11 +150,13 @@ test("step finish records settlement without publishing step ended", async () =>
expect(publisher.stepSettlement()).toMatchObject({ finish: "stop" })
})
test("content-filter finish fails a contentless step", async () => {
test("content-filter finish retains failure evidence until step closeout", async () => {
const { published, publisher } = capture()
await Effect.runPromise(publisher.publish(LLMEvent.stepStart({ index: 0 })))
await Effect.runPromise(publisher.publish(LLMEvent.stepFinish({ index: 0, reason: "content-filter" })))
expect(published.map((event) => event.type)).toEqual(["session.step.started.1"])
await Effect.runPromise(publisher.publishStepFailure())
expect(published.map((event) => event.type)).toEqual(["session.step.started.1", "session.step.failed.1"])
expect(published.at(-1)?.data).toMatchObject({
error: { type: "provider.content-filter", message: "Provider blocked the response" },
@@ -176,6 +178,7 @@ test("content-filter finish preserves partial streamed text and never ends the s
{ discard: true },
),
)
await Effect.runPromise(publisher.publishStepFailure())
expect(published.some((event) => event.type === "session.step.ended.1")).toBe(false)
expect(published.find((event) => event.type === "session.text.ended.1")?.data).toMatchObject({ text: "Partial" })
+283 -24
View File
@@ -157,7 +157,10 @@ const echo = Layer.effectDiscard(
description: "Fail unexpectedly",
input: Schema.Struct({}),
output: Schema.Struct({}),
execute: () => Effect.die("unexpected tool defect"),
execute: () =>
(toolExecutionGate ? Deferred.await(toolExecutionGate) : Effect.void).pipe(
Effect.andThen(Effect.die("unexpected tool defect")),
),
}),
// BigInt output with no model content forces ToolOutputStore.bound onto its
// JSON.stringify encode path, which fails with a typed StorageError.
@@ -423,6 +426,34 @@ const recordedEventTypes = (id: SessionV2.ID) =>
)
})
const recordedStepSettlementEvents = (id: SessionV2.ID, assistantMessageID: SessionMessage.ID) =>
Effect.gen(function* () {
const { db } = yield* Database.Service
const settlementTypes = new Set([
"session.step.started.1",
"session.tool.called.1",
"session.tool.success.1",
"session.tool.failed.1",
"session.step.ended.1",
"session.step.failed.1",
])
return (yield* db
.select({ type: EventTable.type, data: EventTable.data })
.from(EventTable)
.where(eq(EventTable.aggregate_id, id))
.orderBy(asc(EventTable.seq))
.all()
.pipe(Effect.orDie)).filter(
(event) => settlementTypes.has(event.type) && event.data.assistantMessageID === assistantMessageID,
)
})
const requireAssistant = (messages: readonly SessionMessage.Message[]) => {
const assistant = messages.find((message) => message.type === "assistant")
if (!assistant) throw new Error("Assistant message missing")
return assistant
}
const replaySessionProjection = (id: SessionV2.ID) =>
Effect.gen(function* () {
const { db } = yield* Database.Service
@@ -1300,7 +1331,7 @@ describe("SessionRunnerLLM", () => {
currentModel = compactModel
requests.length = 0
responses = [
fragmentFixture("text", "text-summary", ["## Goal\n- Preserve the task"]).completeEvents,
fragmentFixture("text", "text-summary", ["## Continuation Goal\n- Preserve the task"]).completeEvents,
fragmentFixture("text", "text-final", ["Continued"]).completeEvents,
]
yield* session.prompt({
@@ -1311,22 +1342,22 @@ describe("SessionRunnerLLM", () => {
yield* session.resume(sessionID)
expect(requests).toHaveLength(2)
expect(userTexts(requests[0])[0]).toContain("## Goal")
expect(userTexts(requests[0])[0]).toContain("## Continuation Goal")
expect(userTexts(requests[1])).toHaveLength(1)
expect(userTexts(requests[1])[0]).toContain("<summary>\n## Goal\n- Preserve the task\n</summary>")
expect(userTexts(requests[1])[0]).toContain("<summary>\n## Continuation Goal\n- Preserve the task\n</summary>")
expect(userTexts(requests[1])[0]).toContain(`[User]: ${"Recent exact request ".repeat(180)}`)
const context = yield* (yield* SessionStore.Service).context(sessionID)
expect(context.map((message) => message.type)).toEqual(["compaction", "assistant"])
expect(context[0]).toMatchObject({
type: "compaction",
summary: "## Goal\n- Preserve the task",
summary: "## Continuation Goal\n- Preserve the task",
})
requests.length = 0
executions.length = 0
responses = [
fragmentFixture("text", "text-summary-2", ["## Goal\n- Preserve the updated task"]).completeEvents,
fragmentFixture("text", "text-summary-2", ["## Continuation Goal\n- Preserve the updated task"]).completeEvents,
fragmentFixture("text", "text-final-2", ["Continued again"]).completeEvents,
]
yield* session.prompt({
@@ -1338,12 +1369,12 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(2)
expect(userTexts(requests[0])[0]).toContain(
"<previous-summary>\n## Goal\n- Preserve the task\n</previous-summary>",
"<previous-summary>\n## Continuation Goal\n- Preserve the task\n</previous-summary>",
)
expect(userTexts(requests[0])[0]).toContain("Recent exact request")
expect((yield* (yield* SessionStore.Service).context(sessionID))[0]).toMatchObject({
type: "compaction",
summary: "## Goal\n- Preserve the updated task",
summary: "## Continuation Goal\n- Preserve the updated task",
})
}),
)
@@ -1356,17 +1387,17 @@ describe("SessionRunnerLLM", () => {
LLMEvent.stepStart({ index: 0 }),
LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" }),
],
fragmentFixture("text", "text-summary", ["## Goal\n- Recover overflow"]).completeEvents,
fragmentFixture("text", "text-summary", ["## Continuation Goal\n- Recover overflow"]).completeEvents,
fragmentFixture("text", "text-final", ["Recovered"]).completeEvents,
]
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Continue" }), resume: false })
yield* session.resume(sessionID)
expect(requests).toHaveLength(3)
expect(userTexts(requests[1])[0]).toContain("## Goal")
expect(userTexts(requests[2])[0]).toContain("<summary>\n## Goal\n- Recover overflow\n</summary>")
expect(userTexts(requests[1])[0]).toContain("## Continuation Goal")
expect(userTexts(requests[2])[0]).toContain("<summary>\n## Continuation Goal\n- Recover overflow\n</summary>")
expect(yield* session.context(sessionID)).toMatchObject([
{ type: "compaction", summary: "## Goal\n- Recover overflow" },
{ type: "compaction", summary: "## Continuation Goal\n- Recover overflow" },
{ type: "assistant", finish: "stop" },
])
yield* replaySessionProjection(sessionID)
@@ -1386,7 +1417,7 @@ describe("SessionRunnerLLM", () => {
]
responses = [
overflow(),
fragmentFixture("text", "text-summary", ["## Goal\n- Recover once"]).completeEvents,
fragmentFixture("text", "text-summary", ["## Continuation Goal\n- Recover once"]).completeEvents,
overflow(),
]
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Continue" }), resume: false })
@@ -1414,7 +1445,7 @@ describe("SessionRunnerLLM", () => {
}),
)
responses = [
fragmentFixture("text", "text-summary", ["## Goal\n- Recover raw overflow"]).completeEvents,
fragmentFixture("text", "text-summary", ["## Continuation Goal\n- Recover raw overflow"]).completeEvents,
fragmentFixture("text", "text-final", ["Recovered"]).completeEvents,
]
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Continue" }), resume: false })
@@ -1422,7 +1453,7 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(3)
expect(yield* session.context(sessionID)).toMatchObject([
{ type: "compaction", summary: "## Goal\n- Recover raw overflow" },
{ type: "compaction", summary: "## Continuation Goal\n- Recover raw overflow" },
{ type: "assistant", finish: "stop" },
])
}),
@@ -1453,7 +1484,7 @@ describe("SessionRunnerLLM", () => {
const session = yield* setupOverflowRecovery
responses = [
[LLMEvent.providerError({ message: "prompt too long", classification: "context-overflow" })],
fragmentFixture("text", "text-summary", ["## Goal\n- Interrupted"]).completeEvents,
fragmentFixture("text", "text-summary", ["## Continuation Goal\n- Interrupted"]).completeEvents,
]
const firstGate = yield* Deferred.make<void>()
const summaryGate = yield* Deferred.make<void>()
@@ -1641,7 +1672,8 @@ describe("SessionRunnerLLM", () => {
expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "assistant", "tool"])
expect(authorizations).toMatchObject([{ sessionID, toolCallID: "call-echo" }])
expect(executions).toEqual(["hello"])
expect(yield* session.context(sessionID)).toMatchObject([
const context = yield* session.context(sessionID)
expect(context).toMatchObject([
{ type: "user", text: "Echo this" },
{
type: "assistant",
@@ -1662,6 +1694,13 @@ describe("SessionRunnerLLM", () => {
},
{ type: "assistant", finish: "stop", content: [{ type: "text", text: "Done" }] },
])
const assistant = requireAssistant(context)
expect((yield* recordedStepSettlementEvents(sessionID, assistant.id)).map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.success.1",
"session.step.ended.1",
])
}),
)
@@ -2793,7 +2832,8 @@ describe("SessionRunnerLLM", () => {
expect(requests).toHaveLength(2)
expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "assistant", "tool"])
expect(yield* session.context(sessionID)).toMatchObject([
const context = yield* session.context(sessionID)
expect(context).toMatchObject([
{ type: "user", text: "Call defect" },
{
type: "assistant",
@@ -2810,6 +2850,13 @@ describe("SessionRunnerLLM", () => {
},
{ type: "assistant", finish: "stop", content: [{ type: "text", text: "Recovered" }] },
])
const assistant = requireAssistant(context)
expect((yield* recordedStepSettlementEvents(sessionID, assistant.id)).map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.failed.1",
"session.step.ended.1",
])
}),
)
@@ -3002,7 +3049,8 @@ describe("SessionRunnerLLM", () => {
expect(yield* Fiber.join(run).pipe(Effect.flip)).toBe(failure)
toolExecutionGate = undefined
expect(yield* session.context(sessionID)).toMatchObject([
const context = yield* session.context(sessionID)
expect(context).toMatchObject([
{ type: "user", text: "Settle before failing" },
{
type: "assistant",
@@ -3011,6 +3059,13 @@ describe("SessionRunnerLLM", () => {
],
},
])
const assistant = requireAssistant(context)
expect((yield* recordedStepSettlementEvents(sessionID, assistant.id)).map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.success.1",
"session.step.failed.1",
])
}),
)
@@ -3036,7 +3091,8 @@ describe("SessionRunnerLLM", () => {
expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" })
yield* session.interrupt(sessionID)
expect(yield* session.context(sessionID)).toMatchObject([
const context = yield* session.context(sessionID)
expect(context).toMatchObject([
{ type: "user", text: "Interrupt blocked tool" },
{
type: "assistant",
@@ -3049,6 +3105,13 @@ describe("SessionRunnerLLM", () => {
],
},
])
const assistant = requireAssistant(context)
expect((yield* recordedStepSettlementEvents(sessionID, assistant.id)).map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.failed.1",
"session.step.failed.1",
])
yield* replaySessionProjection(sessionID)
@@ -3304,6 +3367,42 @@ describe("SessionRunnerLLM", () => {
}),
)
it.effect("settles a local tool before one content-filter step failure", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Tool before blocked response" }), resume: false })
toolExecutionGate = yield* Deferred.make<void>()
toolExecutionsStarted = yield* Deferred.make<void>()
toolExecutionsReady = 1
response = [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-before-content-filter", name: "echo", input: { text: "settled" } }),
LLMEvent.stepFinish({ index: 0, reason: "content-filter" }),
LLMEvent.finish({ reason: "content-filter" }),
]
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(toolExecutionsStarted)
yield* Deferred.succeed(toolExecutionGate, undefined)
expect((yield* Fiber.join(run).pipe(Effect.flip)).message).toBe("Provider blocked the response")
toolExecutionGate = undefined
toolExecutionsStarted = undefined
const assistant = requireAssistant(yield* session.context(sessionID))
const events = yield* recordedStepSettlementEvents(sessionID, assistant.id)
expect(events.map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.success.1",
"session.step.failed.1",
])
expect(
events.filter((event) => event.type.startsWith("session.step.") && event.type !== "session.step.started.1"),
).toHaveLength(1)
}),
)
it.effect("does not recover context overflow after durable assistant output", () =>
Effect.gen(function* () {
yield* setup
@@ -3490,16 +3589,32 @@ describe("SessionRunnerLLM", () => {
requests.length = 0
const executionCount = executions.length
toolExecutionGate = yield* Deferred.make<void>()
toolExecutionsStarted = yield* Deferred.make<void>()
toolExecutionsReady = 1
response = [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-before-provider-error", name: "echo", input: { text: "settled" } }),
LLMEvent.providerError({ message: "Provider unavailable" }),
]
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("Provider unavailable")
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(toolExecutionsStarted)
yield* Deferred.succeed(toolExecutionGate, undefined)
expect((yield* Fiber.join(run).pipe(Effect.flip)).message).toBe("Provider unavailable")
toolExecutionGate = undefined
toolExecutionsStarted = undefined
expect(requests).toHaveLength(1)
expect(executions.slice(executionCount)).toEqual(["settled"])
const context = yield* session.context(sessionID)
const assistant = requireAssistant(context)
expect((yield* recordedStepSettlementEvents(sessionID, assistant.id)).map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.success.1",
"session.step.failed.1",
])
}),
)
@@ -3524,13 +3639,47 @@ describe("SessionRunnerLLM", () => {
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("Provider unavailable")
expect(requests).toHaveLength(1)
expect(yield* session.context(sessionID)).toMatchObject([
const context = yield* session.context(sessionID)
expect(context).toMatchObject([
{ type: "user", text: "Fail hosted tool durably" },
{
type: "assistant",
content: [{ type: "tool", id: "call-hosted-provider-error", state: { status: "error" } }],
},
])
const assistant = requireAssistant(context)
expect((yield* recordedStepSettlementEvents(sessionID, assistant.id)).map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.failed.1",
"session.step.failed.1",
])
}),
)
it.effect("preserves a tool defect before provider failure settlement", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Defect while provider fails" }), resume: false })
response = [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-defect-provider-error", name: "defect", input: {} }),
LLMEvent.providerError({ message: "Provider unavailable" }),
]
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("Provider unavailable")
const context = yield* session.context(sessionID)
const assistant = requireAssistant(context)
const events = yield* recordedStepSettlementEvents(sessionID, assistant.id)
expect(events.map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.failed.1",
"session.step.failed.1",
])
expect(events[2]?.data.error).toMatchObject({ type: "unknown", message: "unexpected tool defect" })
}),
)
@@ -3549,16 +3698,113 @@ describe("SessionRunnerLLM", () => {
}),
]
yield* session.resume(sessionID)
expect((yield* session.resume(sessionID).pipe(Effect.flip)).message).toBe("Provider did not return a tool result")
const assistant = requireAssistant(yield* session.context(sessionID))
const events = yield* recordedStepSettlementEvents(sessionID, assistant.id)
expect(events.map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.failed.1",
"session.step.failed.1",
])
expect(
events.filter((event) => event.type.startsWith("session.step.") && event.type !== "session.step.started.1"),
).toHaveLength(1)
yield* replaySessionProjection(sessionID)
expect(yield* session.context(sessionID)).toMatchObject([
{ type: "user", text: "Fail hosted tool at EOF" },
{ type: "assistant", content: [{ type: "tool", id: "call-hosted-eof", state: { status: "error" } }] },
{
type: "assistant",
finish: "error",
error: { type: "tool.result-missing" },
content: [{ type: "tool", id: "call-hosted-eof", state: { status: "error" } }],
},
])
}),
)
it.effect("fails an unresolved hosted tool before one clean step end", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({
sessionID,
prompt: Prompt.make({ text: "Settle hosted tool before ending" }),
resume: false,
})
response = [
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({
id: "call-hosted-clean-end",
name: "web_search",
input: { query: "effect" },
providerExecuted: true,
}),
LLMEvent.stepFinish({ index: 0, reason: "stop" }),
LLMEvent.finish({ reason: "stop" }),
]
yield* session.resume(sessionID)
const assistant = requireAssistant(yield* session.context(sessionID))
const events = yield* recordedStepSettlementEvents(sessionID, assistant.id)
expect(events.map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.failed.1",
"session.step.ended.1",
])
expect(
events.filter((event) => event.type.startsWith("session.step.") && event.type !== "session.step.started.1"),
).toHaveLength(1)
}),
)
it.effect("settles unresolved local and hosted tools before one raw provider failure", () =>
Effect.gen(function* () {
yield* setup
const session = yield* SessionV2.Service
yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Fail unresolved tools" }), resume: false })
const failure = invalidRequest()
const providerFailed = yield* Deferred.make<void>()
toolExecutionGate = yield* Deferred.make<void>()
responseStream = Stream.concat(
Stream.fromIterable([
LLMEvent.stepStart({ index: 0 }),
LLMEvent.toolCall({ id: "call-local-raw-failure", name: "defect", input: {} }),
LLMEvent.toolCall({
id: "call-hosted-raw-failure-pair",
name: "web_search",
input: { query: "effect" },
providerExecuted: true,
}),
]),
Stream.fromEffect(Deferred.succeed(providerFailed, undefined)).pipe(Stream.flatMap(() => Stream.fail(failure))),
)
const run = yield* session.resume(sessionID).pipe(Effect.forkChild)
yield* Deferred.await(providerFailed)
yield* Deferred.succeed(toolExecutionGate, undefined)
expect(yield* Fiber.join(run).pipe(Effect.flip)).toBe(failure)
toolExecutionGate = undefined
const assistant = requireAssistant(yield* session.context(sessionID))
const events = yield* recordedStepSettlementEvents(sessionID, assistant.id)
expect(events.map((event) => ({ type: event.type, callID: event.data.callID }))).toEqual([
{ type: "session.step.started.1", callID: undefined },
{ type: "session.tool.called.1", callID: "call-local-raw-failure" },
{ type: "session.tool.called.1", callID: "call-hosted-raw-failure-pair" },
{ type: "session.tool.failed.1", callID: "call-local-raw-failure" },
{ type: "session.tool.failed.1", callID: "call-hosted-raw-failure-pair" },
{ type: "session.step.failed.1", callID: undefined },
])
expect(
events.filter((event) => event.type.startsWith("session.step.") && event.type !== "session.step.started.1"),
).toHaveLength(1)
}),
)
it.effect("durably fails a hosted tool left unresolved by a raw provider stream failure", () =>
Effect.gen(function* () {
yield* setup
@@ -3585,6 +3831,17 @@ describe("SessionRunnerLLM", () => {
expect(yield* session.resume(sessionID).pipe(Effect.flip)).toBe(failure)
expect(requests).toHaveLength(1)
const assistant = requireAssistant(yield* session.context(sessionID))
const events = yield* recordedStepSettlementEvents(sessionID, assistant.id)
expect(events.map((event) => event.type)).toEqual([
"session.step.started.1",
"session.tool.called.1",
"session.tool.failed.1",
"session.step.failed.1",
])
expect(
events.filter((event) => event.type.startsWith("session.step.") && event.type !== "session.step.started.1"),
).toHaveLength(1)
yield* replaySessionProjection(sessionID)
expect(yield* session.context(sessionID)).toMatchObject([
{ type: "user", text: "Fail hosted tool on raw failure" },
@@ -3699,6 +3956,8 @@ describe("SessionRunnerLLM", () => {
LLMEvent.toolInputDelta({ id: "call-parsed", name: "web_search", text: '{"query":"hello"}' }),
LLMEvent.toolInputEnd({ id: "call-parsed", name: "web_search" }),
LLMEvent.toolCall({ id: "call-parsed", name: "web_search", input: { query: "hello" }, providerExecuted: true }),
LLMEvent.stepFinish({ index: 0, reason: "stop" }),
LLMEvent.finish({ reason: "stop" }),
]
yield* session.resume(sessionID)
+103 -9
View File
@@ -30,6 +30,7 @@ type OpenApiDocument = {
const document = (await Bun.file("./openapi.json").json()) as OpenApiDocument
const v2Document = (await Bun.file("./openapi-v2.json").json()) as OpenApiDocument
deduplicateEquivalentComponent(v2Document, "Shell", "Shell1")
renameCollidingComponents(document, v2Document)
document.paths = { ...document.paths, ...v2Document.paths }
document.components = {
@@ -106,16 +107,21 @@ if (
) {
throw new Error("Session history generated duplicate Session event variants")
}
const duplicateSessionErrorStart = generatedTypes.indexOf("export type SessionStructuredError2 =")
const duplicateSessionErrorEnd = generatedTypes.indexOf("\n\nexport type ", duplicateSessionErrorStart + 1)
if (duplicateSessionErrorStart === -1 || duplicateSessionErrorEnd === -1) {
throw new Error("Session structured error duplicate prune did not apply")
const sessionErrorTypesPatched = deduplicateEquivalentGeneratedTypes(
generatedTypes,
"SessionStructuredError",
/^SessionStructuredError\d+$/,
)
const obsoleteSessionNext = [...sessionErrorTypesPatched.matchAll(/export type (SessionNext\w*) =/g)].map(
(match) => match[1],
)
if (obsoleteSessionNext.length > 0) {
throw new Error(`Obsolete SessionNext generated type noise reintroduced: ${obsoleteSessionNext.join(", ")}`)
}
const sessionErrorTypesPatched =
generatedTypes.slice(0, duplicateSessionErrorStart) + generatedTypes.slice(duplicateSessionErrorEnd + 2)
const logTypesPatched = sessionErrorTypesPatched
.replaceAll("SessionStructuredError2", "SessionStructuredError")
.replace(/(export type V2SessionLogData = \{[\s\S]*?query\?: \{\s*after\?: )string/, "$1number")
const logTypesPatched = sessionErrorTypesPatched.replace(
/(export type V2SessionLogData = \{[\s\S]*?query\?: \{\s*after\?: )string/,
"$1number",
)
if (logTypesPatched === sessionErrorTypesPatched) {
throw new Error("Session log numeric query patch did not apply")
}
@@ -143,6 +149,12 @@ if (eventSubscribeTypesPatched === sessionMessagesTypesPatched) {
if (/SessionStructuredError\d/.test(eventSubscribeTypesPatched)) {
throw new Error("Session structured error generated a name-mangled duplicate")
}
if (/\bSessionNext\w*\b/.test(eventSubscribeTypesPatched)) {
throw new Error("Obsolete SessionNext generated type noise reintroduced")
}
if (/export type Shell\d+V2 =/.test(eventSubscribeTypesPatched)) {
throw new Error("Shell generated a name-mangled duplicate")
}
await Bun.write(generatedTypesPath, eventSubscribeTypesPatched)
const querySerializerPath = "./src/v2/gen/client/utils.gen.ts"
@@ -234,6 +246,88 @@ function renameCollidingComponents(target: OpenApiDocument, source: OpenApiDocum
source.paths = rewriteRefs(source.paths, renames) as Record<string, unknown> | undefined
}
function deduplicateEquivalentComponent(document: OpenApiDocument, canonical: string, duplicate: string) {
const schemas = document.components?.schemas
if (!schemas?.[canonical] || !schemas[duplicate]) return
if (JSON.stringify(normalizeSchema(schemas[canonical])) !== JSON.stringify(normalizeSchema(schemas[duplicate]))) {
throw new Error(`${duplicate} no longer has the same wire shape as ${canonical}`)
}
const renames = new Map([[duplicate, canonical]])
const rewritten = rewriteRefs(schemas, renames) as Record<string, unknown>
delete rewritten[duplicate]
document.components = { ...document.components, schemas: rewritten }
document.paths = rewriteRefs(document.paths, renames) as Record<string, unknown> | undefined
}
function deduplicateEquivalentGeneratedTypes(source: string, canonical: string, duplicates: RegExp) {
const canonicalType = generatedType(source, canonical)
if (!canonicalType) throw new Error(`Generated canonical type missing: ${canonical}`)
const names = [...source.matchAll(/export type (\w+) =/g)]
.map((match) => match[1])
.filter((name): name is string => name !== undefined && duplicates.test(name))
return names.reduce((patched, name) => {
const duplicate = generatedType(patched, name)
const currentCanonical = generatedType(patched, canonical)
if (!duplicate || !currentCanonical) throw new Error(`Generated type declaration missing while comparing ${name}`)
if (normalizeGeneratedType(currentCanonical.shape) !== normalizeGeneratedType(duplicate.shape)) {
throw new Error(`${name} no longer has the same generated type shape as ${canonical}`)
}
return (patched.slice(0, duplicate.start) + patched.slice(duplicate.end)).replaceAll(name, canonical)
}, source)
}
function generatedType(source: string, name: string) {
const start = source.indexOf(`export type ${name} =`)
if (start === -1) return undefined
const next = source.indexOf("\n\nexport type ", start + 1)
const shapeEnd = next === -1 ? source.length : next
return {
start,
end: next === -1 ? source.length : next + 2,
shape: source.slice(source.indexOf("=", start) + 1, shapeEnd),
}
}
function normalizeGeneratedType(shape: string) {
return shape.replaceAll(/\s/g, "")
}
function normalizeSchema(value: unknown, key?: string): unknown {
if (Array.isArray(value)) {
const flattened =
key === "anyOf"
? value.flatMap((item) =>
typeof item === "object" && item !== null && Object.keys(item).length === 1 && "anyOf" in item
? Array.isArray(item.anyOf)
? item.anyOf
: [item]
: [item],
)
: value
const expanded =
key === "anyOf"
? flattened.flatMap((item) => {
if (typeof item !== "object" || item === null || !("type" in item) || !("enum" in item)) return [item]
if (Object.keys(item).some((property) => property !== "type" && property !== "enum")) return [item]
if (!Array.isArray(item.enum)) return [item]
return item.enum.map((member) => ({ type: item.type, enum: [member] }))
})
: flattened
const normalized = expanded.map((item) => normalizeSchema(item))
return [...new Map(normalized.map((item) => [JSON.stringify(item), item])).values()].sort((a, b) =>
JSON.stringify(a).localeCompare(JSON.stringify(b)),
)
}
if (typeof value !== "object" || value === null) return value
return Object.fromEntries(
Object.entries(value)
.sort(([left], [right]) => left.localeCompare(right))
.map(([property, child]) => [property, normalizeSchema(child, property)]),
)
}
function rewriteRefs(value: unknown, renames: Map<string, string>): unknown {
if (Array.isArray(value)) return value.map((item) => rewriteRefs(item, renames))
if (typeof value !== "object" || value === null) return value
+3 -21
View File
@@ -8715,24 +8715,6 @@ export type SessionSkillActivated2 = {
}
}
export type Shell1V2 = {
id: string
status: "running" | "exited" | "timeout" | "killed"
command: string
cwd: string
shell: string
file: string
pid?: number
exit?: number | "NaN" | "Infinity" | "-Infinity"
metadata: {
[key: string]: unknown
}
time: {
started: number | "NaN" | "Infinity" | "-Infinity"
completed?: number | "NaN" | "Infinity" | "-Infinity"
}
}
export type SessionShellStarted2 = {
id: string
created: number
@@ -8748,7 +8730,7 @@ export type SessionShellStarted2 = {
location?: LocationRef2
data: {
sessionID: string
shell: Shell1V2
shell: ShellV2
}
}
@@ -8767,7 +8749,7 @@ export type SessionShellEnded2 = {
location?: LocationRef2
data: {
sessionID: string
shell: Shell1V2
shell: ShellV2
output: {
output: string
cursor: number
@@ -10729,7 +10711,7 @@ export type ShellCreated2 = {
type: "shell.created"
location?: LocationRef2
data: {
info: Shell1V2
info: ShellV2
}
}
+8
View File
@@ -1,5 +1,13 @@
# V2 Schema Changelog
## 2026-07-04: Canonicalize Generated Shell Type Name
- Collapse the legacy JavaScript SDK generator's equivalent `Shell1V2` component into the canonical `ShellV2` contract.
Compatibility:
- The Shell wire shape is unchanged. Generated event and endpoint types now consistently reference `ShellV2`.
## 2026-07-03: Add Execution Lifecycle, Retry, And Structured Session Errors
- Replace live-only `session.execution.settled` and unused `session.retried` with durable v1 `session.execution.started`, `session.execution.succeeded`, `session.execution.failed`, `session.execution.interrupted`, and `session.retry.scheduled` events.
+6 -2
View File
@@ -47,7 +47,9 @@ SessionExecution.resume(sessionID)
`SessionExecution` and the read-side `SessionStore` are process-global. `SessionRunner`, catalog, model resolver, tool registry, permission state, and filesystem are cached per Location. No layer takes a Session ID. An omitted `Location.workspaceID` means implicit-local placement; explicit workspace identity remains reserved for future placement semantics.
The local runner issues one explicit `llm.stream(request)` per step, projects each complete local tool call durably before eagerly starting its structured child execution, awaits every started tool fiber after provider-stream closure, and reloads projected history once before continuation. Promoting any new user input resets the selected agent's configured step allowance; multiple steers promoted at one boundary reset it once. Tool settlement events carry the owning assistant message ID because provider-local call IDs may repeat across steps. Before assembling a provider request, the runner durably fails any local tool still projected as `running` from a previous process with `Tool execution interrupted`; abandoned side effects are never silently replayed.
The local runner issues one explicit `llm.stream(request)` per step, projects each complete local tool call durably before eagerly starting its structured child execution, awaits every owned tool fiber after provider-stream closure, and reloads projected history once before continuation. For every in-process step, `session.step.started` precedes its tool calls, every local and hosted call settles as `session.tool.success` or `session.tool.failed`, and only then may the runner publish the single terminal `session.step.ended` or `session.step.failed`. Streamed provider-error evidence is retained until this closeout; thrown provider failures and interruption use the same settlement-first ordering. Promoting any new user input resets the selected agent's configured step allowance; multiple steers promoted at one boundary reset it once.
`callID` is unique only within its owning step, not across the Session. Tool events therefore carry `assistantMessageID`, and consumers correlate a call through the step that owns that assistant message rather than inventing a synthetic composite key. Before assembling a provider request, the runner's cross-drain `failInterruptedTools` recovery durably fails any tool still projected as pending or running from a previous process with `Tool execution interrupted`. This orphan-recovery sweep is the explicit nesting exception: it occurs in a later drain, but attributes every settlement to the original `assistantMessageID`; abandoned side effects are never silently replayed.
`session.execution.started.1` and exactly one of `session.execution.succeeded.1`, `session.execution.failed.1`, or `session.execution.interrupted.1` observe one process-local coordinator busy period, including coalesced drains and joined resumes. These durable rows are history, not a durable execution identity: replay must never infer current liveness, recovery, grouping, or resumability from an unmatched start. A drain has no durable identity or transcript boundary. `/api/session/active` is the authority for current process-local liveness, and is empty after restart. User interruption records `reason: "user"`; owner-scope interruption defaults to `"shutdown"`; `"superseded"` is reserved for explicit replacement.
@@ -120,9 +122,11 @@ Before each step, the runner estimates the complete model-visible request and co
Compaction keeps the full transcript durable while replacing its active model representation with one hidden checkpoint containing a structured rolling summary and token-bounded serialized recent context. Provider-native assistant, reasoning, and tool messages never survive across the boundary, avoiding signature and encrypted-reasoning failures when the earlier prefix changes.
The rolling summary is a continuation checkpoint with this complete heading order: `Continuation Goal`, `Operating Constraints`, `Progress` (`Completed`, `In Flight`, `Blocked`), `Decisions To Preserve`, `Resume From Here`, `Context To Preserve`, and `Working Files`. Every heading remains present even when its value is `(none)`.
`session.compaction.started.1` durably identifies the attempt. Compaction deltas are live-only progress. `session.compaction.ended.1` durably stores the final summary and serialized recent context; only this completed event projects a model-visible compaction message. On the next physical attempt, the runner observes that completed compaction and directly renders a fresh Context Epoch baseline. A failed or interrupted attempt therefore leaves the previous history boundary active.
Assistant text and reasoning follow a strict `started` / live-only `delta` / durable full-value `ended` lifecycle. A publisher permits at most one open fragment of each kind in a step and fails on a second start before the matching end. Provider block IDs remain internal to LLM adapters; projected UI identity is the assistant message ID plus content kind and ordinal. Tool calls retain `callID` because settlements and provider replay correlate through it.
Assistant text and reasoning follow a strict `started` / live-only `delta` / durable full-value `ended` lifecycle. A publisher permits at most one open fragment of each kind in a step and fails on a second start before the matching end. Provider block IDs remain internal to LLM adapters; projected UI identity is the assistant message ID plus content kind and ordinal. Tool calls retain step-scoped `callID` because settlements and provider replay correlate through it.
Provider continuation state is opaque and un-nested at the Session boundary. The publisher selects only the active model provider's entry from LLM provider metadata. Same-model replay re-nests that state under the current provider; model switches and failed assistant steps continue to suppress provider-native continuation state.
+2 -3
View File
@@ -24,6 +24,8 @@ through legacy `SessionPrompt.loop(...)`:
and issues one explicit `llm.stream(request)` step at a time
- durable V2 projections record text, reasoning, provider failures, tool calls,
tool results, and assistant output
- owned local tool fibers and unresolved hosted calls settle before their step's
single terminal event; cross-drain orphan recovery retains original assistant attribution
- a scoped `ToolRegistry` advertises definitions and the first permission-checked
`read` built-in
- local continuation reloads projected history, and promoting new user input resets the selected agent's configured step allowance
@@ -37,9 +39,6 @@ a FIFO until the Session would otherwise become idle and then promote one at a t
Next reviewed slices:
- preserve eager structured local-tool settlement: durably record each complete
call, start its child execution immediately, await every settlement after the
step closes, then reload projected history once
- revisit per-step tool-call limits, output truncation, and operational
backpressure before broadening exposure; eager local execution is deliberately
unbounded in the current local slice while SQLite publication stays serialized