From 2602dcd0a71f0bc221bc85913e0585cc44301e3c Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Wed, 26 Aug 2026 12:41:45 -0400 Subject: [PATCH] refactor(core): encapsulate physical attempt execution (#45294) Extract physical-attempt streaming, tool execution, and durable settlement from the Session runner. Preserve logical-Step retry and recovery policy, use tagged drain outcomes, and fix multi-click selection during auto-copy. --- packages/core/src/session/execution.ts | 2 +- packages/core/src/session/model-request.ts | 2 +- packages/core/src/session/runner/index.ts | 10 +- packages/core/src/session/runner/llm.ts | 592 +++--------------- .../src/session/runner/publish-llm-event.ts | 49 +- packages/core/src/session/runner/retry.ts | 28 +- packages/core/src/session/runner/step.ts | 293 +++++++++ packages/core/test/session-execution.test.ts | 4 +- .../test/session-runner-tool-events.test.ts | 2 +- packages/core/test/session-runner.test.ts | 122 +++- packages/core/test/session-step.test.ts | 140 +++++ packages/tui/src/util/selection.ts | 7 +- .../util/selection-copy-on-select.test.tsx | 1 + packages/tui/test/util/selection.test.ts | 21 +- 14 files changed, 700 insertions(+), 573 deletions(-) create mode 100644 packages/core/src/session/runner/step.ts create mode 100644 packages/core/test/session-step.test.ts diff --git a/packages/core/src/session/execution.ts b/packages/core/src/session/execution.ts index 1796ae42a2c..dd256b5cf1d 100644 --- a/packages/core/src/session/execution.ts +++ b/packages/core/src/session/execution.ts @@ -94,7 +94,7 @@ export const layer = Layer.effect( : Effect.logError("Failed to drain Session", cause).pipe(Effect.annotateLogs({ sessionID })), ), ) - if (result.type === "complete") return + if (result._tag === "Complete") return return yield* drain(sessionID, false, result.continuation, promotable) }) } diff --git a/packages/core/src/session/model-request.ts b/packages/core/src/session/model-request.ts index 7881c330417..2d818ebbf3e 100644 --- a/packages/core/src/session/model-request.ts +++ b/packages/core/src/session/model-request.ts @@ -45,7 +45,7 @@ const declineDefect = (cause: Cause.Cause) => { return decline ? Result.succeed(decline) : Result.fail(cause) } -interface Prepared { +export interface Prepared { readonly request: LLMRequest readonly options: StreamOptions /** diff --git a/packages/core/src/session/runner/index.ts b/packages/core/src/session/runner/index.ts index 4525918586d..38cb3afdef4 100644 --- a/packages/core/src/session/runner/index.ts +++ b/packages/core/src/session/runner/index.ts @@ -1,7 +1,7 @@ export * as SessionRunner from "./index.js" import type { AIError } from "@opencode-ai/ai" -import { Context, Effect } from "effect" +import { Context, Data, Effect } from "effect" import { SessionSchema } from "../schema.js" import type { Promotable } from "../inbox.js" import type { AgentNotFoundError, MessageDecodeError, StepFailedError, UserInterruptedError } from "../error.js" @@ -19,9 +19,11 @@ export type RunError = export type Continuation = { readonly step: number } -export type DrainResult = - | { readonly type: "complete" } - | { readonly type: "moved"; readonly continuation?: Continuation } +export type DrainResult = Data.TaggedEnum<{ + Complete: {} + Moved: { readonly continuation?: Continuation } +}> +export const DrainResult = Data.taggedEnum() /** Runs one local continuation from already-recorded Session history. */ export interface Interface { diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 7420225c9c1..205f021d537 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -1,20 +1,9 @@ export * as SessionRunnerLLM from "./llm.js" -import { - LLMClient, - AIError, - InvalidProviderOutputReason, - LLMEvent, - Message, - isContextOverflowFailure, - type ProviderErrorEvent, - type ToolCall, -} from "@opencode-ai/ai" -import { Cause, Config, Data, Effect, Exit, Fiber, FiberMap, Layer, Option, Pull, Schedule, Stream } from "effect" +import { Message } from "@opencode-ai/ai" +import { Cause, Config, Effect, Exit, FiberMap, Layer, Pull, Schedule } from "effect" import { Database } from "../../database/database.js" import { Bus } from "../../bus.js" -import { Permission } from "../../permission.js" -import { QuestionTool } from "../../tool/plugin/question.js" import { InstructionState } from "../instruction-state.js" import { SessionCompaction } from "../compaction.js" import { SessionContext } from "../context.js" @@ -26,100 +15,18 @@ import { SessionMessage } from "../message.js" import { SessionSchema } from "../schema.js" import { SessionStore } from "../store.js" import { SessionTitle } from "../title.js" -import { Service, type Continuation } from "./index.js" -import { createLLMEventPublisher, type StepRecord } from "./publish-llm-event.js" +import { DrainResult, Service, type Continuation } from "./index.js" import { Snapshot } from "../../snapshot.js" import { makeLocationNode } from "@opencode-ai/util/effect/app-node" import { llmClient } from "../../effect/app-node-platform.js" import { StepFailedError } from "../error.js" -import { toSessionError } from "../to-session-error.js" import { SessionRunnerRetry } from "./retry.js" -import { SessionUsage } from "../usage.js" +import { SessionStep } from "./step.js" import { ToolOutput } from "../../tool-output.js" import { PluginSupervisor } from "../../plugin/supervisor.js" -import { Tool } from "../../tool.js" import { PromptCacheDiagnostics } from "../prompt-cache-diagnostics.js" import { MAX_STEPS_PROMPT } from "./max-steps.js" -/** How one model call ended: settled, awaiting retry/recovery, or restarted by compaction. */ -type CallOutcome = Data.TaggedEnum<{ - Completed: { readonly needsContinuation: boolean; readonly step: number } - Retry: { readonly step: number } - Continue: { - readonly cause: AIError - readonly error: SessionRunnerRetry.RetryableFailure["error"] - readonly step: number - } - RecoverFull: { readonly step: number } - Restart: { readonly step: number; readonly recoveredOverflow: boolean } -}> -const CallOutcome = Data.taggedEnum() - -// Declining an interactive prompt halts the drain instead of becoming model-facing tool output. -const isDecline = ( - error: SessionModelRequest.ExecuteError, -): error is Permission.DeclinedError | QuestionTool.CancelledError => - error._tag === "Permission.DeclinedError" || error._tag === "QuestionTool.CancelledError" - -const isInterruptedStream = (failure: AIError) => { - if (failure.reason._tag === "InvalidProviderOutput") - return failure.reason.classification === "incomplete-stream" - if (failure.reason._tag === "Transport") return failure.reason.operation === "read" - return false -} - -/** - * Classifies how the owned tool fibers ended. Interrupts abort the step; a user decline - * settles its own call and then aborts the step; a defect from a tool implementation - * becomes a failed tool call the model can read; a typed infrastructure failure must - * fail the assistant and then the drain. - */ -const classifyToolExits = ( - settled: Exit.Exit>, never>, - calls: ReadonlyArray, -) => { - // Exits align with calls by construction: one owned fiber per accepted local call. - const exits = settled._tag === "Success" ? settled.value : [] - const declines = exits.flatMap((exit, index) => - exit._tag === "Failure" - ? exit.cause.reasons.flatMap((reason) => - Cause.isFailReason(reason) && isDecline(reason.error) ? [{ call: calls[index], reason: reason.error }] : [], - ) - : [], - ) - const causes = - settled._tag === "Failure" - ? [settled.cause] - : exits.flatMap((exit) => (exit._tag === "Failure" ? [exit.cause] : [])) - // The first non-interrupt, non-decline failure, rebuilt without decline reasons so the - // drain's error channel never carries a decline. - const failure = causes - .flatMap((cause) => { - if (Cause.hasInterrupts(cause)) return [] - const reasons = cause.reasons.flatMap( - (reason): Array> => - Cause.isFailReason(reason) - ? isDecline(reason.error) - ? [] - : // A typed failure here broke the ExecuteError contract (the per-fiber - // `catchTag("Tool.Error")` consumes honest ones). Surfacing it as a defect - // keeps it from being dropped, which would leave its call unsettled forever. - [Cause.makeDieReason(reason.error)] - : [reason], - ) - return reasons.length > 0 ? [Cause.fromReasons(reasons)] : [] - }) - .at(0) - return { - interrupted: causes.some(Cause.hasInterrupts), - declines, - failure, - } -} - -const TOOLS_INTERRUPTED = { type: "aborted", message: "Tool execution interrupted" } as const -const STEP_INTERRUPTED = { type: "aborted", message: "Step interrupted" } as const -const RESULT_MISSING = { type: "tool.result-missing", message: "Provider did not return a tool result" } as const const CONTINUE_AFTER_INCOMPLETE_STREAM = "The previous response was interrupted. Continue from where you left off without repeating completed content." @@ -127,17 +34,15 @@ const layer = Layer.effect( Service, Effect.gen(function* () { const bus = yield* Bus.Service - const llm = yield* LLMClient.Service const store = yield* SessionStore.Service const context = yield* SessionContext.Service const modelRequests = yield* SessionModelRequest.Service const modelTransport = yield* SessionModelTransport.Service - const snapshots = yield* Snapshot.Service const db = (yield* Database.Service).db const compaction = yield* SessionCompaction.Service const plugins = yield* PluginSupervisor.Service const title = yield* SessionTitle.Service - const toolOutput = yield* ToolOutput.Service + const steps = yield* SessionStep.make const diagnostics = yield* Config.boolean("OPENCODE_PROMPT_CACHE_DIAGNOSTICS").pipe( Config.withDefault(false), Effect.orDie, @@ -166,10 +71,7 @@ const layer = Layer.effect( }) // Title generation starts once input is visible and must not delay model execution. const titles = yield* FiberMap.make() - /** - * Drains eligible manual compaction and user input until the Session becomes idle. - * Execution lifecycle is published per busy period by SessionExecution, not here. - */ + const drain = Effect.fn("SessionRunner.drain")(function* (input: { readonly sessionID: SessionSchema.ID readonly force: boolean @@ -179,30 +81,25 @@ const layer = Layer.effect( let force = input.force let continuation = input.continuation const promotable = input.promotable ?? "input" - if (!force && !continuation && !(yield* eligible(input.sessionID, promotable))) - return { type: "complete" as const } + if (!force && !continuation && !(yield* eligible(input.sessionID, promotable))) return DrainResult.Complete() yield* plugins.flush yield* settleStaleToolCalls(input.sessionID) while (true) { - // Between-turn control items run under any drain scope: scope gates which user - // input may promote, not whether admitted housekeeping runs. Steered control - // items go ahead of any queued input; only a queue-delivered control item - // parked behind a queued prompt is not the next eligible item. + // Scope gates input promotion, not a between-step control that is next in line. if (yield* runPendingCompaction(input.sessionID, "input")) { force = false continue } - if (yield* runPendingMove(input.sessionID, "input")) return { type: "moved" as const } + if (yield* runPendingMove(input.sessionID, "input")) return DrainResult.Moved({}) if (!force && !continuation && !(yield* SessionInbox.has(db, input.sessionID, promotable))) - return { type: "complete" as const } + return DrainResult.Complete() const result = yield* runSteps(input.sessionID, continuation, promotable) - if (result.type === "moved") return result + if (result._tag === "Moved") return result force = false continuation = undefined } }) - /** Work this drain may perform: scoped input, or a between-turn control item next in line. */ const eligible = Effect.fnUntraced(function* (sessionID: SessionSchema.ID, promotable: SessionInbox.Promotable) { if (yield* SessionInbox.has(db, sessionID, promotable)) return true if (promotable === "input") return false @@ -210,31 +107,20 @@ const layer = Layer.effect( return next?.type === "compaction" || next?.type === "move" }) - /** - * Runs logical steps until no tool result or newly admitted steer requires another - * model call. Queued inputs remain pending until the current model work reaches idle. - */ + /** Queued inputs wait until the current model work reaches idle; later Steps absorb only steers. */ const runSteps = Effect.fn("SessionRunner.runSteps")(function* ( sessionID: SessionSchema.ID, continuation: Continuation | undefined, drainPromotable: SessionInbox.Promotable, ) { - // Fresh work may promote queued input; resumed turns and later steps absorb steers only. let promotable: SessionInbox.Promotable = continuation ? "steer" : drainPromotable let step = continuation?.step ?? 1 let next = continuation - // The drain admitted this work, so the first step always runs — even after a - // control item consumed at this boundary (unlike drain's one-shot force). let first = true - // Every boundary has the same shape: control items first, then one exit decision, - // then the model. The turn continues only while the first step, a continuation, or - // steer input is owed. Deciding after control items means consuming the last - // steered compaction ends the turn instead of issuing an input-free model call. while (true) { if (yield* runPendingCompaction(sessionID, "steer")) continue - if (yield* runPendingMove(sessionID, "steer")) return { type: "moved" as const, continuation: next } - if (!first && !next && !(yield* SessionInbox.has(db, sessionID, "steer"))) - return { type: "complete" as const } + if (yield* runPendingMove(sessionID, "steer")) return DrainResult.Moved({ continuation: next }) + if (!first && !next && !(yield* SessionInbox.has(db, sessionID, "steer"))) return DrainResult.Complete() const result = yield* runStep(sessionID, promotable, step) first = false promotable = "steer" @@ -243,391 +129,100 @@ const layer = Layer.effect( } }) - /** Completes one logical model step, transparently retrying or rebuilding after compaction. */ - const runStep = Effect.fnUntraced(function* ( + /** Owns logical Step policy; each attempt owns its streaming, tools, and durable settlement. */ + const runStep = Effect.fn("SessionRunner.runStep")(function* ( sessionID: SessionSchema.ID, promotable: SessionInbox.Promotable, step: number, ) { - // Minting message identity before any attempt lets retries resume the same durable - // message. A compaction restart re-mints: the old message is stranded behind the new - // compaction boundary, so the rebuilt step needs identity inside the new epoch. let assistantMessageID = SessionMessage.ID.create() - const retry = yield* Schedule.toStepWithSleep( - SessionRunnerRetry.schedule(bus, sessionID, () => assistantMessageID), - ) - /** - * Consumes one retry allowance: sleeps the scheduled backoff, or publishes - * Step.Failed and fails once attempts are exhausted. The step loop performs - * the retry itself on the next iteration. - */ - const waitForRetry = (failure: SessionRunnerRetry.RetryableFailure) => - retry(failure).pipe( - Effect.as(CallOutcome.Retry({ step: failure.step })), - Pull.catchDone(() => - bus - .publish(SessionEvent.Step.Failed, { - sessionID, - assistantMessageID, - error: failure.error, - }) - .pipe(Effect.andThen(Effect.fail(failure.cause))), - ), - ) + const retry = yield* Schedule.toStepWithSleep(SessionRunnerRetry.schedule(bus, sessionID)) let currentPromotable: SessionInbox.Promotable | undefined = promotable let currentStep = step - // Overflow recovery is one-shot: a call after recovery must not recover another overflow. let recoverOverflow = true - // Continuation rejection permits one immediate full-context Physical Attempt without generic backoff. let recoverContinuation = true while (true) { - const outcome = yield* callModel( - sessionID, - currentPromotable, - currentStep, - recoverOverflow, - recoverContinuation, - assistantMessageID, - ).pipe(Effect.catchTag("SessionRunner.RetryableFailure", waitForRetry)) - if (outcome._tag === "Completed") return { needsContinuation: outcome.needsContinuation, step: outcome.step } - if (outcome._tag === "Continue") { - yield* retry( - new SessionRunnerRetry.RetryableFailure({ - cause: outcome.cause, - error: outcome.error, - step: outcome.step, - }), - ).pipe(Pull.catchDone(() => Effect.fail(outcome.cause))) - yield* bus.publish(SessionEvent.Synthetic, { - sessionID, - text: CONTINUE_AFTER_INCOMPLETE_STREAM, + const selected = yield* context.select(sessionID) + // A blocked initial instruction baseline must leave admitted input pending. + yield* InstructionState.prepare(db, bus, selected.instructions, selected.session.id) + const promoted = currentPromotable + ? yield* SessionInbox.promote(db, bus, selected.session.id, currentPromotable) + : 0 + if (promoted > 0 && !selected.session.parentID && SessionTitle.isUntitled(selected.session)) + yield* FiberMap.run(titles, sessionID, title.generate(sessionID).pipe(Effect.ignore), { + onlyIfMissing: true, }) - assistantMessageID = SessionMessage.ID.create() - } - if (outcome._tag === "Restart") { - if (outcome.recoveredOverflow) recoverOverflow = false - assistantMessageID = SessionMessage.ID.create() - } - if (outcome._tag === "RecoverFull") recoverContinuation = false - // Neither a retry nor a compaction restart re-promotes input. + currentStep = promoted > 0 ? 1 : currentStep currentPromotable = undefined - currentStep = outcome.step + const loaded = yield* context.load(selected) + const compactionInput = { session: loaded.session, messages: loaded.messages, resolved: loaded.model } + if (compaction.required(compactionInput)) { + const compacted = yield* compaction.compact(compactionInput) + if (compacted.status !== "completed") return yield* new StepFailedError({ error: compacted.error }) + assistantMessageID = SessionMessage.ID.create() + continue + } + const stepLimitReached = loaded.agent.info.steps !== undefined && currentStep >= loaded.agent.info.steps + const transcript = SessionModelRequest.baseTranscript({ + agent: loaded.agent.info, + model: loaded.model, + tools: loaded.tools, + initial: loaded.initial, + messages: loaded.messages, + }) + const prepared = yield* modelRequests.prepare({ + scope: { session: loaded.session, agentID: loaded.agent.id, model: loaded.model, tools: loaded.tools }, + transcript: { + system: transcript.system, + messages: stepLimitReached + ? [...transcript.messages, Message.assistant(MAX_STEPS_PROMPT)] + : transcript.messages, + }, + // Keep tool definitions on the final Step to preserve the provider's cached prefix. + toolChoice: stepLimitReached ? "none" : undefined, + webSocket: "session", + }) + yield* diagnosePromptCache(sessionID, prepared.request) + const outcome = yield* steps.attempt({ + sessionID, + assistantMessageID, + agent: loaded.agent.id, + model: loaded.model, + prepared, + toolsDisabled: stepLimitReached, + recoverContinuation, + recoverOverflow: Effect.suspend(() => + recoverOverflow && compaction.enabled() + ? compaction.compact(compactionInput).pipe(Effect.map((result) => result.status === "completed")) + : Effect.succeed(false), + ), + }) + if (outcome._tag === "Completed") return { needsContinuation: outcome.needsContinuation, step: currentStep } + if (outcome._tag === "Retry" || outcome._tag === "Continue") { + yield* retry({ cause: outcome.cause, error: outcome.error, assistantMessageID }).pipe( + Pull.catchDone(() => + Effect.gen(function* () { + if (outcome._tag === "Retry") + yield* bus.publish(SessionEvent.Step.Failed, { sessionID, assistantMessageID, error: outcome.error }) + return yield* outcome.cause + }), + ), + ) + if (outcome._tag === "Continue") { + yield* bus.publish(SessionEvent.Synthetic, { sessionID, text: CONTINUE_AFTER_INCOMPLETE_STREAM }) + assistantMessageID = SessionMessage.ID.create() + } + continue + } + if (outcome._tag === "Compacted") { + recoverOverflow = false + assistantMessageID = SessionMessage.ID.create() + continue + } + recoverContinuation = false } }) - /** - * Prepares and runs at most one model call, executes its local tools, and durably - * settles the step. Compaction may instead request that the logical step restart. - */ - const callModel = Effect.fn("SessionRunner.callModel")(function* ( - sessionID: SessionSchema.ID, - promotable: SessionInbox.Promotable | undefined, - step: number, - recoverOverflow: boolean, - recoverContinuation: boolean, - assistantMessageID: SessionMessage.ID, - ) { - const selected = yield* context.select(sessionID) - // Establish what the model knows before admitting what the user said, so - // a blocked first step leaves pending inputs untouched. - yield* InstructionState.prepare(db, bus, selected.instructions, selected.session.id) - const promoted = promotable ? yield* SessionInbox.promote(db, bus, selected.session.id, promotable) : 0 - if (promoted > 0 && !selected.session.parentID && SessionTitle.isUntitled(selected.session)) - yield* FiberMap.run(titles, sessionID, title.generate(sessionID).pipe(Effect.ignore), { - onlyIfMissing: true, - }) - // Promoted input opens a fresh step allowance. - const currentStep = promoted > 0 ? 1 : step - const loaded = yield* context.load(selected) - const { session, agent } = loaded - const resolved = loaded.model - // Make room: history must fit the context window before the call. A pending manual - // compaction owns this instead; the runner executes it between steps. - const compactionInput = { session, messages: loaded.messages, resolved } - if (compaction.required(compactionInput)) { - const compacted = yield* compaction.compact(compactionInput) - if (compacted.status === "completed") - return CallOutcome.Restart({ step: currentStep, recoveredOverflow: false }) - return yield* new StepFailedError({ error: compacted.error }) - } - const stepLimitReached = agent.info.steps !== undefined && currentStep >= agent.info.steps - const transcript = SessionModelRequest.baseTranscript({ - agent: agent.info, - model: resolved, - tools: loaded.tools, - initial: loaded.initial, - messages: loaded.messages, - }) - const prepared = yield* modelRequests.prepare({ - scope: { session, agentID: agent.id, model: resolved, tools: loaded.tools }, - transcript: { - system: transcript.system, - messages: stepLimitReached - ? [...transcript.messages, Message.assistant(MAX_STEPS_PROMPT)] - : transcript.messages, - }, - // The final Step keeps definitions available to protocols with native "none", - // preserving their prompt cache prefix. Calls are still rejected at execution. - toolChoice: stepLimitReached ? "none" : undefined, - webSocket: "session", - }) - yield* diagnosePromptCache(session.id, prepared.request) - const executeTool = (input: Parameters[0]) => { - if (stepLimitReached) return new Tool.Error({ message: "Tools are disabled after the maximum agent steps" }) - return prepared.executeTool(input) - } - // Every local tool call forked here is owned until it reaches one durable settlement. - const toolRuns: Array<{ - readonly call: ToolCall - readonly fiber: Fiber.Fiber - }> = [] - const interruptTools = Effect.suspend(() => Fiber.interruptAll(toolRuns.map((run) => run.fiber))) - const startSnapshot = yield* snapshots.capture() - const publisher = createLLMEventPublisher(bus, { - sessionID: session.id, - agent: agent.id, - // The selected catalog identity, not model.id: route-level ids are provider API - // model ids (for example gpt-5.5-fast resolves to api id gpt-5.5). - model: resolved.ref, - providerMetadataKey: transcript.providerMetadataKey, - snapshot: startSnapshot, - assistantMessageID, - }) - const stepUsage = (finish: NonNullable) => ({ - cost: SessionUsage.calculateCost(resolved.cost, finish.tokens), - tokens: finish.tokens, - }) - - const captureStepEnd = Effect.fnUntraced(function* () { - const snapshot = yield* snapshots.capture() - const files = - startSnapshot && snapshot - ? startSnapshot === snapshot - ? [] - : yield* snapshots - .files({ from: startSnapshot, to: snapshot }) - .pipe(Effect.orElseSucceed(() => undefined)) - : undefined - return { snapshot, files } - }) - - const publishStepEnd = (finish: NonNullable) => - Effect.gen(function* () { - const end = yield* captureStepEnd() - yield* bus.publish(SessionEvent.Step.Ended, { - sessionID: session.id, - assistantMessageID: yield* publisher.startAssistant(), - finish: finish.finish, - rawFinish: finish.rawFinish, - providerState: finish.providerState, - ...stepUsage(finish), - ...end, - }) - }) - - // Concurrent writers, no lock: the provider loop and each tool fiber publish - // durable events unserialized. This is safe because every publisher method commits - // its state marks synchronously before its first await (see publish-llm-event.ts), - // every required event order is per-source (each source is one sequential fiber), - // and a fiber's events are causally after its own Tool.Called: the fork happens - // below that publish. Cross-source order is unconstrained; either interleaving is - // a truthful history of concurrent work. - // - // The stream is defined here but runs inside the settlement mask below: publish each - // event durably, fork one fiber per local tool call, and hold back a virgin - // context-overflow provider error so settlement may recover it via compaction. - let overflowFailure: ProviderErrorEvent | undefined - const providerStream = llm.stream(prepared.request, prepared.options).pipe( - Stream.runForEach((event) => - Effect.gen(function* () { - if (overflowFailure || publisher.hasProviderError()) return - if ( - LLMEvent.is.providerError(event) && - isContextOverflowFailure(event) && - !publisher.record().outputStarted - ) { - overflowFailure = event - return - } - yield* publisher.publish(event) - if (event.type !== "tool-call" || event.providerExecuted) return - const assistantMessageID = yield* publisher.assistantMessageID(event.id) - toolRuns.push({ - call: event, - fiber: yield* Effect.uninterruptibleMask((restore) => - restore( - executeTool({ - sessionID: session.id, - agent: agent.id, - messageID: assistantMessageID, - call: event, - // Progress is ephemeral, not durable history: nothing to order. - progress: (update) => publisher.progress(event.id, update), - }), - ).pipe( - // The fiber owns its call: it publishes its own completion, masked so a - // finished execution always reaches its durable settlement. - Effect.flatMap(toolOutput.truncate), - Effect.flatMap((outcome) => publisher.toolExecution(event.id, event.name, outcome)), - Effect.catchTag("Tool.Error", (error) => - publisher.failTool(event.id, toSessionError(error), error.metadata).pipe(Effect.asVoid), - ), - ), - ).pipe(Effect.forkScoped), - }) - }), - ), - Effect.ensuring(publisher.flush()), - ) - - // Settle: only the stream and the fiber joins are interruptible (restore); every - // other line is protected so a started call always reaches one durable outcome. - return yield* Effect.uninterruptibleMask((restore) => - Effect.gen(function* () { - const stream = yield* restore(providerStream).pipe(Effect.exit) - const streamFailure = Option.getOrUndefined(Exit.findErrorOption(stream)) - // Note: Exit.hasInterrupts is a type guard whose false branch unsoundly narrows - // away non-interrupt failures, so both interrupt checks stay Cause-based. - const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) - if (!overflowFailure && publisher.hasStarted()) yield* publisher.streamed() - - // Join every owned tool run first: await all exits, not just the first failure. - // Afterwards no fiber is alive, settlement is the only writer, and the record - // is final. A failed join means the waiting itself was interrupted, so the runs - // we abandoned are interrupted before settlement closes them out. - if (streamInterrupted) yield* interruptTools - const joined = yield* restore( - Effect.forEach(toolRuns, (run) => Fiber.await(run.fiber), { concurrency: "unbounded" }), - ).pipe(Effect.exit) - if (joined._tag === "Failure") yield* interruptTools - const tools = classifyToolExits( - joined, - toolRuns.map((run) => run.call), - ) - - // A context overflow before any assistant output is recoverable: compact and - // restart the step instead of surfacing the provider error. - if ( - recoverOverflow && - compaction.enabled() && - !publisher.record().outputStarted && - isContextOverflowFailure(overflowFailure ?? streamFailure) && - (yield* restore(compaction.compact(compactionInput))).status === "completed" - ) - return CallOutcome.Restart({ step: currentStep, recoveredOverflow: true }) - - // An unrecovered held-back overflow becomes the step's durable provider error. - if (overflowFailure) yield* publisher.publish(overflowFailure) - // A thrown LLM failure not already recorded as the provider error either - // escapes as a scheduled retry or fails the assistant durably. - const unknownFinish = - stream._tag === "Success" && publisher.record().finish?.finish === "unknown" - ? new AIError({ - module: "session", - method: "stream", - reason: new InvalidProviderOutputReason({ - classification: "incomplete-stream", - message: "The provider response ended with an unknown finish reason.", - }), - }) - : undefined - const llmFailure = streamFailure instanceof AIError ? streamFailure : unknownFinish - const llmError = llmFailure && !publisher.record().providerFailed ? toSessionError(llmFailure) : undefined - if ( - recoverContinuation && - llmFailure?.reason._tag === "Transport" && - (llmFailure.reason.recovery === "retry-full" || llmFailure.reason.recovery === "rotate-and-retry-full") && - !publisher.record().outputStarted - ) - return CallOutcome.RecoverFull({ step: currentStep }) - if ( - llmFailure && - llmError && - SessionRunnerRetry.isRetryable(llmFailure) && - !publisher.record().outputStarted - ) { - // RetryScheduled and Step.Failed fold onto an existing assistant message, so - // Step.Started must be durable before the failure escapes. - yield* publisher.startAssistant() - return yield* new SessionRunnerRetry.RetryableFailure({ - cause: llmFailure, - error: llmError, - step: currentStep, - }) - } - if (llmError) yield* publisher.failAssistant(llmError) - - // Close every unsettled call with the reason it could not settle truthfully, - // and fail the assistant when the step itself cannot complete. A declined call - // settles with its own reason before the generic sweeps. - for (const decline of tools.declines) - yield* publisher.failTool(decline.call.id, { - type: "aborted", - message: - decline.reason._tag === "QuestionTool.CancelledError" - ? decline.reason.message - : "The user declined this tool call", - }) - if (tools.declines.length > 0 || streamInterrupted || tools.interrupted) { - yield* publisher.failUnsettledTools(TOOLS_INTERRUPTED) - yield* publisher.failAssistant(STEP_INTERRUPTED) - } - if (tools.failure !== undefined) { - const error = toSessionError(Cause.squash(tools.failure)) - yield* publisher.failUnsettledTools(error) - } - // Local calls have joined, so the remaining sweeps only close hosted calls the - // provider promised but never resolved. - if (publisher.record().providerFailed) yield* publisher.failUnsettledTools(TOOLS_INTERRUPTED) - if (llmError) yield* publisher.failUnsettledTools(RESULT_MISSING, "hosted") - // A clean stream that still left hosted calls unresolved fails the step itself. - if (stream._tag === "Success" && !publisher.record().providerFailed) { - const hostedResultMissing = yield* publisher.failUnsettledTools(RESULT_MISSING, "hosted") - if (hostedResultMissing && !publisher.record().finish) yield* publisher.failAssistant(RESULT_MISSING) - } - - // One terminal event: Step.Ended on a clean finish, Step.Failed otherwise. - const record = publisher.record() - if (record.finish && !record.failure) yield* publishStepEnd(record.finish) - if (record.failure) { - const end = yield* captureStepEnd() - yield* publisher.publishStepFailure({ - ...(record.finish ? stepUsage(record.finish) : {}), - ...end, - }) - } - - if ( - llmFailure && - llmError && - isInterruptedStream(llmFailure) && - record.outputStarted && - tools.declines.length === 0 && - !tools.interrupted - ) - return CallOutcome.Continue({ - cause: llmFailure, - error: llmError, - step: currentStep, - }) - - if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause) - if (tools.declines.length > 0) return yield* Effect.interrupt - if (tools.interrupted && tools.failure) return yield* Effect.failCause(tools.failure) - if (tools.interrupted && joined._tag === "Failure") return yield* Effect.failCause(joined.cause) - if (record.failure) return yield* new StepFailedError({ error: record.failure }) - return CallOutcome.Completed({ - // A local call or malformed tool input requires another model step, unless - // this step already exhausted the agent's allowance. - needsContinuation: - !stepLimitReached && record.calls.some((call) => !call.providerExecuted && (call.called || call.settled)), - step: currentStep, - }) - }), - ) - }, Effect.scoped) - - /** Executes a previously admitted manual compaction request, if one is pending. */ const runPendingCompaction = Effect.fn("SessionRunner.runPendingCompaction")(function* ( sessionID: SessionSchema.ID, promotable: SessionInbox.Promotable, @@ -699,7 +294,6 @@ const layer = Layer.effect( ) }) - /** Closes stale tool calls left active by an earlier interrupted drain. */ const settleStaleToolCalls = Effect.fn("SessionRunner.settleStaleToolCalls")(function* ( sessionID: SessionSchema.ID, ) { diff --git a/packages/core/src/session/runner/publish-llm-event.ts b/packages/core/src/session/runner/publish-llm-event.ts index 336d449bcca..d50a9578fa9 100644 --- a/packages/core/src/session/runner/publish-llm-event.ts +++ b/packages/core/src/session/runner/publish-llm-event.ts @@ -1,5 +1,5 @@ import { type LLMEvent, type ProviderMetadata, type ToolResultValue } from "@opencode-ai/ai" -import { Clock, Effect } from "effect" +import { Clock, Effect, Iterable } from "effect" import { Bus } from "../../bus.js" import { Model } from "../../model.js" import { SessionEvent } from "../event.js" @@ -39,13 +39,7 @@ export interface StepRecord { readonly providerState?: SessionMessage.ProviderState readonly tokens: ReturnType } - readonly calls: ReadonlyArray<{ - readonly id: string - readonly name: string - readonly called: boolean - readonly settled: boolean - readonly providerExecuted: boolean - }> + readonly needsContinuation: boolean } /** Derives canonical model content from a provider-hosted tool result. */ @@ -85,7 +79,6 @@ const hostedContent = (result: ToolResultValue): NonEmptyContent => { export const createLLMEventPublisher = (bus: Pick, input: Input) => { const deltaBatchInterval = 100 type ToolState = { - readonly assistantMessageID: SessionMessage.ID readonly name: string called: boolean settled: boolean @@ -250,7 +243,7 @@ export const createLLMEventPublisher = (bus: Pick, inp if (!tool) return yield* Effect.die(new Error(`Tool input end before start: ${id}`)) yield* bus.publish(SessionEvent.Tool.Input.Ended, { sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, + assistantMessageID, id, text: value, }) @@ -269,9 +262,8 @@ export const createLLMEventPublisher = (bus: Pick, inp readonly providerExecuted?: boolean }) { if (tools.has(event.id)) return yield* Effect.die(new Error(`Duplicate tool input start: ${event.id}`)) - const assistantMessageID = yield* startAssistant() + yield* startAssistant() const tool: ToolState = { - assistantMessageID, name: event.name, called: false, settled: false, @@ -314,7 +306,7 @@ export const createLLMEventPublisher = (bus: Pick, inp tool.settled = true yield* bus.publish(SessionEvent.Tool.Failed, { sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, + assistantMessageID, id: event.id, error: { type: "tool.input-json", @@ -333,7 +325,7 @@ export const createLLMEventPublisher = (bus: Pick, inp tool.settled = true yield* bus.publish(SessionEvent.Tool.Failed, { sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, + assistantMessageID, id, error, ...failureSnapshot(tool, metadata), @@ -383,11 +375,6 @@ export const createLLMEventPublisher = (bus: Pick, inp (error: SessionError.Error, scope: "hosted" | "all" = "all") => failTools(error, scope), ) - const assistantMessageIDForTool = (id: string) => { - const tool = tools.get(id) - return tool ? Effect.succeed(tool.assistantMessageID) : Effect.die(new Error(`Unknown tool call: ${id}`)) - } - const publish = Effect.fn("SessionRunner.publishLLMEvent")(function* (event: LLMEvent) { switch (event.type) { case "step-start": @@ -455,7 +442,7 @@ export const createLLMEventPublisher = (bus: Pick, inp tool.providerExecuted = event.providerExecuted === true yield* bus.publish(SessionEvent.Tool.Called, { sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, + assistantMessageID, id: event.id, input: asRecord(event.input), executed: tool.providerExecuted, @@ -481,7 +468,7 @@ export const createLLMEventPublisher = (bus: Pick, inp if (event.result.type === "error") { yield* bus.publish(SessionEvent.Tool.Failed, { sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, + assistantMessageID, id: event.id, error: { type: "tool.execution", message: stringify(event.result.value) }, ...failureSnapshot(tool), @@ -492,7 +479,7 @@ export const createLLMEventPublisher = (bus: Pick, inp } yield* bus.publish(SessionEvent.Tool.Success, { sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, + assistantMessageID, id: event.id, content: hostedContent(event.result), executed, @@ -509,7 +496,7 @@ export const createLLMEventPublisher = (bus: Pick, inp tool.settled = true yield* bus.publish(SessionEvent.Tool.Failed, { sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, + assistantMessageID, id: event.id, error: event.message === `Unknown tool: ${event.name}` @@ -551,7 +538,7 @@ export const createLLMEventPublisher = (bus: Pick, inp tool.progress = update yield* bus.publish(SessionEvent.Tool.Progress, { sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, + assistantMessageID, id, metadata: update, }) @@ -574,7 +561,7 @@ export const createLLMEventPublisher = (bus: Pick, inp if (content.length === 0) return yield* Effect.die(new Error(`Tool execution has no content: ${id}`)) yield* bus.publish(SessionEvent.Tool.Success, { sessionID: input.sessionID, - assistantMessageID: tool.assistantMessageID, + assistantMessageID, id, content: [content[0], ...content.slice(1)], ...(result.metadata === undefined ? {} : { metadata: result.metadata }), @@ -599,16 +586,12 @@ export const createLLMEventPublisher = (bus: Pick, inp providerFailed, failure: stepFailure, finish: stepSettlement, - calls: Array.from(tools, ([id, tool]) => ({ - id, - name: tool.name, - called: tool.called, - settled: tool.settled, - providerExecuted: tool.providerExecuted, - })), + needsContinuation: Iterable.some( + tools.values(), + (tool) => !tool.providerExecuted && (tool.called || tool.settled), + ), }), startAssistant, streamed, - assistantMessageID: assistantMessageIDForTool, } } diff --git a/packages/core/src/session/runner/retry.ts b/packages/core/src/session/runner/retry.ts index 7b44aa892b9..10f66800973 100644 --- a/packages/core/src/session/runner/retry.ts +++ b/packages/core/src/session/runner/retry.ts @@ -2,17 +2,17 @@ export * as SessionRunnerRetry from "./retry.js" import { AIError } from "@opencode-ai/ai" import { SessionError } from "@opencode-ai/schema/session-error" -import { Data, Duration, Effect, Schedule } from "effect" +import { Duration, Effect, Schedule } from "effect" import { Bus } from "../../bus.js" import { SessionEvent } from "../event.js" import { SessionMessage } from "../message.js" import { SessionSchema } from "../schema.js" -export class RetryableFailure extends Data.TaggedError("SessionRunner.RetryableFailure")<{ +export interface Input { readonly cause: AIError readonly error: SessionError.Error - readonly step: number -}> {} + readonly assistantMessageID: SessionMessage.ID +} export function isRetryable(error: AIError) { const override = "http" in error.reason ? error.reason.http?.response?.headers["x-should-retry"] : undefined @@ -40,29 +40,25 @@ export function isRetryable(error: AIError) { } } -const retryAfter = (failure: RetryableFailure) => { - if (failure.cause.reason._tag === "RateLimit" || failure.cause.reason._tag === "ProviderInternal") - return failure.cause.reason.retryAfterMs +const retryAfter = (input: Input) => { + if (input.cause.reason._tag === "RateLimit" || input.cause.reason._tag === "ProviderInternal") + return input.cause.reason.retryAfterMs return undefined } -export const schedule = ( - bus: Bus.Interface, - sessionID: SessionSchema.ID, - assistantMessageID: () => SessionMessage.ID, -) => +export const schedule = (bus: Bus.Interface, sessionID: SessionSchema.ID) => Schedule.max([Schedule.exponential("2 seconds"), Schedule.recurs(4)]).pipe( Schedule.jittered, - Schedule.setInputType(), - Schedule.modifyDelay(({ input: failure, duration: delay }) => { - const minimum = retryAfter(failure) + Schedule.setInputType(), + Schedule.modifyDelay(({ input, duration: delay }) => { + const minimum = retryAfter(input) const duration = minimum === undefined ? delay : Duration.max(delay, Duration.millis(minimum)) return Effect.succeed(Duration.millis(Math.ceil(Duration.toMillis(duration)))) }), Schedule.tap((metadata) => bus.publish(SessionEvent.RetryScheduled, { sessionID, - assistantMessageID: assistantMessageID(), + assistantMessageID: metadata.input.assistantMessageID, attempt: metadata.attempt + 1, at: metadata.now + Duration.toMillis(metadata.duration), error: metadata.input.error, diff --git a/packages/core/src/session/runner/step.ts b/packages/core/src/session/runner/step.ts new file mode 100644 index 00000000000..f103a2728d2 --- /dev/null +++ b/packages/core/src/session/runner/step.ts @@ -0,0 +1,293 @@ +export * as SessionStep from "./step.js" + +import { + AIError, + InvalidProviderOutputReason, + LLMClient, + LLMEvent, + isContextOverflowFailure, + type ProviderErrorEvent, + type ToolCall, +} from "@opencode-ai/ai" +import { Cause, Data, Effect, Exit, Fiber, Option, Stream } from "effect" +import { SessionError } from "@opencode-ai/schema/session-error" +import { Agent } from "../../agent.js" +import { Bus } from "../../bus.js" +import { Permission } from "../../permission.js" +import { Snapshot } from "../../snapshot.js" +import { Tool } from "../../tool.js" +import { ToolOutput } from "../../tool-output.js" +import { QuestionTool } from "../../tool/plugin/question.js" +import { StepFailedError } from "../error.js" +import { SessionEvent } from "../event.js" +import { SessionMessage } from "../message.js" +import { SessionModelRequest } from "../model-request.js" +import { SessionSchema } from "../schema.js" +import { toSessionError } from "../to-session-error.js" +import { SessionUsage } from "../usage.js" +import { SessionRunnerModel } from "./model.js" +import { createLLMEventPublisher } from "./publish-llm-event.js" +import { SessionRunnerRetry } from "./retry.js" + +export type Outcome = Data.TaggedEnum<{ + Completed: { readonly needsContinuation: boolean } + Retry: { readonly cause: AIError; readonly error: SessionError.Error } + Continue: { readonly cause: AIError; readonly error: SessionError.Error } + RecoverFull: {} + Compacted: {} +}> +const Outcome = Data.taggedEnum() + +interface Input { + readonly sessionID: SessionSchema.ID + readonly assistantMessageID: SessionMessage.ID + readonly agent: Agent.ID + readonly model: SessionRunnerModel.Resolved + readonly prepared: SessionModelRequest.Prepared + readonly toolsDisabled: boolean + readonly recoverContinuation: boolean + /** The runner owns compaction policy; the attempt invokes it only before durable output. */ + readonly recoverOverflow: Effect.Effect +} + +const TOOLS_INTERRUPTED = { type: "aborted", message: "Tool execution interrupted" } as const +const STEP_INTERRUPTED = { type: "aborted", message: "Step interrupted" } as const +const RESULT_MISSING = { type: "tool.result-missing", message: "Provider did not return a tool result" } as const + +/** Captures Location-scoped dependencies without introducing another service or execution loop. */ +export const make = Effect.gen(function* () { + const bus = yield* Bus.Service + const llm = yield* LLMClient.Service + const snapshots = yield* Snapshot.Service + const toolOutput = yield* ToolOutput.Service + + const attempt = Effect.fn("SessionStep.attempt")(function* (input: Input) { + const startSnapshot = yield* snapshots.capture() + const publisher = createLLMEventPublisher(bus, { + sessionID: input.sessionID, + assistantMessageID: input.assistantMessageID, + agent: input.agent, + model: input.model.ref, + providerMetadataKey: input.model.model.route.providerMetadataKey ?? input.model.model.provider, + snapshot: startSnapshot, + }) + const toolRuns: Array<{ + readonly call: ToolCall + readonly fiber: Fiber.Fiber + }> = [] + const interruptTools = Effect.suspend(() => Fiber.interruptAll(toolRuns.map((run) => run.fiber))) + const executeTool = (call: ToolCall) => { + if (input.toolsDisabled) return new Tool.Error({ message: "Tools are disabled after the maximum agent steps" }) + return input.prepared.executeTool({ + sessionID: input.sessionID, + agent: input.agent, + messageID: input.assistantMessageID, + call, + progress: (update) => publisher.progress(call.id, update), + }) + } + + // Provider and tool fibers retain per-source order without a shared writer queue. + // A local execution starts only after its Tool.Called publication completes. + let overflowFailure: ProviderErrorEvent | undefined + // Read to the end, not just the finish event, so the next request can reuse this response. + const providerStream = llm.stream(input.prepared.request, input.prepared.options).pipe( + Stream.runForEach((event) => + Effect.gen(function* () { + if (overflowFailure || publisher.hasProviderError()) return + if ( + LLMEvent.is.providerError(event) && + isContextOverflowFailure(event) && + !publisher.record().outputStarted + ) { + overflowFailure = event + return + } + yield* publisher.publish(event) + if (event.type !== "tool-call" || event.providerExecuted) return + toolRuns.push({ + call: event, + fiber: yield* Effect.uninterruptibleMask((restore) => + restore(executeTool(event)).pipe( + Effect.flatMap(toolOutput.truncate), + Effect.flatMap((outcome) => publisher.toolExecution(event.id, event.name, outcome)), + Effect.catchTag("Tool.Error", (error) => + publisher.failTool(event.id, toSessionError(error), error.metadata).pipe(Effect.asVoid), + ), + ), + ).pipe(Effect.forkScoped), + }) + }), + ), + Effect.ensuring(publisher.flush()), + ) + + // Keep the final tool and Step events uninterruptible, even when the work itself is cancelled. + return yield* Effect.uninterruptibleMask((restore) => + Effect.gen(function* () { + const stream = yield* restore(providerStream).pipe(Effect.exit) + const streamFailure = Option.getOrUndefined(Exit.findErrorOption(stream)) + const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) + if (!overflowFailure && publisher.hasStarted()) yield* publisher.streamed() + if (streamInterrupted) yield* interruptTools + const joined = yield* restore(Fiber.awaitAll(toolRuns.map((run) => run.fiber))).pipe(Effect.exit) + if (joined._tag === "Failure") yield* interruptTools + const tools = classifyToolExits( + joined, + toolRuns.map((run) => run.call), + ) + + if ( + !publisher.record().outputStarted && + isContextOverflowFailure(overflowFailure ?? streamFailure) && + (yield* restore(input.recoverOverflow)) + ) + return Outcome.Compacted() + + if (overflowFailure) yield* publisher.publish(overflowFailure) + const recorded = publisher.record() + const unknownFinish = + stream._tag === "Success" && recorded.finish?.finish === "unknown" + ? new AIError({ + module: "session", + method: "stream", + reason: new InvalidProviderOutputReason({ + classification: "incomplete-stream", + message: "The provider response ended with an unknown finish reason.", + }), + }) + : undefined + const llmFailure = streamFailure instanceof AIError ? streamFailure : unknownFinish + const llmError = llmFailure && !recorded.providerFailed ? toSessionError(llmFailure) : undefined + if ( + input.recoverContinuation && + llmFailure?.reason._tag === "Transport" && + (llmFailure.reason.recovery === "retry-full" || llmFailure.reason.recovery === "rotate-and-retry-full") && + !recorded.outputStarted + ) + return Outcome.RecoverFull() + if (llmFailure && llmError && SessionRunnerRetry.isRetryable(llmFailure) && !recorded.outputStarted) { + // Retry state projects onto the existing assistant, even before it has produced output. + yield* publisher.startAssistant() + return Outcome.Retry({ cause: llmFailure, error: llmError }) + } + if (llmError) yield* publisher.failAssistant(llmError) + + for (const decline of tools.declines) + yield* publisher.failTool(decline.call.id, { + type: "aborted", + message: + decline.reason._tag === "QuestionTool.CancelledError" + ? decline.reason.message + : "The user declined this tool call", + }) + const interrupted = tools.declines.length > 0 || streamInterrupted || tools.interrupted + const toolFailure = interrupted + ? TOOLS_INTERRUPTED + : tools.failure !== undefined + ? toSessionError(Cause.squash(tools.failure)) + : recorded.providerFailed + ? TOOLS_INTERRUPTED + : undefined + if (toolFailure) yield* publisher.failUnsettledTools(toolFailure) + if (interrupted) yield* publisher.failAssistant(STEP_INTERRUPTED) + + // All local fibers have joined; only provider-hosted results can still be missing. + if (llmError || (stream._tag === "Success" && !recorded.providerFailed)) { + const missing = yield* publisher.failUnsettledTools(RESULT_MISSING, "hosted") + if (missing && !llmError && !recorded.finish) yield* publisher.failAssistant(RESULT_MISSING) + } + + const record = publisher.record() + if (record.finish || record.failure) { + const snapshot = yield* snapshots.capture() + const files = + startSnapshot && snapshot + ? startSnapshot === snapshot + ? [] + : yield* snapshots + .files({ from: startSnapshot, to: snapshot }) + .pipe(Effect.orElseSucceed(() => undefined)) + : undefined + const usage = record.finish + ? { cost: SessionUsage.calculateCost(input.model.cost, record.finish.tokens), tokens: record.finish.tokens } + : undefined + if (record.failure) yield* publisher.publishStepFailure({ ...usage, snapshot, files }) + if (record.finish && usage && !record.failure) + yield* bus.publish(SessionEvent.Step.Ended, { + sessionID: input.sessionID, + assistantMessageID: yield* publisher.startAssistant(), + finish: record.finish.finish, + rawFinish: record.finish.rawFinish, + providerState: record.finish.providerState, + ...usage, + snapshot, + files, + }) + } + + if ( + llmFailure && + llmError && + isInterruptedStream(llmFailure) && + record.outputStarted && + tools.declines.length === 0 && + !tools.interrupted + ) + return Outcome.Continue({ cause: llmFailure, error: llmError }) + + if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause) + if (tools.declines.length > 0) return yield* Effect.interrupt + if (tools.interrupted && tools.failure) return yield* Effect.failCause(tools.failure) + if (tools.interrupted && joined._tag === "Failure") return yield* Effect.failCause(joined.cause) + if (record.failure) return yield* new StepFailedError({ error: record.failure }) + return Outcome.Completed({ + needsContinuation: !input.toolsDisabled && record.needsContinuation, + }) + }), + ) + }, Effect.scoped) + + return { attempt } +}) + +const isDecline = ( + error: SessionModelRequest.ExecuteError, +): error is Permission.DeclinedError | QuestionTool.CancelledError => + error._tag === "Permission.DeclinedError" || error._tag === "QuestionTool.CancelledError" + +const isInterruptedStream = (failure: AIError) => { + if (failure.reason._tag === "InvalidProviderOutput") return failure.reason.classification === "incomplete-stream" + if (failure.reason._tag === "Transport") return failure.reason.operation === "read" + return false +} + +/** Keep every joined exit associated with its call; a decline is not an infrastructure failure. */ +const classifyToolExits = ( + settled: Exit.Exit>>, + calls: ReadonlyArray, +) => { + const exits = settled._tag === "Success" ? settled.value : [] + const declines = exits.flatMap((exit, index) => + exit._tag === "Failure" + ? exit.cause.reasons.flatMap((reason) => + Cause.isFailReason(reason) && isDecline(reason.error) ? [{ call: calls[index], reason: reason.error }] : [], + ) + : [], + ) + const causes = + settled._tag === "Failure" + ? [settled.cause] + : exits.flatMap((exit) => (exit._tag === "Failure" ? [exit.cause] : [])) + const failure = causes + .flatMap((cause) => { + if (Cause.hasInterrupts(cause)) return [] + const reasons = cause.reasons.flatMap( + (reason): Array> => + Cause.isFailReason(reason) ? (isDecline(reason.error) ? [] : [Cause.makeDieReason(reason.error)]) : [reason], + ) + return reasons.length > 0 ? [Cause.fromReasons(reasons)] : [] + }) + .at(0) + return { interrupted: causes.some(Cause.hasInterrupts), declines, failure } +} diff --git a/packages/core/test/session-execution.test.ts b/packages/core/test/session-execution.test.ts index 511056b041b..bdf2109cb2b 100644 --- a/packages/core/test/session-execution.test.ts +++ b/packages/core/test/session-execution.test.ts @@ -538,7 +538,9 @@ function buildExecution( const store = yield* SessionStore.Service const runner = Layer.succeed( SessionRunner.Service, - SessionRunner.Service.of({ drain: (input) => drain(input).pipe(Effect.as({ type: "complete" as const })) }), + SessionRunner.Service.of({ + drain: (input) => drain(input).pipe(Effect.as(SessionRunner.DrainResult.Complete())), + }), ) const locations = Layer.effect( LocationServiceMap.Service, diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index 1ed0f16fac5..e6f80c30933 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -342,7 +342,7 @@ test("step finish records settlement without publishing step ended", async () => await Effect.runPromise(publisher.publish(LLMEvent.stepStart({ index: 0 }))) await Effect.runPromise(publisher.publish(LLMEvent.stepFinish({ index: 0, reason: { normalized: "stop" } }))) - expect(published.some((event) => event.type === "step.ended.2")).toBe(false) + expect(published.map((event) => event.type)).toEqual(["session.step.started.1"]) expect(publisher.record().finish).toMatchObject({ finish: "stop" }) }) diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index 656a322b3af..22d339ac67c 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -76,7 +76,7 @@ import { SessionSystemPrompt } from "@opencode-ai/core/session/system-prompt" import { ID } from "@opencode-ai/core/model" import { Location } from "@opencode-ai/core/location" import { Provider } from "@opencode-ai/core/provider" -import { Cause, Deferred, Effect, Exit, Fiber, Layer, Schema, Scope, Stream } from "effect" +import { Cause, Deferred, Effect, Exit, Fiber, Layer, Queue, Schema, Scope, Stream } from "effect" import { TestClock } from "effect/testing" import { HttpClientRequest, HttpClientResponse } from "effect/unstable/http" import { asc, desc, eq } from "drizzle-orm" @@ -419,7 +419,7 @@ const execution = Layer.effect( .drain({ sessionID, force, continuation }) .pipe( Effect.flatMap((result) => - result.type === "complete" ? Effect.void : drain(sessionID, false, result.continuation), + result._tag === "Complete" ? Effect.void : drain(sessionID, false, result.continuation), ), ) } @@ -2926,25 +2926,41 @@ describe("SessionRunnerLLM", () => { }), ) - it.effect("records the stream boundary before local tools complete", () => + it.effect("consumes the full provider stream before recording its boundary and settling local tools", () => Effect.gen(function* () { const session = yield* setup const bus = yield* Bus.Service yield* admit(session, "Echo this") - yield* TestLLM.push(TestLLM.tool("call-streamed", "echo", { text: "hello" }), TestLLM.stop()) + const tail = yield* Deferred.make() + const complete = yield* Deferred.make() + const finished = yield* Deferred.make() + yield* TestLLM.push( + Stream.fromIterable(TestLLM.tool("call-streamed", "echo", { text: "hello" })).pipe( + Stream.concat( + Stream.fromEffect(Deferred.succeed(tail, undefined).pipe(Effect.andThen(Deferred.await(complete)))).pipe( + Stream.drain, + ), + ), + Stream.onEnd(Deferred.succeed(finished, undefined)), + ), + TestLLM.stop(), + ) const tools = yield* blockTools() - const streamed = yield* bus - .subscribe(SessionEvent.Step.Streamed) - .pipe( - Stream.filter((event) => event.data.sessionID === sessionID), - Stream.take(1), - Stream.runDrain, - Effect.forkScoped({ startImmediately: true }), - ) + const streamed = yield* bus.subscribe(SessionEvent.Step.Streamed).pipe( + Stream.filter((event) => event.data.sessionID === sessionID), + Stream.runHead, + Effect.forkScoped({ startImmediately: true }), + ) const run = yield* Effect.forkChild(session.resume(sessionID)) yield* tools.started + yield* Deferred.await(tail) + expect(requests).toHaveLength(1) + expect(yield* recordedEventTypes(sessionID)).not.toContain("session.step.streamed.1") + expect(requireAssistant(yield* session.context(sessionID)).time.completed).toBeUndefined() + yield* Deferred.succeed(complete, undefined) yield* Fiber.join(streamed) + expect(yield* Deferred.isDone(finished)).toBe(true) const assistant = requireAssistant(yield* session.context(sessionID)) expect(assistant.time.streamed).toBeDefined() expect(assistant.time.completed).toBeUndefined() @@ -2952,6 +2968,10 @@ describe("SessionRunnerLLM", () => { yield* tools.release yield* Fiber.join(run) + const events = yield* recordedEventTypes(sessionID) + expect(events.indexOf("session.step.streamed.1")).toBeLessThan(events.indexOf("session.tool.success.2")) + expect(events.indexOf("session.tool.success.2")).toBeLessThan(events.indexOf("session.step.ended.1")) + expect(events.filter((type) => type === "session.step.streamed.1")).toHaveLength(2) }), ) @@ -4265,16 +4285,24 @@ describe("SessionRunnerLLM", () => { it.effect("durably fails blocked local tools when interrupted while awaiting settlement", () => Effect.gen(function* () { const session = yield* setup + const bus = yield* Bus.Service yield* admit(session, "Interrupt tool settlement") const tools = yield* blockTools() yield* TestLLM.push(TestLLM.tool("call-await-interrupt", "echo", { text: "blocked" })) + const streamed = yield* bus.subscribe(SessionEvent.Step.Streamed).pipe( + Stream.filter((event) => event.data.sessionID === sessionID), + Stream.runHead, + Effect.forkScoped({ startImmediately: true }), + ) const runner = yield* SessionRunner.Service const run = yield* runner.drain({ sessionID, force: true }).pipe(Effect.forkChild) yield* tools.started + yield* Fiber.join(streamed) yield* Fiber.interrupt(run) - expect(yield* Fiber.await(run)).toMatchObject({ _tag: "Failure" }) + const exit = yield* Fiber.await(run) + expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true) expect(yield* session.context(sessionID)).toMatchObject([ { type: "user", text: "Interrupt tool settlement" }, { @@ -4291,8 +4319,11 @@ describe("SessionRunnerLLM", () => { }, ]) const eventTypes = yield* recordedEventTypes(sessionID) - expect(eventTypes).toContain("session.step.failed.1") + expect(eventTypes.filter((type) => type === "session.tool.failed.2")).toHaveLength(1) + expect(eventTypes.filter((type) => type === "session.step.failed.1")).toHaveLength(1) expect(eventTypes).not.toContain("session.step.ended.1") + expect(eventTypes).not.toContain("session.retry.scheduled.1") + expect(requests).toHaveLength(1) }), ) @@ -4572,6 +4603,30 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("does not start another physical attempt after interruption during retry backoff", () => + Effect.gen(function* () { + const session = yield* setup + const bus = yield* Bus.Service + yield* admit(session, "Interrupt retry backoff") + yield* TestLLM.push(Stream.fail(providerUnavailable()), TestLLM.text("Must not run", "unused-retry")) + const scheduled = yield* bus.subscribe(SessionEvent.RetryScheduled).pipe( + Stream.filter((event) => event.data.sessionID === sessionID), + Stream.runHead, + Effect.forkScoped({ startImmediately: true }), + ) + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* Fiber.join(scheduled) + yield* session.interrupt(sessionID) + const exit = yield* Fiber.await(run) + expect(Exit.isFailure(exit) && Cause.hasInterruptsOnly(exit.cause)).toBe(true) + yield* TestClock.adjust("1 minute") + expect(requests).toHaveLength(1) + const events = yield* recordedEventTypes(sessionID) + expect(events.filter((type) => type === "session.retry.scheduled.1")).toHaveLength(1) + expect(events).not.toContain("session.synthetic.1") + }), + ) + it.effect("immediately rebuilds once after explicit continuation rejection", () => Effect.gen(function* () { const session = yield* setup @@ -4921,6 +4976,45 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("shares retry accounting and assistant identity across transparent retries and partial continuations", () => + Effect.gen(function* () { + const session = yield* setup + const bus = yield* Bus.Service + const scheduled = yield* Queue.unbounded() + yield* bus.subscribe(SessionEvent.RetryScheduled).pipe( + Stream.filter((event) => event.data.sessionID === sessionID), + Stream.runForEach((event) => Queue.offer(scheduled, event.data.assistantMessageID)), + Effect.forkScoped({ startImmediately: true }), + ) + yield* admit(session, "Mix retry paths") + const failure = incompleteStream() + const partial = TestLLM.failAfter( + failure, + LLMEvent.stepStart({ index: 0 }), + LLMEvent.textStart({ id: "mixed-partial" }), + LLMEvent.textDelta({ id: "mixed-partial", text: "Partial" }), + ) + yield* TestLLM.push(Stream.fail(failure), partial, Stream.fail(failure), partial, partial) + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + const identities: SessionMessage.ID[] = [] + for (const delay of [2_400, 4_800, 9_600, 19_200]) { + identities.push(yield* Queue.take(scheduled)) + yield* TestClock.adjust(delay) + } + expect(yield* Fiber.join(run).pipe(Effect.flip)).toBe(failure) + expect(requests).toHaveLength(5) + expect(identities[0]).toBe(identities[1]) + expect(identities[2]).toBe(identities[3]) + expect(identities[0]).not.toBe(identities[2]) + const messages = yield* session.context(sessionID) + expect(messages.filter((message) => message.type === "assistant")).toHaveLength(3) + expect(messages.filter((message) => message.type === "synthetic")).toHaveLength(2) + const events = yield* recordedEventTypes(sessionID) + expect(events.filter((type) => type === "session.retry.scheduled.1")).toHaveLength(4) + expect(events.filter((type) => type === "session.step.failed.1")).toHaveLength(3) + }), + ) + it.effect("stops incomplete stream continuations after five total attempts", () => Effect.gen(function* () { const session = yield* setup diff --git a/packages/core/test/session-step.test.ts b/packages/core/test/session-step.test.ts new file mode 100644 index 00000000000..3d2621c888b --- /dev/null +++ b/packages/core/test/session-step.test.ts @@ -0,0 +1,140 @@ +import { expect } from "bun:test" +import { LanguageModel, LLM, LLMClient, LLMEvent } from "@opencode-ai/ai" +import { OpenAIChat } from "@opencode-ai/ai/protocols/openai-chat" +import { TestLLM } from "@opencode-ai/ai/testing" +import { Agent } from "@opencode-ai/core/agent" +import { Bus } from "@opencode-ai/core/bus" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { EventTable } from "@opencode-ai/core/event/sql" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath, RelativePath } from "@opencode-ai/core/schema" +import { Session } from "@opencode-ai/core/session" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" +import { SessionStep } from "@opencode-ai/core/session/runner/step" +import { SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { Snapshot } from "@opencode-ai/core/snapshot" +import { ToolOutput } from "@opencode-ai/core/tool-output" +import { Money } from "@opencode-ai/schema/money" +import { LayerNode } from "@opencode-ai/util/effect/layer-node" +import { asc, eq } from "drizzle-orm" +import { Effect, Exit, Layer } from "effect" +import { testEffect } from "./lib/effect" + +const it = testEffect( + Layer.merge( + AppNodeBuilder.build(LayerNode.group([Database.node, Bus.node, SessionProjector.node, ToolOutput.node]), [ + [Bus.node, Bus.configured({ persist: true })], + ]), + TestLLM.layer(), + ), +) + +for (const finish of ["stop", "content-filter"] as const) { + it.effect(`settles ${finish} with snapshot files and nonzero usage after its tool`, () => + Effect.gen(function* () { + const db = (yield* Database.Service).db + const llm = yield* TestLLM.Service + const sessionID = Session.ID.create() + const assistantMessageID = SessionMessage.ID.create() + const start = Snapshot.ID.make("before") + const end = Snapshot.ID.make("after") + const files = [RelativePath.make("changed.ts")] + let captures = 0 + const steps = yield* SessionStep.make.pipe( + Effect.provideService(LLMClient.Service, llm.client), + Effect.provide( + Layer.mock(Snapshot.Service)({ + capture: () => Effect.sync(() => (captures++ === 0 ? start : end)), + files: (input) => { + expect(input).toEqual({ from: start, to: end }) + return Effect.succeed(files) + }, + }), + ), + ) + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .run() + yield* db + .insert(SessionTable) + .values({ id: sessionID, project_id: Project.ID.global, slug: "step", directory: "/project", version: "test" }) + .run() + const model = SessionRunnerModel.resolved( + LanguageModel.make({ id: "test-model", provider: "test", route: OpenAIChat.route }), + { + capabilities: { tools: true, input: ["text"], output: ["text"] }, + limit: { context: 100_000, output: 1_000 }, + cost: [ + { + input: Money.USDPerMillionTokens.make(1), + output: Money.USDPerMillionTokens.make(2), + cache: { read: Money.USDPerMillionTokens.make(0.1), write: Money.USDPerMillionTokens.make(0.5) }, + }, + ], + }, + ) + yield* llm.push( + TestLLM.complete( + { + reason: { normalized: finish }, + usage: { + inputTokens: 15, + outputTokens: 6, + nonCachedInputTokens: 10, + cacheReadInputTokens: 3, + cacheWriteInputTokens: 2, + reasoningTokens: 2, + }, + }, + LLMEvent.toolCall({ id: "call-test", name: "test", input: {} }), + ), + ) + const result = yield* steps + .attempt({ + sessionID, + assistantMessageID, + agent: Agent.defaultID, + model, + prepared: { + request: LLM.request({ model: model.model, prompt: "Run one tool" }), + options: {}, + executeTool: () => Effect.succeed({ content: "Completed tool" }), + }, + toolsDisabled: false, + recoverContinuation: true, + recoverOverflow: Effect.succeed(false), + }) + .pipe(Effect.exit) + expect(Exit.isSuccess(result)).toBe(finish === "stop") + expect(llm.requests).toHaveLength(1) + expect(captures).toBe(2) + const message = yield* db + .select() + .from(SessionMessageTable) + .where(eq(SessionMessageTable.id, assistantMessageID)) + .get() + expect(message?.data).toMatchObject({ + finish, + tokens: { input: 10, output: 4, reasoning: 2, cache: { read: 3, write: 2 } }, + snapshot: { start, end, files }, + content: [{ type: "tool", state: { status: "completed" } }], + }) + expect(message?.data).toHaveProperty("cost", expect.closeTo(0.0000233, 10)) + const events = yield* db + .select({ type: EventTable.type }) + .from(EventTable) + .where(eq(EventTable.aggregate_id, sessionID)) + .orderBy(asc(EventTable.seq)) + .all() + const types = events.map((event) => event.type) + const terminal = finish === "stop" ? "session.step.ended.1" : "session.step.failed.1" + expect(types.filter((type) => type === terminal)).toHaveLength(1) + expect(types.indexOf("session.tool.success.2")).toBeLessThan(types.indexOf(terminal)) + }), + ) +} diff --git a/packages/tui/src/util/selection.ts b/packages/tui/src/util/selection.ts index 1e21b93ceb7..a3d218db3ff 100644 --- a/packages/tui/src/util/selection.ts +++ b/packages/tui/src/util/selection.ts @@ -1,3 +1,4 @@ +import type { SelectionBehavior } from "@opentui/core" import type { ClipboardService } from "../context/clipboard" type Toast = { @@ -15,6 +16,7 @@ type Renderer = { getSelectedText: () => string selectedRenderables: FocusableSelectionTarget[] isStart: boolean + behavior: SelectionBehavior } | null clearSelection: () => void currentFocusedRenderable?: FocusableSelectionTarget | null @@ -34,13 +36,16 @@ export function copyOnSelectRelease( clipboard: ClipboardService, ): boolean { if (!event.isDragging) return false + const selection = renderer.getSelection() + // Preserve the first click so OpenTUI can recognize the following double/triple click. + if (selection?.isStart && selection.behavior === "cell") return false return copy(renderer, toast, clipboard) } export function copy(renderer: Renderer, toast: Toast, clipboard: ClipboardService): boolean { const selection = renderer.getSelection() if (!selection) return false - if (selection.isStart) { + if (selection.isStart && selection.behavior === "cell") { renderer.clearSelection() return false } diff --git a/packages/tui/test/util/selection-copy-on-select.test.tsx b/packages/tui/test/util/selection-copy-on-select.test.tsx index 713f93f6606..7b9547c50b1 100644 --- a/packages/tui/test/util/selection-copy-on-select.test.tsx +++ b/packages/tui/test/util/selection-copy-on-select.test.tsx @@ -45,6 +45,7 @@ test("copy-on-select keeps a word highlight so a third click can select the line await app.mockMouse.click(6, 0) expect(app.renderer.getSelection()?.getSelectedText() ?? "").toBe("") + expect(writes).toEqual([]) await app.mockMouse.click(6, 0) expect(app.renderer.getSelection()?.getSelectedText()).toBe("beta") diff --git a/packages/tui/test/util/selection.test.ts b/packages/tui/test/util/selection.test.ts index c6dba3f4308..14ee4dac8a3 100644 --- a/packages/tui/test/util/selection.test.ts +++ b/packages/tui/test/util/selection.test.ts @@ -1,4 +1,5 @@ import { expect, test } from "bun:test" +import type { SelectionBehavior } from "@opentui/core" import type { ClipboardService } from "../../src/context/clipboard" import { Selection, copy, copyOnSelectRelease } from "../../src/util/selection" @@ -8,12 +9,13 @@ function renderer() { getSelectedText: () => "beta", selectedRenderables: [], isStart: false, + behavior: "cell" as const, }), clearSelection: () => {}, } } -function setup(text: string, isStart: boolean) { +function setup(text: string, isStart: boolean, behavior: SelectionBehavior = "cell") { const writes: string[] = [] let clears = 0 const clipboard: ClipboardService = { @@ -23,7 +25,7 @@ function setup(text: string, isStart: boolean) { }, } const renderer = { - getSelection: () => ({ getSelectedText: () => text, selectedRenderables: [], isStart }), + getSelection: () => ({ getSelectedText: () => text, selectedRenderables: [], isStart, behavior }), clearSelection: () => { clears++ }, @@ -41,6 +43,7 @@ test("copy writes selected text without clearing the highlight", () => { getSelectedText: () => "beta", selectedRenderables: [], isStart: false, + behavior: "cell", }), clearSelection: () => { cleared = true @@ -75,6 +78,20 @@ test("copy-on-select ignores a later non-drag release", () => { expect(writes).toEqual(["beta"]) }) +test("copy-on-select preserves a click-only selection for subsequent clicks", () => { + const value = setup("", true) + expect(copyOnSelectRelease({ isDragging: true }, value.renderer, value.toast, value.clipboard)).toBeFalse() + expect(value.clears()).toBe(0) + expect(value.writes).toEqual([]) +}) + +test.each(["word", "line"] as const)("copy-on-select copies a %s selection without pointer movement", (behavior) => { + const value = setup("selected", true, behavior) + expect(copyOnSelectRelease({ isDragging: true }, value.renderer, value.toast, value.clipboard)).toBeTrue() + expect(value.clears()).toBe(0) + expect(value.writes).toEqual(["selected"]) +}) + test("clears a click-only selection without copying", () => { const value = setup("x", true) expect(Selection.copy(value.renderer, value.toast, value.clipboard)).toBeFalse()