From cd0b27485626ebe46d8c7b4ee2f2a90174de779d Mon Sep 17 00:00:00 2001 From: Kit Langton Date: Thu, 2 Jul 2026 21:38:44 -0400 Subject: [PATCH] refactor(core): simplify v2 prompt lifecycle and execution coordination (#35047) --- packages/core/src/session/execution/local.ts | 25 ++- packages/core/src/session/input.ts | 32 +--- packages/core/src/session/run-coordinator.ts | 126 +++++++------ packages/core/src/session/runner/llm.ts | 177 ++++++++---------- packages/core/test/session-compact.test.ts | 13 +- packages/core/test/session-create.test.ts | 14 +- packages/core/test/session-projector.test.ts | 14 ++ packages/core/test/session-prompt.test.ts | 68 +------ .../core/test/session-run-coordinator.test.ts | 62 ++++++ packages/core/test/session-runner.test.ts | 66 ++++++- packages/core/test/session-title.test.ts | 43 +++-- 11 files changed, 362 insertions(+), 278 deletions(-) diff --git a/packages/core/src/session/execution/local.ts b/packages/core/src/session/execution/local.ts index 51b6446c13d..8609162c82d 100644 --- a/packages/core/src/session/execution/local.ts +++ b/packages/core/src/session/execution/local.ts @@ -1,6 +1,8 @@ -import { Cause, Effect, Layer } from "effect" +import { Cause, DateTime, Effect, Exit, Layer } from "effect" +import { EventV2 } from "../../event" import { LocationServiceMap } from "../../location-service-map" import { makeGlobalNode } from "../../effect/app-node" +import { SessionEvent } from "../event" import { SessionRunCoordinator } from "../run-coordinator" import { SessionRunner } from "../runner" import { SessionSchema } from "../schema" @@ -13,6 +15,7 @@ const layer = Layer.effect( Effect.gen(function* () { const store = yield* SessionStore.Service const locations = yield* LocationServiceMap.Service + const events = yield* EventV2.Service const coordinator = yield* SessionRunCoordinator.make({ drain: Effect.fnUntraced(function* (sessionID: SessionSchema.ID, force) { const session = yield* store.get(sessionID) @@ -26,6 +29,24 @@ const layer = Layer.effect( ), ) }), + // One ExecutionSettled per execution (busy period), covering every coalesced drain. + settled: (sessionID, exit) => + Effect.gen(function* () { + const failure = + Exit.isFailure(exit) && !Cause.hasInterrupts(exit.cause) ? Cause.squash(exit.cause) : undefined + yield* events.publish(SessionEvent.ExecutionSettled, { + sessionID, + timestamp: yield* DateTime.now, + outcome: Exit.isSuccess(exit) ? "success" : Cause.hasInterrupts(exit.cause) ? "interrupted" : "failure", + error: + failure !== undefined + ? { type: "unknown", message: failure instanceof Error ? failure.message : String(failure) } + : undefined, + }) + }).pipe( + Effect.catchCause(() => Effect.void), + Effect.asVoid, + ), }) return SessionExecution.Service.of({ @@ -41,7 +62,7 @@ const layer = Layer.effect( export const node = makeGlobalNode({ service: SessionExecution.Service, layer, - deps: [SessionStore.node, LocationServiceMap.node], + deps: [SessionStore.node, LocationServiceMap.node, EventV2.node], }) export * as SessionExecutionLocal from "./local" diff --git a/packages/core/src/session/input.ts b/packages/core/src/session/input.ts index 14b613678db..2c6aeda460b 100644 --- a/packages/core/src/session/input.ts +++ b/packages/core/src/session/input.ts @@ -1,6 +1,6 @@ export * as SessionInput from "./input" -import { and, asc, eq, isNull, lte } from "drizzle-orm" +import { and, asc, eq, isNull } from "drizzle-orm" import { DateTime, Effect, Schema } from "effect" import { Admitted, Delivery } from "@opencode-ai/schema/session-input" import type { Database } from "../database/database" @@ -145,26 +145,11 @@ export const projectPrompted = Effect.fn("SessionInput.projectPrompted")(functio return } + // Every Prompted event is published from an admitted inbox row, so a missing or + // divergent row on replay is an invariant violation. const stored = yield* find(db, input.id) - if (stored) { - if (!matchesProjection(stored, input) || stored.promotedSeq !== input.promotedSeq) - return yield* Effect.die(new LifecycleConflict({ id: input.id })) - return - } - - yield* db - .insert(SessionInputTable) - .values({ - id: input.id, - session_id: input.sessionID, - prompt: encodePrompt(input.prompt), - delivery: input.delivery, - admitted_seq: input.promotedSeq, - promoted_seq: input.promotedSeq, - time_created: DateTime.toEpochMillis(input.timeCreated), - }) - .run() - .pipe(Effect.orDie) + if (!stored || !matchesProjection(stored, input) || stored.promotedSeq !== input.promotedSeq) + return yield* Effect.die(new LifecycleConflict({ id: input.id })) }) export const hasPending = Effect.fn("SessionInput.hasPending")(function* ( @@ -195,9 +180,8 @@ export const equivalent = ( readonly prompt: Prompt readonly delivery: Delivery }, -) => input.delivery === expected.delivery && matchesPrompt(input, expected) - -const matchesPrompt = (input: Admitted, expected: { readonly sessionID: SessionSchema.ID; readonly prompt: Prompt }) => +) => + input.delivery === expected.delivery && input.sessionID === expected.sessionID && JSON.stringify(encodePrompt(input.prompt)) === JSON.stringify(encodePrompt(expected.prompt)) @@ -246,7 +230,6 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* ( db: DatabaseService, events: EventV2.Interface, sessionID: SessionSchema.ID, - cutoff: number, ) { const rows = yield* db .select() @@ -256,7 +239,6 @@ export const promoteSteers = Effect.fn("SessionInput.promoteSteers")(function* ( eq(SessionInputTable.session_id, sessionID), isNull(SessionInputTable.promoted_seq), eq(SessionInputTable.delivery, "steer"), - lte(SessionInputTable.admitted_seq, cutoff), ), ) .orderBy(asc(SessionInputTable.admitted_seq)) diff --git a/packages/core/src/session/run-coordinator.ts b/packages/core/src/session/run-coordinator.ts index 8cba78b3f86..e7834b766b0 100644 --- a/packages/core/src/session/run-coordinator.ts +++ b/packages/core/src/session/run-coordinator.ts @@ -6,111 +6,117 @@ import { Deferred, Effect, Exit, Fiber, FiberSet, Scope } from "effect" export interface Coordinator { /** Snapshots keys with an execution owned by this coordinator. */ readonly active: Effect.Effect> - /** Starts execution while idle or joins the active execution. */ + /** Starts an execution while idle or joins the active execution. */ readonly run: (key: Key) => Effect.Effect /** Registers one coalesced follow-up after newly recorded work. */ readonly wake: (key: Key) => Effect.Effect - /** Stops active execution and waits for its cleanup. */ + /** Stops the active execution and waits for its cleanup. */ readonly interrupt: (key: Key) => Effect.Effect /** Resolves once no execution is active for the key. Returns immediately when already idle and never starts work. */ readonly awaitIdle: (key: Key) => Effect.Effect } -type Entry = { +/** + * One execution is a busy period for one key: one fiber that drains from the first wake + * until the key would stay idle. `pendingWake` is the doorbell: work recorded during the + * execution rings it, and the execution loop drains again instead of ending. The doorbell + * closes the gap between a drain's last eligibility check and the idle transition, since + * those cannot be one atomic step. `done` resolves joiners with this execution's exit. + */ +type Execution = { readonly done: Deferred.Deferred - owner?: Fiber.Fiber + owner?: Fiber.Fiber pendingWake: boolean stopping: boolean } export const make = (options: { readonly drain: (key: Key, force: boolean) => Effect.Effect + /** + * Runs in the execution fiber for every exit, including interruption, after the final + * drain and before the execution settles (waiters resolve after it completes). + */ + readonly settled?: (key: Key, exit: Exit.Exit) => Effect.Effect }): Effect.Effect, never, Scope.Scope> => Effect.gen(function* () { - const active = new Map>() + const executions = new Map>() const fork = yield* FiberSet.makeRuntime() - const makeEntry = (): Entry => ({ - done: Deferred.makeUnsafe(), - pendingWake: false, - stopping: false, - }) + const loop = (key: Key, execution: Execution, force: boolean): Effect.Effect => + Effect.suspend(() => options.drain(key, force)).pipe( + Effect.flatMap(() => + Effect.suspend(() => { + if (execution.stopping || !execution.pendingWake) return Effect.void + execution.pendingWake = false + // Trampoline so drains that complete synchronously cannot grow the stack. + return Effect.yieldNow.pipe(Effect.andThen(loop(key, execution, false))) + }), + ), + ) - const start = (key: Key, entry: Entry, force: boolean, successor = false) => { - const ready = Deferred.makeUnsafe() - const owner = fork( - (successor ? Effect.yieldNow : Deferred.await(ready)).pipe( - Effect.andThen(Effect.suspend(() => options.drain(key, force))), - Effect.onExit((exit) => Effect.sync(() => settle(key, entry, exit))), + const start = (key: Key, force: boolean) => { + const execution: Execution = { done: Deferred.makeUnsafe(), pendingWake: false, stopping: false } + executions.set(key, execution) + // The leading yield lets `owner` be assigned before the drain can settle, and keeps + // failing self-waking executions from growing the stack across successor starts. + // Drains start one tick after wake; callers observe progress through events or run. + execution.owner = fork( + Effect.yieldNow.pipe( + Effect.andThen(loop(key, execution, force)), + Effect.onExit((exit) => options.settled?.(key, exit) ?? Effect.void), + Effect.onExit((exit) => Effect.sync(() => settle(key, execution, exit))), Effect.exit, Effect.asVoid, ), ) - entry.owner = owner - if (!successor) Deferred.doneUnsafe(ready, Effect.void) + return execution } - const settle = (key: Key, entry: Entry, exit: Exit.Exit) => { - if (Exit.isSuccess(exit) && !entry.stopping && entry.pendingWake) { - entry.pendingWake = false - start(key, entry, false, true) - return - } - - const successor = entry.pendingWake ? makeEntry() : undefined - if (successor === undefined) active.delete(key) - else { - active.set(key, successor) - start(key, successor, false, true) - } - Deferred.doneUnsafe(entry.done, exit) + // A doorbell that survives the execution loop (rung after the loop decided to end, or + // during failure or interruption cleanup) starts a fresh execution for the remaining work. + const settle = (key: Key, execution: Execution, exit: Exit.Exit) => { + if (execution.pendingWake) start(key, false) + else executions.delete(key) + Deferred.doneUnsafe(execution.done, exit) } const run = (key: Key): Effect.Effect => Effect.uninterruptibleMask((restore) => { - const entry = active.get(key) - if (entry !== undefined) { - if (entry.stopping) return restore(Deferred.await(entry.done).pipe(Effect.andThen(run(key)))) - return restore(Deferred.await(entry.done)) + const execution = executions.get(key) + if (execution !== undefined) { + if (execution.stopping) return restore(Deferred.await(execution.done).pipe(Effect.andThen(run(key)))) + return restore(Deferred.await(execution.done)) } - - const next = makeEntry() - active.set(key, next) - start(key, next, true) - return restore(Deferred.await(next.done)) + return restore(Deferred.await(start(key, true).done)) }) const wake = (key: Key) => Effect.sync(() => { - const entry = active.get(key) - if (entry !== undefined) { - entry.pendingWake = true + const execution = executions.get(key) + if (execution !== undefined) { + execution.pendingWake = true return } - - const next = makeEntry() - active.set(key, next) - start(key, next, false) + start(key, false) }) const interrupt = (key: Key): Effect.Effect => Effect.suspend(() => { - const entry = active.get(key) - if (entry?.owner === undefined) return Effect.void - entry.stopping = true - entry.pendingWake = false - return Fiber.interrupt(entry.owner) + const execution = executions.get(key) + if (execution?.owner === undefined) return Effect.void + execution.stopping = true + execution.pendingWake = false + return Fiber.interrupt(execution.owner) }) - // Each successful drain reuses its entry.done across coalesced wakes, so one await - // already spans steered and queued continuation. Re-check after it settles to cover a - // fresh wake (or a failure/stopping successor) that installs a new entry. + // One execution's `done` already spans coalesced continuations; re-check after it + // settles to cover a successor execution started by a late doorbell. const awaitIdle = (key: Key): Effect.Effect => Effect.suspend(() => { - const entry = active.get(key) - if (entry === undefined) return Effect.void - return Deferred.await(entry.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key))) + const execution = executions.get(key) + if (execution === undefined) return Effect.void + return Deferred.await(execution.done).pipe(Effect.exit, Effect.andThen(awaitIdle(key))) }) - return { active: Effect.sync(() => new Set(active.keys())), run, wake, interrupt, awaitIdle } + return { active: Effect.sync(() => new Set(executions.keys())), run, wake, interrupt, awaitIdle } }) diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 458a3d9f6ff..5a5eff7ee80 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -34,7 +34,7 @@ import { SessionInput } from "../input" import { SessionSchema } from "../schema" import { SessionStore } from "../store" import { SessionTitle } from "../title" -import { type RunError, Service } from "./index" +import { Service } from "./index" import { SessionRunnerModel } from "./model" import { createLLMEventPublisher } from "./publish-llm-event" import { toLLMMessages } from "./to-llm-message" @@ -157,22 +157,6 @@ const layer = Layer.effect( const isQuestionRejected = (cause: Cause.Cause) => cause.reasons.some((reason) => Cause.isDieReason(reason) && reason.defect instanceof QuestionTool.RejectedError) - type TurnTransition = - // Automatic compaction completed; rebuild the request from compacted history. - | { readonly _tag: "ContinueAfterCompaction"; readonly step: number } - // Overflow compaction completed; rebuild once through the path without overflow recovery. - | { readonly _tag: "ContinueAfterOverflowCompaction"; readonly step: number } - - class TurnTransitionError extends Error { - constructor(readonly transition: TurnTransition) { - super() - } - } - - const continueAfterCompaction = (step: number) => new TurnTransitionError({ _tag: "ContinueAfterCompaction", step }) - const continueAfterOverflowCompaction = (step: number) => - new TurnTransitionError({ _tag: "ContinueAfterOverflowCompaction", step }) - const loadSystemContext = (agent: AgentV2.Selection, sessionID: SessionSchema.ID) => Effect.all( [ @@ -208,12 +192,11 @@ const layer = Layer.effect( let needsContinuation = false let currentStep = step if (promotion) { - const cutoff = yield* EventV2.latestSequence(db, session.id) let promoted = 0 - if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id, cutoff) + if (promotion === "steer") promoted = yield* SessionInput.promoteSteers(db, events, session.id) if (promotion === "queue") { promoted += Number(yield* SessionInput.promoteNextQueued(db, events, session.id)) - promoted += yield* SessionInput.promoteSteers(db, events, session.id, cutoff) + promoted += yield* SessionInput.promoteSteers(db, events, session.id) } if (promoted > 0) currentStep = 1 } @@ -239,8 +222,9 @@ const layer = Layer.effect( tools: toolMaterialization?.definitions ?? [], toolChoice: isLastStep ? "none" : undefined, }) + // Automatic compaction completed; rebuild the request from compacted history. if (yield* compaction.compactIfNeeded({ sessionID: session.id, messages: context, request })) - return yield* Effect.die(continueAfterCompaction(currentStep)) + return { _tag: "RestartAfterCompaction", step: currentStep } as const const startSnapshot = yield* snapshots.capture() const publisher = createLLMEventPublisher(events, { sessionID: session.id, @@ -303,44 +287,71 @@ const layer = Layer.effect( return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { + // Gather the evidence: how did the provider stream end? const stream = yield* restore(providerStream).pipe(Effect.exit) - const failure = - stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined + 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) + + // A context overflow before any assistant output is recoverable: compact and + // restart the turn instead of surfacing the provider error. if ( recoverOverflow && !publisher.hasAssistantStarted() && - isContextOverflowFailure(overflowFailure ?? failure) && + isContextOverflowFailure(overflowFailure ?? streamFailure) && (yield* restore(recoverOverflow({ sessionID: session.id, messages: context, request }))) ) - return yield* Effect.die(continueAfterOverflowCompaction(currentStep)) + return { _tag: "RestartAfterOverflowCompaction", step: currentStep } as const + + // An unrecovered held-back overflow becomes the turn'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. if (overflowFailure) yield* publish(overflowFailure) - const llmFailure = failure instanceof LLMError ? failure : undefined + const llmFailure = streamFailure instanceof LLMError ? streamFailure : undefined if (llmFailure && !publisher.hasProviderError()) { yield* publication.withPermit(publisher.failUnsettledTools("Provider did not return a tool result", true)) yield* publication.withPermit(publisher.failAssistant(llmFailure.reason.message)) } - if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers) + // 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. + if (streamInterrupted) yield* FiberSet.clear(toolFibers) const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit) - const streamInterrupted = stream._tag === "Failure" && Cause.hasInterrupts(stream.cause) const toolsInterrupted = settled._tag === "Failure" && Cause.hasInterrupts(settled.cause) - if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { + const questionDismissed = settled._tag === "Failure" && isQuestionRejected(settled.cause) + + if (questionDismissed || streamInterrupted || toolsInterrupted) { yield* FiberSet.clear(toolFibers) yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted")) yield* publication.withPermit(publisher.failAssistant("Provider turn interrupted")) - return yield* Effect.interrupt + // Match V1: dismissing a question halts the loop like an interruption. + if (questionDismissed) return yield* Effect.interrupt } - if (streamInterrupted || toolsInterrupted) { - yield* FiberSet.clear(toolFibers) - yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted")) - yield* publication.withPermit(publisher.failAssistant("Provider turn interrupted")) - } - if (settled._tag === "Failure" && !Cause.hasInterrupts(settled.cause)) { - const failure = Cause.squash(settled.cause) + // 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 turn 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 ? settled.cause : undefined + const infraError = + settledFailure === undefined ? undefined : Option.getOrUndefined(Cause.findErrorOption(settledFailure)) + if (settledFailure !== undefined) { + const failure = infraError ?? Cause.squash(settledFailure) const message = failure instanceof Error ? failure.message : String(failure) yield* publication.withPermit(publisher.failUnsettledTools(`Tool execution failed: ${message}`)) + if (infraError !== undefined) + yield* publication.withPermit(publisher.failAssistant(`Tool execution failed: ${message}`)) } + const stepSettlement = publisher.stepSettlement() - if (stepSettlement && !streamInterrupted && !toolsInterrupted && !publisher.hasProviderError()) { + if ( + stepSettlement && + !streamInterrupted && + !toolsInterrupted && + infraError === undefined && + !providerFailed + ) { const endSnapshot = yield* snapshots.capture() const files = startSnapshot && endSnapshot @@ -361,49 +372,43 @@ const layer = Layer.effect( }), ) } - if (publisher.hasProviderError()) - yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted")) - if (stream._tag === "Success" && !publisher.hasProviderError()) + // A provider error orphans recorded local calls; a clean stream can still leave + // hosted calls without results. + if (providerFailed) yield* publication.withPermit(publisher.failUnsettledTools("Tool execution interrupted")) + if (stream._tag === "Success" && !providerFailed) yield* publication.withPermit(publisher.failUnsettledTools("Provider did not return a tool result", true)) + if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause) - if (settled._tag === "Failure" && Cause.hasInterrupts(settled.cause)) + if (settled._tag === "Failure" && (toolsInterrupted || infraError !== undefined)) return yield* Effect.failCause(settled.cause) - return { needsContinuation: !publisher.hasProviderError() && needsContinuation, step: currentStep } + return { + _tag: "Completed", + needsContinuation: !providerFailed && needsContinuation, + step: currentStep, + } as const }), ) }, Effect.scoped) - type RunTurn = ( + + const runTurn = Effect.fnUntraced(function* ( sessionID: SessionSchema.ID, promotion: SessionInput.Delivery | undefined, step: number, - ) => Effect.Effect<{ readonly needsContinuation: boolean; readonly step: number }, RunError> - - const runAfterOverflowCompaction: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { - return yield* runTurnAttempt(sessionID, promotion, step).pipe( - Effect.catchDefect( - Effect.fnUntraced(function* (defect) { - if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect) - if (defect.transition._tag === "ContinueAfterOverflowCompaction") - return yield* Effect.die("Post-compaction provider attempt cannot recover another overflow") - yield* Effect.yieldNow - return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step) - }), - ), - ) - }) - - const runTurn: RunTurn = Effect.fnUntraced(function* (sessionID, promotion, step) { - return yield* runTurnAttempt(sessionID, promotion, step, compaction.compactAfterOverflow).pipe( - Effect.catchDefect( - Effect.fnUntraced(function* (defect) { - if (!(defect instanceof TurnTransitionError)) return yield* Effect.die(defect) - yield* Effect.yieldNow - if (defect.transition._tag === "ContinueAfterOverflowCompaction") - return yield* runAfterOverflowCompaction(sessionID, undefined, defect.transition.step) - return yield* runTurn(sessionID, undefined, defect.transition.step) - }), - ), - ) + ) { + // Compaction restarts rebuild the request from compacted history without re-promoting. + // Overflow recovery is one-shot: a post-compaction attempt must not recover another + // overflow, so the recovery hook is dropped after it fires. + let recoverOverflow: typeof compaction.compactAfterOverflow | undefined = compaction.compactAfterOverflow + let currentPromotion = promotion + let currentStep = step + while (true) { + const attempt = yield* runTurnAttempt(sessionID, currentPromotion, currentStep, recoverOverflow) + if (attempt._tag === "Completed") return { needsContinuation: attempt.needsContinuation, step: attempt.step } + if (attempt._tag === "RestartAfterOverflowCompaction") recoverOverflow = undefined + yield* Effect.yieldNow + currentPromotion = undefined + currentStep = attempt.step + } }) const drain = Effect.fnUntraced(function* (input: { @@ -437,32 +442,10 @@ const layer = Layer.effect( } }) - const run = Effect.fn("SessionRunner.run")( - (input: { readonly sessionID: SessionSchema.ID; readonly force: boolean }) => - drain(input).pipe( - Effect.onExit((exit) => - Effect.gen(function* () { - const failure = - Exit.isFailure(exit) && !Cause.hasInterrupts(exit.cause) ? Cause.squash(exit.cause) : undefined - yield* events.publish(SessionEvent.ExecutionSettled, { - sessionID: input.sessionID, - timestamp: yield* DateTime.now, - outcome: Exit.isSuccess(exit) ? "success" : Cause.hasInterrupts(exit.cause) ? "interrupted" : "failure", - error: - failure !== undefined - ? { type: "unknown", message: failure instanceof Error ? failure.message : String(failure) } - : undefined, - }) - }).pipe( - Effect.catchCause(() => Effect.void), - Effect.asVoid, - ), - ), - ), - ) - return Service.of({ - run, + // ExecutionSettled is published per execution (busy period) by SessionExecution, + // not per drain here. + run: Effect.fn("SessionRunner.run")(drain), }) }), ) diff --git a/packages/core/test/session-compact.test.ts b/packages/core/test/session-compact.test.ts index dc61a31e2f0..cee32ade046 100644 --- a/packages/core/test/session-compact.test.ts +++ b/packages/core/test/session-compact.test.ts @@ -81,11 +81,20 @@ describe("SessionV2.compact", () => { const events = yield* EventV2.Service const created = yield* session.create({ location }) + const messageID = SessionMessage.ID.create() + const prompt = Prompt.make({ text: "Please compact this session history." }) + yield* events.publish(SessionEvent.PromptAdmitted, { + sessionID: created.id, + messageID, + timestamp: DateTime.makeUnsafe(0), + prompt, + delivery: "steer", + }) yield* events.publish(SessionEvent.Prompted, { sessionID: created.id, - messageID: SessionMessage.ID.create(), + messageID, timestamp: DateTime.makeUnsafe(0), - prompt: Prompt.make({ text: "Please compact this session history." }), + prompt, delivery: "steer", }) diff --git a/packages/core/test/session-create.test.ts b/packages/core/test/session-create.test.ts index 90d460ecb79..9c41d1d9ac6 100644 --- a/packages/core/test/session-create.test.ts +++ b/packages/core/test/session-create.test.ts @@ -187,7 +187,7 @@ describe("SessionV2.create", () => { prompt: Prompt.make({ text: "First" }), resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, parent.id) yield* events.publish(SessionEvent.Synthetic, { sessionID: parent.id, messageID: SessionMessage.ID.create(), @@ -219,9 +219,9 @@ describe("SessionV2.create", () => { }) yield* session.prompt({ sessionID: parent.id, prompt: Prompt.make({ text: "Parent changed" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, parent.id) yield* session.prompt({ sessionID: forked.id, prompt: Prompt.make({ text: "Child continues" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, forked.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, forked.id) expect((yield* session.context(parent.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"]) expect((yield* session.context(forked.id)).map((message) => message.type)).toEqual(["user", "synthetic", "user"]) @@ -246,13 +246,13 @@ describe("SessionV2.create", () => { prompt: Prompt.make({ text: "First" }), resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, parent.id) const second = yield* session.prompt({ sessionID: parent.id, prompt: Prompt.make({ text: "Second" }), resume: false, }) - yield* SessionInput.promoteSteers(db, events, parent.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, parent.id) const forked = yield* session.fork({ sessionID: parent.id, messageID: second.id }) @@ -368,7 +368,7 @@ describe("SessionV2.create", () => { const { db } = yield* Database.Service const created = yield* session.create({ location }) yield* session.prompt({ sessionID: created.id, prompt: Prompt.make({ text: "Hello" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, created.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, created.id) expect( Array.from(yield* logEvents(session, created.id, true).pipe(Stream.take(2), Stream.runCollect)), @@ -390,7 +390,7 @@ describe("SessionV2.create", () => { prompt: Prompt.make({ text: "Replay lifecycle" }), resume: false, }) - yield* SessionInput.promoteSteers(sourceDb, sourceEvents, created.id, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(sourceDb, sourceEvents, created.id) const serialized = (yield* sourceDb .select() .from(EventTable) diff --git a/packages/core/test/session-projector.test.ts b/packages/core/test/session-projector.test.ts index 7b6f2f219ea..4191a587151 100644 --- a/packages/core/test/session-projector.test.ts +++ b/packages/core/test/session-projector.test.ts @@ -129,6 +129,13 @@ describe("SessionProjector", () => { .pipe(Effect.orDie) const events = yield* EventV2.Service + yield* events.publish(SessionEvent.PromptAdmitted, { + sessionID, + messageID: SessionMessage.ID.make("msg_first"), + timestamp: created, + prompt: Prompt.make({ text: "first" }), + delivery: "steer", + }) yield* events.publish( SessionEvent.Prompted, { @@ -140,6 +147,13 @@ describe("SessionProjector", () => { }, { id: EventV2.ID.make("evt_z") }, ) + yield* events.publish(SessionEvent.PromptAdmitted, { + sessionID, + messageID: SessionMessage.ID.make("msg_second"), + timestamp: created, + prompt: Prompt.make({ text: "second" }), + delivery: "steer", + }) yield* events.publish( SessionEvent.Prompted, { diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index baf3c303b2d..58e26b7b5cf 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -196,7 +196,7 @@ describe("SessionV2.prompt", () => { prompt: Prompt.make({ text: "boundary" }), resume: false, }) - yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, sessionID) const stale = SessionMessage.ID.make("msg_stale_assistant") yield* db.insert(SessionMessageTable).values(assistantRow(stale, 100)).run().pipe(Effect.orDie) yield* events.publish(SessionEvent.RevertEvent.Staged, { @@ -253,7 +253,7 @@ describe("SessionV2.prompt", () => { yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) - yield* SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers(db, events, sessionID) const streamed = Array.from(yield* Fiber.join(fiber)) expect(streamed.map((event): [number | undefined, string] => [event.durable?.seq, event.type])).toEqual([ @@ -425,10 +425,7 @@ describe("SessionV2.prompt", () => { yield* session.prompt({ id: messageID, sessionID, prompt: Prompt.make({ text: "Promote once" }), resume: false }) yield* Effect.all( - [ - SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER), - SessionInput.promoteSteers(db, events, sessionID, Number.MAX_SAFE_INTEGER), - ], + [SessionInput.promoteSteers(db, events, sessionID), SessionInput.promoteSteers(db, events, sessionID)], { concurrency: "unbounded" }, ) @@ -440,23 +437,6 @@ describe("SessionV2.prompt", () => { }), ) - it.effect("promotes steers only through the captured inbox cutoff", () => - Effect.gen(function* () { - yield* setup - const { db } = yield* Database.Service - const session = yield* SessionV2.Service - const events = yield* EventV2.Service - const first = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Before cutoff" }), resume: false }) - const cutoff = first.admittedSeq - const second = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "After cutoff" }), resume: false }) - - yield* SessionInput.promoteSteers(db, events, sessionID, cutoff) - - expect(yield* admitted(first.id)).toHaveProperty("promotedSeq") - expect(yield* admitted(second.id)).not.toHaveProperty("promotedSeq") - }), - ) - it.effect("reprojects pending inbox input without scheduling execution", () => Effect.gen(function* () { yield* setup @@ -500,48 +480,6 @@ describe("SessionV2.prompt", () => { }), ) - it.effect("returns an exact retry of a legacy projected prompt", () => - Effect.gen(function* () { - yield* setup - const session = yield* SessionV2.Service - const events = yield* EventV2.Service - const prompt = Prompt.make({ text: "Historical prompt" }) - yield* events.publish(SessionEvent.Prompted, { - sessionID, - messageID, - timestamp: yield* DateTime.now, - prompt, - delivery: "steer", - }) - - const retried = yield* session.prompt({ id: messageID, sessionID, prompt, resume: false }) - - expect(retried).toMatchObject({ id: messageID, prompt: { text: "Historical prompt" } }) - expect(yield* admitted(messageID)).toHaveProperty("promotedSeq") - }), - ) - - it.effect("returns an exact retry of a legacy projected queued prompt", () => - Effect.gen(function* () { - yield* setup - const session = yield* SessionV2.Service - const events = yield* EventV2.Service - const prompt = Prompt.make({ text: "Historical queued prompt" }) - yield* events.publish(SessionEvent.Prompted, { - sessionID, - messageID, - timestamp: yield* DateTime.now, - prompt, - delivery: "queue", - }) - - const retried = yield* session.prompt({ id: messageID, sessionID, prompt, delivery: "queue", resume: false }) - - expect(retried).toMatchObject({ id: messageID, prompt: { text: "Historical queued prompt" } }) - expect(yield* admitted(messageID)).toMatchObject({ delivery: "queue" }) - }), - ) - it.effect("rejects reuse of one globally unique message ID across sessions", () => Effect.gen(function* () { yield* setup diff --git a/packages/core/test/session-run-coordinator.test.ts b/packages/core/test/session-run-coordinator.test.ts index dfbeda664c5..ecc4e613ac2 100644 --- a/packages/core/test/session-run-coordinator.test.ts +++ b/packages/core/test/session-run-coordinator.test.ts @@ -392,6 +392,68 @@ describe("SessionRunCoordinator", () => { ), ) + it.effect("settles once per execution across coalesced drains", () => + Effect.scoped( + Effect.gen(function* () { + const started = yield* Deferred.make() + const gate = yield* Deferred.make() + const idle = yield* Deferred.make() + let drains = 0 + const settled: Exit.Exit[] = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => + Effect.sync(() => ++drains).pipe( + Effect.flatMap((run) => + run === 1 + ? Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(gate))) + : Effect.void, + ), + Effect.asVoid, + ), + settled: (_key, exit) => + Effect.sync(() => void settled.push(exit)).pipe( + Effect.andThen(Deferred.succeed(idle, undefined)), + Effect.asVoid, + ), + }) + + yield* coordinator.wake("session") + yield* Deferred.await(started) + yield* coordinator.wake("session") + yield* Deferred.succeed(gate, undefined) + yield* Deferred.await(idle) + + expect(drains).toBe(2) + expect(settled).toHaveLength(1) + expect(Exit.isSuccess(settled[0]!)).toBe(true) + }), + ), + ) + + it.effect("settles interrupted executions before waiters resolve", () => + Effect.scoped( + Effect.gen(function* () { + const started = yield* Deferred.make() + const gate = yield* Deferred.make() + const settled: Exit.Exit[] = [] + const coordinator = yield* SessionRunCoordinator.make({ + drain: () => Deferred.succeed(started, undefined).pipe(Effect.andThen(Deferred.await(gate))), + settled: (_key, exit) => Effect.sync(() => void settled.push(exit)), + }) + + yield* coordinator.wake("session") + yield* Deferred.await(started) + yield* coordinator.interrupt("session") + + expect(settled).toHaveLength(1) + expect(settled[0] !== undefined && Exit.isFailure(settled[0]) && Cause.hasInterrupts(settled[0].cause)).toBe( + true, + ) + expect(yield* coordinator.active).toEqual(new Set()) + }), + ), + ) + it.effect("trampolines synchronous self-waking execution", () => Effect.scoped( Effect.gen(function* () { diff --git a/packages/core/test/session-runner.test.ts b/packages/core/test/session-runner.test.ts index b2c58ba6104..42321120941 100644 --- a/packages/core/test/session-runner.test.ts +++ b/packages/core/test/session-runner.test.ts @@ -156,6 +156,14 @@ const echo = Layer.effectDiscard( output: Schema.Struct({}), execute: () => 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. + storefail: Tool.make({ + description: "Produce output that cannot be persisted", + input: Schema.Struct({}), + output: Schema.Any, + execute: () => Effect.succeed({ big: 1n }), + }), }), ), ) @@ -653,6 +661,7 @@ describe("SessionRunnerLLM", () => { response = [] const message = yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Run automatically" }) }) + yield* session.wait(sessionID) expect(requests).toHaveLength(1) expect(yield* session.messages({ sessionID })).toMatchObject([ @@ -677,7 +686,7 @@ describe("SessionRunnerLLM", () => { expect(requests).toHaveLength(1) expect(requests[0]?.model).toBe(model) - expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect"]) + expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect", "storefail"]) expect(requests[0]?.messages.map((message) => ({ role: message.role, content: message.content }))).toEqual([ { role: "user", content: [{ type: "text", text: "First" }] }, { role: "user", content: [{ type: "text", text: "Second" }] }, @@ -712,6 +721,7 @@ describe("SessionRunnerLLM", () => { systemUnavailable = false yield* session.prompt({ id: messageID, sessionID, prompt: Prompt.make({ text: "First" }) }) + yield* session.wait(sessionID) expect(requests).toHaveLength(1) expect(requests[0]?.messages.map((message) => message.role)).toEqual(["user"]) @@ -1553,7 +1563,7 @@ describe("SessionRunnerLLM", () => { yield* session.resume(sessionID) expect(requests).toHaveLength(1) - expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect"]) + expect(requests[0]?.tools.map((tool) => tool.name)).toEqual(["echo", "defect", "storefail"]) expect(yield* session.context(sessionID)).toMatchObject([ { type: "user", text: "Use tools" }, { @@ -2389,7 +2399,7 @@ describe("SessionRunnerLLM", () => { const session = yield* SessionV2.Service const events = yield* EventV2.Service yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Recover interrupted tool" }), resume: false }) - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -2453,7 +2463,7 @@ describe("SessionRunnerLLM", () => { prompt: Prompt.make({ text: "Recover interrupted hosted tool" }), resume: false, }) - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -2513,7 +2523,7 @@ describe("SessionRunnerLLM", () => { prompt: Prompt.make({ text: "Recover interrupted tool input" }), resume: false, }) - yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID, Number.MAX_SAFE_INTEGER) + yield* SessionInput.promoteSteers((yield* Database.Service).db, events, sessionID) const assistantMessageID = SessionMessage.ID.create() yield* events.publish(SessionEvent.Step.Started, { sessionID, @@ -2626,8 +2636,9 @@ describe("SessionRunnerLLM", () => { const first = yield* session.resume(sessionID).pipe(Effect.forkChild) yield* Deferred.await(streamStarted) + streamStarted = yield* Deferred.make() const second = yield* session.resume(otherSessionID).pipe(Effect.forkChild) - yield* Effect.yieldNow + yield* Deferred.await(streamStarted) expect(requests).toHaveLength(2) expect(requests.map((request) => request.providerOptions?.openai?.promptCacheKey)).toEqual([ @@ -2797,6 +2808,49 @@ describe("SessionRunnerLLM", () => { }), ) + it.effect("fails the drain when tool output persistence fails", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Call storefail" }), resume: false }) + + requests.length = 0 + responses = [ + [ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolCall({ id: "call-storefail", name: "storefail", input: {} }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + LLMEvent.finish({ reason: "tool-calls" }), + ], + [], + ] + + const exit = yield* session.resume(sessionID).pipe(Effect.exit) + + expect(Exit.isFailure(exit)).toBe(true) + expect(requests).toHaveLength(1) + expect(yield* session.context(sessionID)).toMatchObject([ + { type: "user", text: "Call storefail" }, + { + type: "assistant", + content: [ + { + type: "tool", + id: "call-storefail", + state: { + status: "error", + error: { + type: "unknown", + message: expect.stringContaining("Tool execution failed: Failed to encode tool output"), + }, + }, + }, + ], + }, + ]) + }), + ) + it.effect("interrupts runner continuation when a question is dismissed", () => Effect.gen(function* () { yield* setup diff --git a/packages/core/test/session-title.test.ts b/packages/core/test/session-title.test.ts index ec8a9818b0b..1e9bcfbeb82 100644 --- a/packages/core/test/session-title.test.ts +++ b/packages/core/test/session-title.test.ts @@ -41,7 +41,14 @@ const models = Layer.mock(SessionRunnerModel.Service)({ }) const it = testEffect( AppNodeBuilder.build( - LayerNode.group([Database.node, EventV2.node, SessionProjector.node, SessionStore.node, AgentV2.node, SessionTitle.node]), + LayerNode.group([ + Database.node, + EventV2.node, + SessionProjector.node, + SessionStore.node, + AgentV2.node, + SessionTitle.node, + ]), [ [llmClient, client], [SessionRunnerModel.node, models], @@ -76,9 +83,17 @@ const insertSession = (id: SessionV2.ID) => const prompt = (sessionID: SessionV2.ID, text: string) => Effect.gen(function* () { const events = yield* EventV2.Service + const messageID = SessionMessage.ID.create() + yield* events.publish(SessionEvent.PromptAdmitted, { + sessionID, + messageID, + timestamp: DateTime.makeUnsafe(0), + prompt: Prompt.make({ text }), + delivery: "steer", + }) yield* events.publish(SessionEvent.Prompted, { sessionID, - messageID: SessionMessage.ID.create(), + messageID, timestamp: DateTime.makeUnsafe(0), prompt: Prompt.make({ text }), delivery: "steer", @@ -101,9 +116,9 @@ it.effect("generates a title from the sole user message and renames the session" yield* prompt(sessionID, "Help me debug the failing build") const store = yield* SessionStore.Service - const session = yield* store.get(sessionID).pipe( - Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))), - ) + const session = yield* store + .get(sessionID) + .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing")))) const title = yield* SessionTitle.Service yield* title.generateForFirstPrompt(session) @@ -131,9 +146,9 @@ it.effect("does not generate once a second user message exists", () => yield* prompt(sessionID, "Second message") const store = yield* SessionStore.Service - const session = yield* store.get(sessionID).pipe( - Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))), - ) + const session = yield* store + .get(sessionID) + .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing")))) const title = yield* SessionTitle.Service yield* title.generateForFirstPrompt(session) @@ -179,9 +194,9 @@ it.effect("does not generate for a child session", () => yield* prompt(sessionID, "Do this subtask") const store = yield* SessionStore.Service - const session = yield* store.get(sessionID).pipe( - Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))), - ) + const session = yield* store + .get(sessionID) + .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing")))) const title = yield* SessionTitle.Service yield* title.generateForFirstPrompt(session) @@ -197,9 +212,9 @@ it.effect("does not generate when the title agent is removed", () => yield* prompt(sessionID, "Help me debug the failing build") const store = yield* SessionStore.Service - const session = yield* store.get(sessionID).pipe( - Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing"))), - ) + const session = yield* store + .get(sessionID) + .pipe(Effect.flatMap((session) => (session ? Effect.succeed(session) : Effect.die("session missing")))) const title = yield* SessionTitle.Service yield* title.generateForFirstPrompt(session)