Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 8 additions & 8 deletions packages/core/src/session/compaction.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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>

Expand Down
148 changes: 86 additions & 62 deletions packages/core/src/session/runner/llm.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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)
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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())),
Expand Down Expand Up @@ -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()) {
Expand All @@ -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) {
Expand All @@ -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",
}),
)

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()
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)
// 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",
Expand Down
24 changes: 16 additions & 8 deletions packages/core/src/session/runner/publish-llm-event.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -206,26 +206,32 @@ 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,
})
})

const failUnsettledTools = Effect.fn("SessionRunner.failUnsettledTools")(function* (
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,
Expand All @@ -234,6 +240,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
executed: tool.providerExecuted,
})
}
return failed
})

const assistantMessageIDForTool = (callID: string) => {
Expand Down Expand Up @@ -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) }
Expand All @@ -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
}
})
Expand All @@ -414,6 +421,7 @@ export const createLLMEventPublisher = (events: EventV2.Interface, input: Input)
publish,
flush,
failAssistant,
publishStepFailure,
failUnsettledTools,
hasProviderError: () => providerFailed,
hasRetryEvidence: () => retryEvidence,
Expand Down
26 changes: 26 additions & 0 deletions packages/core/test/session-compaction.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -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 = []
Expand Down
Loading
Loading