From 98899f1612f11d44eabe753ff173cafd2029973d Mon Sep 17 00:00:00 2001 From: lex Date: Fri, 7 Aug 2026 10:55:55 +0800 Subject: [PATCH] =?UTF-8?q?feat(core):=20session=20runner=20hot=20path=20?= =?UTF-8?q?=E2=80=94=20turn=20timeout,=20incremental=20history,=20snapshot?= =?UTF-8?q?=20dedupe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - agent 配置新增 timeout 字段（NonNegativeInt；运行时 0/未设回退默认 600s，杜绝 0=立即超时）；resolveTurnTimeout 按 Config.entries lowest→highest 取最后匹配（findLast，修全局配置压过项目配置的优先级反转） - DEFAULT_PROVIDER_TURN_TIMEOUT 独立常量（packages/core 不跨包引用 opencode dag 配置，注释说明） - history 增量读取（afterSeq 游标）+ decode 回归 Schema.decodeUnknownEffect typed 错误通道（去 try/catch 与 as 断言） - runner 光标缓存随 drain 全出口 ensuring 逐出，baselineSeq 校验防陈旧；批处理 withBatch 收敛 - snapshot 助手更名 captureDeduped（名实相符：始终 capture、tree ID 去重），调用点同步 - 测试：history-incremental、session-runner-hotpath（busy-wait 改有界 waitUntil+timeoutOrElse，stub 按 test AGENTS.md 约定）、tool-events 适配 --- packages/core/src/config/agent.ts | 6 +- packages/core/src/session/history.ts | 46 +- packages/core/src/session/runner/llm.ts | 232 ++++++- packages/core/src/snapshot.ts | 16 + .../test/session-runner-tool-events.test.ts | 13 + .../test/session/history-incremental.test.ts | 235 +++++++ .../session/session-runner-hotpath.test.ts | 582 ++++++++++++++++++ 7 files changed, 1093 insertions(+), 37 deletions(-) create mode 100644 packages/core/test/session/history-incremental.test.ts create mode 100644 packages/core/test/session/session-runner-hotpath.test.ts diff --git a/packages/core/src/config/agent.ts b/packages/core/src/config/agent.ts index 63df995f85..14556c2ebb 100644 --- a/packages/core/src/config/agent.ts +++ b/packages/core/src/config/agent.ts @@ -3,7 +3,7 @@ export * as ConfigAgent from "./agent" import { Schema } from "effect" import { Permission } from "@opencode-ai/schema/permission" import { ConfigProvider } from "./provider" -import { PositiveInt } from "../schema" +import { NonNegativeInt, PositiveInt } from "../schema" export const Color = Schema.Union([ Schema.String.check(Schema.isPattern(/^#[0-9a-fA-F]{6}$/)), @@ -20,6 +20,10 @@ export class Info extends Schema.Class("ConfigV2.Agent")({ hidden: Schema.Boolean.pipe(Schema.optional), color: Color.pipe(Schema.optional), steps: PositiveInt.pipe(Schema.optional), + timeout: NonNegativeInt.pipe(Schema.optional).annotate({ + description: + "Provider turn timeout in seconds for sessions running this agent (default 600). Bounds the provider stream and the tool-wait after it.", + }), disabled: Schema.Boolean.pipe(Schema.optional), permissions: Permission.Ruleset.pipe(Schema.optional), }) {} diff --git a/packages/core/src/session/history.ts b/packages/core/src/session/history.ts index fb55ab0756..b3f718c1ce 100644 --- a/packages/core/src/session/history.ts +++ b/packages/core/src/session/history.ts @@ -26,6 +26,7 @@ const messageRows = Effect.fnUntraced(function* ( sessionID: SessionSchema.ID, compaction: { readonly seq: number } | undefined, baselineSeq?: number, + afterSeq?: number, ) { const rows = yield* db .select() @@ -44,6 +45,7 @@ const messageRows = Effect.fnUntraced(function* ( baselineSeq === undefined ? undefined : or(ne(SessionMessageTable.type, "system"), gt(SessionMessageTable.seq, baselineSeq)), + afterSeq === undefined ? undefined : gt(SessionMessageTable.seq, afterSeq), ), ) .orderBy(asc(SessionMessageTable.seq)) @@ -63,6 +65,11 @@ const decodeMessageRow = (row: typeof SessionMessageTable.$inferSelect) => ), ) +const decodeEntries = (rows: typeof SessionMessageTable.$inferSelect[]) => + Effect.forEach(rows, (row) => + decodeMessageRow(row).pipe(Effect.map((message) => ({ seq: row.seq, message }))), + ) + export const load = Effect.fn("SessionHistory.load")(function* (db: DatabaseService, sessionID: SessionSchema.ID) { const [epoch, compaction] = yield* Effect.all( [ @@ -76,7 +83,8 @@ export const load = Effect.fn("SessionHistory.load")(function* (db: DatabaseServ ], { concurrency: "unbounded" }, ) - return yield* Effect.forEach(yield* messageRows(db, sessionID, compaction, epoch?.baselineSeq), decodeMessageRow) + const entries = yield* decodeEntries(yield* messageRows(db, sessionID, compaction, epoch?.baselineSeq)) + return entries.map((entry) => entry.message) }) export const loadForRunner = Effect.fn("SessionHistory.loadForRunner")(function* ( @@ -93,9 +101,39 @@ export const entriesForRunner = Effect.fn("SessionHistory.entriesForRunner")(fun baselineSeq: number, ) { const rows = yield* messageRows(db, sessionID, yield* latestCompaction(db, sessionID), baselineSeq) - return yield* Effect.forEach(rows, (row) => - decodeMessageRow(row).pipe(Effect.map((message) => ({ seq: row.seq, message }))), - ) + return yield* decodeEntries(rows) +}) + +/** + * Incremental read for the runner hot path: returns only entries with + * `seq > afterSeq` (the caller's last-read cursor), so a session of length N + * costs O(new messages) per turn instead of a full O(N) scan. + * + * - `entries` are subject to the same compaction and epoch-baseline filters as + * `entriesForRunner`, so appending them to the caller's cached entries is + * equivalent to a fresh full read. + * - `lastSeq` is the highest `seq` returned (unchanged when nothing new was + * written) and doubles as the next `afterSeq`. + * - `reset` is true when a compaction has crossed the cursor since the last + * read. Compaction changes the read window (`seq >= compaction.seq`), so the + * caller must discard its cached entries and replace them with `entries`, + * which already contain the full read in that case. + * + * Epoch-baseline changes are reported by the caller (it owns the epoch) and + * are not detected here. + */ +export const entriesAfter = Effect.fn("SessionHistory.entriesAfter")(function* ( + db: DatabaseService, + sessionID: SessionSchema.ID, + baselineSeq: number, + afterSeq: number, +) { + const compaction = yield* latestCompaction(db, sessionID) + const reset = compaction !== undefined && compaction.seq > afterSeq + const rows = yield* messageRows(db, sessionID, compaction, baselineSeq, reset ? undefined : afterSeq) + const entries = yield* decodeEntries(rows) + const lastSeq = entries.length === 0 ? afterSeq : entries[entries.length - 1].seq + return { entries, lastSeq, reset } }) export * as SessionHistory from "./history" diff --git a/packages/core/src/session/runner/llm.ts b/packages/core/src/session/runner/llm.ts index 7dd87587d0..cd77d184de 100644 --- a/packages/core/src/session/runner/llm.ts +++ b/packages/core/src/session/runner/llm.ts @@ -5,10 +5,11 @@ import { LLMEvent, Message, SystemPart, + TransportReason, isContextOverflowFailure, type ProviderErrorEvent, } from "@opencode-ai/llm" -import { Cause, DateTime, Deferred, Effect, FiberSet, Layer, Option, Semaphore, Stream } from "effect" +import { Cause, DateTime, Deferred, Duration, Effect, FiberSet, Layer, Option, Semaphore, Stream } from "effect" import { AgentV2 } from "../../agent" import { Config } from "../../config" import { Database } from "../../database/database" @@ -28,6 +29,7 @@ import { SessionCompaction } from "../compaction" import { SessionEvent } from "../event" import { SessionHistory } from "../history" import { SessionInput } from "../input" +import { SessionMessage } from "../message" import { SessionSchema } from "../schema" import { SessionStore } from "../store" import { type RunError, Service } from "./index" @@ -37,6 +39,34 @@ import { toLLMMessages } from "./to-llm-message" import { MAX_STEPS_PROMPT } from "./max-steps" import { Snapshot } from "../../snapshot" +// Runner-level per-turn provider deadline. This is the runner's own cutoff +// (10 minutes); it is independent of the DAG node timeout +// (packages/opencode/src/dag/dag.ts DEFAULT_WORKFLOW_CONFIG.nodeTimeoutMs), +// which is an orchestration concern this package does not reference. Agents +// may override it per session via the `agents..timeout` config field +// (seconds). +// +// Coverage: the deadline bounds (1) the per-request HTTP transport timeout +// (`request.http.timeout`), (2) the total provider-stream turn below, and (3) +// the tool-fiber wait that follows the stream. (3) uses the same duration but +// is applied separately AFTER the provider turn completes, so a turn lasts at +// most ~2× the deadline — a hung tool can no longer hang a turn forever. +const DEFAULT_PROVIDER_TURN_TIMEOUT = Duration.minutes(10) + +const turnTimeoutError = () => + new LLMError({ + module: "SessionRunner", + method: "stream", + reason: new TransportReason({ message: "Provider turn timed out", kind: "Timeout" }), + }) + +const toolWaitTimeoutError = () => + new LLMError({ + module: "SessionRunner", + method: "stream", + reason: new TransportReason({ message: "Tool execution timed out", kind: "Timeout" }), + }) + /** * Runs one durable coding-agent Session until it settles. * @@ -113,6 +143,102 @@ export const layer = Layer.effect( const getContext = Effect.fn("SessionRunner.getContext")(function* (sessionID: SessionSchema.ID) { return yield* store.context(sessionID) }) + + type HistoryCursor = { + readonly baselineSeq: number + readonly lastSeq: number + readonly entries: readonly { readonly seq: number; readonly message: SessionMessage.Message }[] + readonly snapshots: { last: Snapshot.ID | undefined } + } + // Within-drain incremental-read cache. Each session's entry is evicted when + // its run settles (see `run` below), so this never grows to + // O(sessions × history) for the location's lifetime. A later run re-reads + // the full view from the store — one extra read per run, never a + // correctness change (snapshots are content-addressed, and the baseline + // revalidation below still guards stale epochs). + const cursors = new Map() + + // Resolve the per-agent provider-turn deadline: the `agents..timeout` + // config field (seconds) wins, otherwise the runner default. Config entries + // run lowest-to-highest priority (Config.Interface.entries), so the latest + // matching document wins — matching Config.latest / options findLast. A + // value of 0 is treated as unset so it falls back to the default instead of + // timing the turn out immediately. + const resolveTurnTimeout = Effect.fnUntraced(function* (agentID: AgentV2.ID) { + let resolved: number | undefined + for (const document of yield* config.entries()) { + if (document.type !== "document") continue + const timeout = document.info.agents?.[agentID]?.timeout + if (timeout !== undefined && timeout > 0) resolved = timeout + } + return resolved === undefined ? DEFAULT_PROVIDER_TURN_TIMEOUT : Duration.seconds(resolved) + }) + + // Incremental history read for the hot path: the first read (or any epoch + // baseline change) loads the full runner view and establishes the cursor; + // later turns read only entries after the cursor. A compaction reset signal + // replaces the cached entries with the full read returned by the API. + const readHistory = Effect.fnUntraced(function* (sessionID: SessionSchema.ID, baselineSeq: number) { + const cached = cursors.get(sessionID) + if (cached === undefined || cached.baselineSeq !== baselineSeq) { + const entries = yield* SessionHistory.entriesForRunner(db, sessionID, baselineSeq) + const cursor: HistoryCursor = { + baselineSeq, + lastSeq: entries.at(-1)?.seq ?? 0, + entries, + snapshots: { last: cached?.snapshots.last }, + } + cursors.set(sessionID, cursor) + return { entries: cursor.entries, snapshots: cursor.snapshots } + } + const result = yield* SessionHistory.entriesAfter(db, sessionID, baselineSeq, cached.lastSeq) + if (result.reset) { + const cursor: HistoryCursor = { + baselineSeq, + lastSeq: result.lastSeq, + entries: result.entries, + snapshots: cached.snapshots, + } + cursors.set(sessionID, cursor) + return { entries: cursor.entries, snapshots: cursor.snapshots } + } + const entries = [...cached.entries, ...result.entries] + cursors.set(sessionID, { ...cached, lastSeq: result.lastSeq, entries }) + return { entries, snapshots: cached.snapshots } + }) + + // Batch wrapper for the publisher's durable events: live-only events + // (streaming deltas) flush the pending durable batch first so pubsub order + // matches publish order, then publish immediately. Durable events are + // committed through EventV2.publishMany at deterministic boundaries. + const withBatch = (events: EventV2.Interface) => { + let buffer: EventV2.BatchEvent[] = [] + const flush = Effect.fnUntraced(function* () { + yield* Effect.uninterruptible( + Effect.gen(function* () { + const batch = buffer + buffer = [] + if (batch.length === 0) return + yield* events.publishMany(batch) + }), + ) + }) + const publish = ( + definition: D, + data: EventV2.Data, + options?: EventV2.PublishOptions, + ) => + definition?.durable + ? Effect.sync(() => { + buffer.push({ definition, data, options }) + return { id: options?.id ?? EventV2.ID.create(), type: definition.type, data } as EventV2.Payload + }) + : flush().pipe(Effect.andThen(() => events.publish(definition, data, options))) + return { + events: { ...events, publish }, + flush, + } + } const failInterruptedTools = Effect.fn("SessionRunner.failInterruptedTools")(function* ( sessionID: SessionSchema.ID, ) { @@ -183,6 +309,7 @@ export const layer = Layer.effect( if (session.location.directory !== location.directory || session.location.workspaceID !== location.workspaceID) return yield* Effect.interrupt const agent = yield* agents.select(session.agent) + const turnTimeout = yield* resolveTurnTimeout(agent.id) const initialized = yield* SessionContextEpoch.initialize(db, loadSystemContext(agent), session.id) const toolFibers = yield* FiberSet.make() let needsContinuation = false @@ -200,7 +327,8 @@ export const layer = Layer.effect( const system = initialized ?? (yield* SessionContextEpoch.prepare(db, events, loadSystemContext(agent), session.id)) const model = yield* models.resolve(session) - const entries = yield* SessionHistory.entriesForRunner(db, session.id, system.baselineSeq) + const history = yield* readHistory(session.id, system.baselineSeq) + const entries = history.entries const context = entries.map((entry) => entry.message) const isLastStep = agent.info?.steps !== undefined && currentStep >= agent.info.steps const toolMaterialization = isLastStep ? undefined : yield* tools.materialize(agent.info?.permissions) @@ -208,6 +336,7 @@ export const layer = Layer.effect( const request = LLM.request({ model, providerOptions: { openai: { promptCacheKey } }, + http: { timeout: turnTimeout }, system: [agent.info?.system, system.baseline] .filter((part): part is string => part !== undefined && part.length > 0) .map(SystemPart.make), @@ -217,8 +346,9 @@ export const layer = Layer.effect( }) if (yield* compaction.compactIfNeeded({ sessionID: session.id, entries, model, request })) return yield* Effect.die(continueAfterCompaction(currentStep)) - const startSnapshot = yield* snapshots.capture() - const publisher = createLLMEventPublisher(events, { + const startSnapshot = yield* Snapshot.captureDeduped(history.snapshots, snapshots.capture) + const batch = withBatch(events) + const publisher = createLLMEventPublisher(batch.events, { sessionID: session.id, agent: agent.id, model: { @@ -250,6 +380,7 @@ export const layer = Layer.effect( } needsContinuation = true const assistantMessageID = yield* publisher.assistantMessageID(event.id) + yield* withPublication(batch.flush()) yield* Effect.uninterruptibleMask((restore) => restore( toolMaterialization.settle({ @@ -274,12 +405,26 @@ export const layer = Layer.effect( ).pipe(FiberSet.run(toolFibers)) }), ), - Effect.ensuring(withPublication(publisher.flush())), + Effect.ensuring( + withPublication( + Effect.gen(function* () { + yield* publisher.flush() + yield* batch.flush() + }), + ), + ), ) return yield* Effect.uninterruptibleMask((restore) => Effect.gen(function* () { - const stream = yield* restore(providerStream).pipe(Effect.exit) + const stream = yield* restore( + providerStream.pipe( + Effect.timeoutOrElse({ + duration: turnTimeout, + orElse: () => Effect.fail(turnTimeoutError()), + }), + ), + ).pipe(Effect.exit) const failure = stream._tag === "Failure" ? Option.getOrUndefined(Cause.findErrorOption(stream.cause)) : undefined if ( @@ -296,10 +441,22 @@ export const layer = Layer.effect( yield* withPublication(publisher.failAssistant(llmFailure.reason.message)) } if (stream._tag === "Failure" && Cause.hasInterrupts(stream.cause)) yield* FiberSet.clear(toolFibers) - const settled = yield* restore(awaitToolFibers(toolFibers)).pipe(Effect.exit) + // The tool wait is bounded by the same per-agent deadline, applied + // separately after the provider turn: a hung tool fails the turn + // instead of hanging it forever. Remaining tool fibers are + // interrupted by the runTurnAttempt scope close. + const settled = yield* restore( + awaitToolFibers(toolFibers).pipe( + Effect.timeoutOrElse({ + duration: turnTimeout, + orElse: () => Effect.fail(toolWaitTimeoutError()), + }), + ), + ).pipe(Effect.exit) if (settled._tag === "Failure" && isQuestionRejected(settled.cause)) { yield* FiberSet.clear(toolFibers) yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) + yield* withPublication(batch.flush()) return yield* Effect.interrupt } if ( @@ -318,15 +475,17 @@ export const layer = Layer.effect( } const stepSettlement = publisher.stepSettlement() if (stepSettlement && !publisher.hasProviderError()) { - const endSnapshot = yield* snapshots.capture() + const endSnapshot = yield* Snapshot.captureDeduped(history.snapshots, snapshots.capture) const files = - startSnapshot && endSnapshot - ? yield* snapshots - .files({ from: startSnapshot, to: endSnapshot }) - .pipe(Effect.catch(() => Effect.succeed(undefined))) - : undefined + startSnapshot === undefined || endSnapshot === undefined + ? undefined + : startSnapshot === endSnapshot + ? [] + : yield* snapshots + .files({ from: startSnapshot, to: endSnapshot }) + .pipe(Effect.catch(() => Effect.succeed(undefined))) yield* withPublication( - events.publish(SessionEvent.Step.Ended, { + batch.events.publish(SessionEvent.Step.Ended, { sessionID: session.id, timestamp: yield* DateTime.now, assistantMessageID: yield* publisher.startAssistant(), @@ -342,6 +501,7 @@ export const layer = Layer.effect( yield* withPublication(publisher.failUnsettledTools("Tool execution interrupted")) if (stream._tag === "Success" && !publisher.hasProviderError()) yield* withPublication(publisher.failUnsettledTools("Provider did not return a tool result", true)) + yield* withPublication(batch.flush()) if (stream._tag === "Failure") return yield* Effect.failCause(stream.cause) if (settled._tag === "Failure") return yield* Effect.failCause(settled.cause) return { needsContinuation: !publisher.hasProviderError() && needsContinuation, step: currentStep } @@ -386,25 +546,33 @@ export const layer = Layer.effect( readonly sessionID: SessionSchema.ID readonly force: boolean }) { - const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer") - const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue") - if (!input.force && !hasSteer && !hasQueue) return - yield* failInterruptedTools(input.sessionID) - let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined - let shouldRun = input.force || hasSteer || hasQueue - while (shouldRun) { - let needsContinuation = true - let step = 1 - while (needsContinuation) { - const result = yield* runTurn(input.sessionID, promotion, step) - needsContinuation = result.needsContinuation - step = result.step + 1 - promotion = "steer" - if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer") + // Drain-body wrapped with cursor eviction: the incremental-read cache is + // only useful while this drain runs, so its entry is dropped on every + // exit (success, failure, interrupt, or early no-work return). A later + // run falls back to a full store read — correct, and one read per run. + // Concurrent same-session drains (not expected under the run + // coordinator) degrade to full reads, never to stale history. + return yield* Effect.gen(function* () { + const hasSteer = yield* SessionInput.hasPending(db, input.sessionID, "steer") + const hasQueue = hasSteer ? false : yield* SessionInput.hasPending(db, input.sessionID, "queue") + if (!input.force && !hasSteer && !hasQueue) return + yield* failInterruptedTools(input.sessionID) + let promotion: SessionInput.Delivery | undefined = hasSteer ? "steer" : hasQueue ? "queue" : undefined + let shouldRun = input.force || hasSteer || hasQueue + while (shouldRun) { + let needsContinuation = true + let step = 1 + while (needsContinuation) { + const result = yield* runTurn(input.sessionID, promotion, step) + needsContinuation = result.needsContinuation + step = result.step + 1 + promotion = "steer" + if (!needsContinuation) needsContinuation = yield* SessionInput.hasPending(db, input.sessionID, "steer") + } + shouldRun = yield* SessionInput.hasPending(db, input.sessionID, "queue") + promotion = shouldRun ? "queue" : undefined } - shouldRun = yield* SessionInput.hasPending(db, input.sessionID, "queue") - promotion = shouldRun ? "queue" : undefined - } + }).pipe(Effect.ensuring(Effect.sync(() => cursors.delete(input.sessionID)))) }) return Service.of({ diff --git a/packages/core/src/snapshot.ts b/packages/core/src/snapshot.ts index 631bca2a23..21862d945d 100644 --- a/packages/core/src/snapshot.ts +++ b/packages/core/src/snapshot.ts @@ -240,6 +240,22 @@ export const noopLayer = Layer.succeed( }), ) +/** + * Hot-path snapshot dedupe: run the capture, but reuse the previous tree ID + * when the fresh capture returns the same content-addressed tree, so identical + * consecutive states never produce a new snapshot identity and callers can + * skip the downstream diff computation for unchanged trees. + */ +export const captureDeduped = ( + state: { last: ID | undefined }, + capture: () => Effect.Effect, +): Effect.Effect => + Effect.map(capture(), (id) => { + if (id === undefined || id === state.last) return state.last + state.last = id + return id + }) + function failure(operation: Error["operation"], cause: unknown) { if (cause instanceof Error && cause.operation === operation) return cause return new Error({ diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index f96ea4dea2..cf51a18a4f 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -26,6 +26,19 @@ const capture = () => { }) return event }), + publishMany: (events) => + Effect.sync(() => + events.map(({ definition, data }) => { + const event = { id: EventV2.ID.create(), type: definition.type, data } as EventV2.Payload + published.push({ + type: definition.durable + ? EventV2.versionedType(definition.type, definition.durable.version) + : definition.type, + data, + }) + return event + }), + ), subscribe: () => Stream.empty, all: () => Stream.empty, durable: () => Stream.empty, diff --git a/packages/core/test/session/history-incremental.test.ts b/packages/core/test/session/history-incremental.test.ts new file mode 100644 index 0000000000..076fa3bd63 --- /dev/null +++ b/packages/core/test/session/history-incremental.test.ts @@ -0,0 +1,235 @@ +import { describe, expect } from "bun:test" +import { Database } from "@opencode-ai/core/database/database" +import { ModelV2 } from "@opencode-ai/core/model" +import { ProjectV2 } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionHistory } from "@opencode-ai/core/session/history" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionSchema } from "@opencode-ai/core/session/schema" +import { SessionMessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { Effect, DateTime, Schema } from "effect" +import { testEffect } from "../lib/effect" + +const it = testEffect(Database.defaultLayer) + +const projectID = ProjectV2.ID.global +const sessionID = SessionSchema.ID.create() +const created = DateTime.makeUnsafe(0) +const id = (value: string) => SessionMessage.ID.make(`msg_${value}`) + +const user = (text: string) => + SessionMessage.User.make({ id: id(text), type: "user", text, time: { created } }) + +const system = (text: string) => + SessionMessage.System.make({ id: id(text), type: "system", text, time: { created } }) + +const assistant = (text: string) => + SessionMessage.Assistant.make({ + id: id(text), + type: "assistant", + agent: "build", + model: { id: ModelV2.ID.make("model"), providerID: ProviderV2.ID.make("provider") }, + content: [SessionMessage.AssistantText.make({ type: "text", id: id(`${text}-part`), text })], + time: { created, completed: created }, + }) + +const compaction = (summary: string) => + SessionMessage.Compaction.make({ + id: id(`compaction-${summary}`), + type: "compaction", + reason: "auto", + summary, + recent: summary, + time: { created }, + }) + +const setup = (db: Database.Interface["db"]) => + Effect.gen(function* () { + yield* db + .insert(ProjectTable) + .values({ + id: projectID, + worktree: AbsolutePath.make("/project"), + sandboxes: [AbsolutePath.make("/project")], + }) + .run() + .pipe(Effect.orDie) + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: projectID, + slug: sessionID, + directory: "/project", + title: "history", + version: "1", + }) + .run() + .pipe(Effect.orDie) + }) + +const insertMessage = (db: Database.Interface["db"], seq: number, message: SessionMessage.Message) => { + const { id: messageID, type, ...data } = Schema.encodeSync(SessionMessage.Message)(message) + return db + .insert(SessionMessageTable) + .values({ + id: SessionMessage.ID.make(messageID), + session_id: sessionID, + type, + seq, + time_created: DateTime.toEpochMillis(message.time.created), + data, + }) + .run() + .pipe(Effect.orDie) +} + +describe("SessionHistory.entriesAfter", () => { + it.effect("returns only messages written after the cursor", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* setup(db) + yield* insertMessage(db, 1, user("one")) + yield* insertMessage(db, 2, assistant("two")) + yield* insertMessage(db, 3, user("three")) + + const empty = yield* SessionHistory.entriesAfter(db, sessionID, 0, 3) + expect(empty.reset).toBe(false) + expect(empty.entries).toEqual([]) + expect(empty.lastSeq).toBe(3) + + yield* insertMessage(db, 4, user("four")) + const one = yield* SessionHistory.entriesAfter(db, sessionID, 0, 3) + expect(one.reset).toBe(false) + expect(one.entries.map((entry) => entry.seq)).toEqual([4]) + expect(one.entries[0]?.message.type).toBe("user") + expect(one.lastSeq).toBe(4) + + yield* insertMessage(db, 5, assistant("five")) + yield* insertMessage(db, 6, user("six")) + const two = yield* SessionHistory.entriesAfter(db, sessionID, 0, 4) + expect(two.reset).toBe(false) + expect(two.entries.map((entry) => entry.seq)).toEqual([5, 6]) + expect(two.lastSeq).toBe(6) + }), + ) + + it.effect("is equivalent to a full read when the cursor is advanced incrementally", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* setup(db) + yield* insertMessage(db, 1, user("one")) + yield* insertMessage(db, 2, assistant("two")) + yield* insertMessage(db, 3, user("three")) + yield* insertMessage(db, 4, system("context")) + yield* insertMessage(db, 5, assistant("five")) + const baseline = 3 + + const first = yield* SessionHistory.entriesForRunner(db, sessionID, baseline) + let entries = first + let lastSeq = first.length === 0 ? 0 : first[first.length - 1]!.seq + + yield* insertMessage(db, 6, assistant("six")) + yield* insertMessage(db, 7, user("seven")) + let result = yield* SessionHistory.entriesAfter(db, sessionID, baseline, lastSeq) + expect(result.reset).toBe(false) + entries = [...entries, ...result.entries] + lastSeq = result.lastSeq + expect(entries.map((entry) => entry.seq)).toEqual([1, 2, 3, 4, 5, 6, 7]) + expect(entries).toEqual(yield* SessionHistory.entriesForRunner(db, sessionID, baseline)) + + yield* insertMessage(db, 8, system("new-context")) + yield* insertMessage(db, 9, user("nine")) + result = yield* SessionHistory.entriesAfter(db, sessionID, baseline, lastSeq) + expect(result.reset).toBe(false) + entries = [...entries, ...result.entries] + lastSeq = result.lastSeq + expect(entries.map((entry) => entry.seq)).toEqual([1, 2, 3, 4, 5, 6, 7, 8, 9]) + expect(entries).toEqual(yield* SessionHistory.entriesForRunner(db, sessionID, baseline)) + expect(lastSeq).toBe(9) + }), + ) + + it.effect("signals reset and returns the full read when a compaction crosses the cursor", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* setup(db) + yield* insertMessage(db, 1, user("one")) + yield* insertMessage(db, 2, user("two")) + yield* insertMessage(db, 3, assistant("three")) + + const first = yield* SessionHistory.entriesForRunner(db, sessionID, 0) + expect(first.map((entry) => entry.seq)).toEqual([1, 2, 3]) + + yield* insertMessage(db, 4, compaction("summary")) + yield* insertMessage(db, 5, user("five")) + const result = yield* SessionHistory.entriesAfter(db, sessionID, 0, 3) + expect(result.reset).toBe(true) + expect(result.entries.map((entry) => entry.seq)).toEqual([4, 5]) + expect(result.entries[0]?.message.type).toBe("compaction") + expect(result.lastSeq).toBe(5) + expect(result.entries).toEqual(yield* SessionHistory.entriesForRunner(db, sessionID, 0)) + + const settled = yield* SessionHistory.entriesAfter(db, sessionID, 0, result.lastSeq) + expect(settled.reset).toBe(false) + expect(settled.entries).toEqual([]) + + yield* insertMessage(db, 6, user("six")) + const next = yield* SessionHistory.entriesAfter(db, sessionID, 0, 5) + expect(next.reset).toBe(false) + expect(next.entries.map((entry) => entry.seq)).toEqual([6]) + expect(next.entries).toEqual((yield* SessionHistory.entriesForRunner(db, sessionID, 0)).slice(2)) + }), + ) + + it.effect("keeps the epoch baseline filter on the incremental path", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* setup(db) + yield* insertMessage(db, 1, system("stale-context")) + yield* insertMessage(db, 2, user("two")) + yield* insertMessage(db, 3, system("current-context")) + yield* insertMessage(db, 4, assistant("four")) + const baseline = 3 + + const full = yield* SessionHistory.entriesForRunner(db, sessionID, baseline) + expect(full.map((entry) => entry.seq)).toEqual([2, 4]) + + const cached = full.filter((entry) => entry.seq <= 2) + const result = yield* SessionHistory.entriesAfter(db, sessionID, baseline, 2) + expect(result.reset).toBe(false) + expect(result.entries.map((entry) => entry.seq)).toEqual([4]) + expect([...cached, ...result.entries].map((entry) => entry.seq)).toEqual([2, 4]) + + yield* insertMessage(db, 5, system("new-context")) + const next = yield* SessionHistory.entriesAfter(db, sessionID, baseline, 4) + expect(next.reset).toBe(false) + expect(next.entries.map((entry) => entry.seq)).toEqual([5]) + expect(next.entries.map((entry) => entry.message.type)).toEqual(["system"]) + expect([...cached, ...result.entries, ...next.entries].map((entry) => entry.seq)).toEqual([2, 4, 5]) + expect([...cached, ...result.entries, ...next.entries]).toEqual( + yield* SessionHistory.entriesForRunner(db, sessionID, baseline), + ) + }), + ) + + it.effect("fails with MessageDecodeError on an undecodable row", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* setup(db) + yield* insertMessage(db, 1, user("one")) + yield* db + .insert(SessionMessageTable) + .values({ id: id("corrupt"), session_id: sessionID, type: "user", seq: 2, data: {} as never }) + .run() + .pipe(Effect.orDie) + + const error = yield* SessionHistory.entriesAfter(db, sessionID, 0, 0).pipe(Effect.flip) + expect(error._tag).toBe("Session.MessageDecodeError") + expect(error.messageID).toBe(id("corrupt")) + expect(error.sessionID).toBe(sessionID) + }), + ) +}) diff --git a/packages/core/test/session/session-runner-hotpath.test.ts b/packages/core/test/session/session-runner-hotpath.test.ts new file mode 100644 index 0000000000..1d49e2a54f --- /dev/null +++ b/packages/core/test/session/session-runner-hotpath.test.ts @@ -0,0 +1,582 @@ +import { describe, expect } from "bun:test" +import { + LLMClient, + LLMError, + LLMEvent, + Model, + TransportReason, + type LLMClientShape, + type LLMRequest, +} from "@opencode-ai/llm" +import * as OpenAIChat from "@opencode-ai/llm/protocols/openai-chat" +import { Database } from "@opencode-ai/core/database/database" +import { EventV2 } from "@opencode-ai/core/event" +import { EventTable } from "@opencode-ai/core/event/sql" +import { PermissionV2 } from "@opencode-ai/core/permission" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { SessionV2 } from "@opencode-ai/core/session" +import { LocationServiceMap } from "@opencode-ai/core/location-layer" +import { Snapshot } from "@opencode-ai/core/snapshot" +import { Prompt } from "@opencode-ai/core/session/prompt" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionExecution } from "@opencode-ai/core/session/execution" +import { SessionRunCoordinator } from "@opencode-ai/core/session/run-coordinator" +import { SessionRunner } from "@opencode-ai/core/session/runner" +import * as SessionRunnerLLM from "@opencode-ai/core/session/runner/llm" +import { SessionRunnerModel } from "@opencode-ai/core/session/runner/model" +import { ToolRegistry } from "@opencode-ai/core/tool/registry" +import { ToolOutputStore } from "@opencode-ai/core/tool-output-store" +import { ApplicationTools } from "@opencode-ai/core/tool/application-tools" +import { AgentV2 } from "@opencode-ai/core/agent" +import { Config } from "@opencode-ai/core/config" +import { ConfigAgent } from "@opencode-ai/core/config/agent" +import { Tool } from "@opencode-ai/core/tool/tool" +import { SessionEvent } from "@opencode-ai/core/session/event" +import { SessionMessage } from "@opencode-ai/core/session/message" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { SessionStore } from "@opencode-ai/core/session/store" +import { SystemContext } from "@opencode-ai/core/system-context" +import { SystemContextRegistry } from "@opencode-ai/core/system-context/registry" +import { SkillGuidance } from "@opencode-ai/core/skill/guidance" +import { ReferenceGuidance } from "@opencode-ai/core/reference/guidance" +import { Location } from "@opencode-ai/core/location" +import { Cause, DateTime, Deferred, Duration, Effect, Exit, Fiber, Layer, Schema, Stream } from "effect" +import { and, asc, eq } from "drizzle-orm" +import * as TestClock from "effect/testing/TestClock" +import { testEffect } from "../lib/effect" + +const sessionID = SessionV2.ID.make("ses_runner_hotpath") +const requests: LLMRequest[] = [] +let response: LLMEvent[] = [] +let responses: LLMEvent[][] | undefined +let responseStream: Stream.Stream | undefined +const client = Layer.succeed( + LLMClient.Service, + LLMClient.Service.of({ + prepare: () => Effect.die("unused"), + stream: ((request: LLMRequest) => { + requests.push(request) + if (responseStream) { + const stream = responseStream + responseStream = undefined + return stream + } + return Stream.fromIterable(responses === undefined ? response : (responses.shift() ?? [])) + }) as unknown as LLMClientShape["stream"], + generate: () => Effect.die("unused"), + }), +) + +// Counts EventV2 usage at the service boundary so the runner's batching is observable. +const counts = { publish: 0, publishMany: 0 } +const events = Layer.effect( + EventV2.Service, + Effect.gen(function* () { + const service = yield* EventV2.Service + return EventV2.Service.of({ + ...service, + publish: ( + definition: D, + data: EventV2.Data, + options?: EventV2.PublishOptions, + ) => { + counts.publish++ + return service.publish(definition, data, options) + }, + publishMany: (batch: ReadonlyArray, options?: { readonly location?: Location.Ref }) => { + counts.publishMany++ + return service.publishMany(batch, options) + }, + }) + }), +).pipe(Layer.provide(EventV2.defaultLayer)) + +// Scripted snapshots: capture returns the next queued tree ID (default "tree-1", +// i.e. an unchanged tree), files records the compared pair. +const probe = { captures: new Array(), captureCalls: 0, filesCalls: 0, filesPairs: [] as string[][] } +const snapshot = Layer.succeed( + Snapshot.Service, + Snapshot.Service.of({ + capture: () => + Effect.sync(() => { + probe.captureCalls++ + const value = probe.captures.length > 0 ? probe.captures.shift()! : "tree-1" + return value === undefined ? undefined : Snapshot.ID.make(value) + }), + files: ({ from, to }) => + Effect.sync(() => { + probe.filesCalls++ + probe.filesPairs.push([String(from), String(to)]) + return [] + }), + diff: () => Effect.succeed([]), + preview: () => Effect.succeed([]), + restore: () => Effect.void, + checkout: () => Effect.void, + }), +) + +// The tool is gated on a Deferred so the test can inspect the durable event +// table while the side effect is running, proving Tool.Called is committed +// before execution. +const executions: string[] = [] +let toolExecutionGate: Deferred.Deferred | undefined +const permission = Layer.mock(PermissionV2.Service, { + assert: () => Effect.die("unused"), + ask: () => Effect.die("unused"), + reply: () => Effect.die("unused"), + get: () => Effect.die("unused"), + forSession: () => Effect.die("unused"), + list: () => Effect.die("unused"), +}) +const applications = ApplicationTools.layer +const registry = ToolRegistry.layer.pipe( + Layer.provide(permission), + Layer.provide(applications), + Layer.provide(ToolOutputStore.defaultLayer), +) +const echo = Layer.effectDiscard( + ToolRegistry.Service.use((registry) => + registry.register({ + echo: Tool.make({ + description: "Echo text", + input: Schema.Struct({ text: Schema.String }), + output: Schema.Struct({ text: Schema.String }), + toModelOutput: ({ output }) => [{ type: "text", text: output.text }], + execute: ({ text }, _context) => + Effect.gen(function* () { + executions.push(text) + if (toolExecutionGate) yield* Deferred.await(toolExecutionGate) + return { text } + }), + }), + }), + ), +).pipe(Layer.provide(registry)) +const agents = AgentV2.layer +const model = Model.make({ id: "fake-model", provider: "fake", route: OpenAIChat.route }) +const models = SessionRunnerModel.layerWith(() => Effect.succeed(model)) +const systemContext = SystemContextRegistry.layer +const location = Location.layer({ directory: AbsolutePath.make("/project") }).pipe(Layer.provide(Project.defaultLayer)) +const skillGuidance = Layer.mock(SkillGuidance.Service, { load: () => Effect.succeed(SystemContext.empty) }) +const referenceGuidance = Layer.mock(ReferenceGuidance.Service, { load: () => Effect.succeed(SystemContext.empty) }) +// Config documents are read lazily per turn by the runner, so tests can set +// this before a run to exercise per-agent timeout resolution. +let configEntries: Config.Entry[] = [] +const config = Layer.succeed(Config.Service, Config.Service.of({ entries: () => Effect.succeed(configEntries) })) +const runner = SessionRunnerLLM.layer.pipe( + Layer.provide(snapshot), + Layer.provide(Database.defaultLayer), + Layer.provide(SessionStore.defaultLayer), + Layer.provide(events), + Layer.provide(client), + Layer.provide(registry), + Layer.provide(models), + Layer.provide(systemContext), + Layer.provide(location), + Layer.provide(agents), + Layer.provide(skillGuidance), + Layer.provide(referenceGuidance), + Layer.provide(config), +) +const execution = Layer.effect( + SessionExecution.Service, + Effect.gen(function* () { + const sessionRunner = yield* SessionRunner.Service + const coordinator = yield* SessionRunCoordinator.make({ + drain: (sessionID, force) => sessionRunner.run({ sessionID, force }), + }) + return SessionExecution.Service.of({ + resume: coordinator.run, + wake: coordinator.wake, + interrupt: coordinator.interrupt, + }) + }), +).pipe(Layer.provide(runner)) +const sessions = SessionV2.layer.pipe( + Layer.provide(LocationServiceMap.layer), + Layer.provide(events), + Layer.provide(Database.defaultLayer), + Layer.provide(SessionStore.defaultLayer), + Layer.provide(Project.defaultLayer), + Layer.provide(execution), +) +const it = testEffect( + Layer.mergeAll( + Database.defaultLayer, + events, + SessionProjector.defaultLayer, + SessionStore.defaultLayer, + client, + permission, + applications, + agents, + registry, + echo, + models, + systemContext, + location, + skillGuidance, + referenceGuidance, + config, + runner, + execution, + sessions, + ), +) + +const textTurn = (id: string, text: string): LLMEvent[] => [ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.textStart({ id }), + LLMEvent.textDelta({ id, text }), + LLMEvent.textEnd({ id }), + LLMEvent.stepFinish({ index: 0, reason: "stop" }), + LLMEvent.finish({ reason: "stop" }), +] + +const toolTurn: LLMEvent[] = [ + LLMEvent.stepStart({ index: 0 }), + LLMEvent.toolInputStart({ id: "call-echo", name: "echo" }), + LLMEvent.toolInputDelta({ id: "call-echo", name: "echo", text: '{"text":"Hi"}' }), + LLMEvent.toolInputEnd({ id: "call-echo", name: "echo" }), + LLMEvent.toolCall({ id: "call-echo", name: "echo", input: { text: "Hi" } }), + LLMEvent.stepFinish({ index: 0, reason: "tool-calls" }), + LLMEvent.finish({ reason: "tool-calls" }), +] + +const insertSession = (id: SessionV2.ID) => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db + .insert(SessionTable) + .values({ + id, + project_id: Project.ID.global, + slug: id, + directory: "/project", + title: "test", + version: "test", + }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + }) + +const setup = Effect.gen(function* () { + const { db } = yield* Database.Service + response = [] + responses = undefined + responseStream = undefined + requests.length = 0 + counts.publish = 0 + counts.publishMany = 0 + probe.captures = [] + probe.captureCalls = 0 + probe.filesCalls = 0 + probe.filesPairs = [] + executions.length = 0 + toolExecutionGate = undefined + configEntries = [] + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .onConflictDoNothing() + .run() + .pipe(Effect.orDie) + yield* insertSession(sessionID) +}) + +const durableEventTypes = (id: SessionV2.ID) => + Effect.gen(function* () { + const { db } = yield* Database.Service + return (yield* db + .select({ type: EventTable.type }) + .from(EventTable) + .where(eq(EventTable.aggregate_id, id)) + .orderBy(asc(EventTable.seq)) + .all()).map((event) => event.type) + }) + +const stepEndedData = (id: SessionV2.ID) => + Effect.gen(function* () { + const { db } = yield* Database.Service + return (yield* db + .select({ data: EventTable.data }) + .from(EventTable) + .where(and(eq(EventTable.aggregate_id, id), eq(EventTable.type, "session.next.step.ended.2"))) + .orderBy(asc(EventTable.seq)) + .all()).map((event) => event.data) + }) + +// Bounded readiness poll: yields to forked fibers so they can publish the +// awaited side effect (TestClock-neutral — it does not depend on virtual +// time), and fails loudly instead of spinning forever if the side effect never +// lands. The timeout is a safety net for live runs; under TestClock the loop +// terminates via the condition once the forked fiber has run. +const waitUntil = (check: Effect.Effect, message: string) => + Effect.gen(function* () { + while (!(yield* check)) yield* Effect.yieldNow + }).pipe( + Effect.timeoutOrElse({ duration: "5 seconds", orElse: () => Effect.fail(new Error(message)) }), + ) + +describe("SessionRunnerLLM hot path", () => { + it.effect("batches durable publishes and preserves order across incremental turns", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) + response = textTurn("text-first", "Hello") + counts.publish = 0 + counts.publishMany = 0 + yield* session.resume(sessionID) + + // Text turn: one batch per flush boundary (Text.Started; Text.Ended; + // Step.Ended) instead of one transaction per durable event; only the + // live delta goes through the single-event publish path. + expect(counts.publishMany).toBe(3) + expect(counts.publish).toBe(2) + + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) + responses = [toolTurn, textTurn("text-done", "Done")] + counts.publish = 0 + counts.publishMany = 0 + const gate = yield* Deferred.make() + toolExecutionGate = gate + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* waitUntil(Effect.sync(() => executions.length >= 1), "echo tool never started") + const { db } = yield* Database.Service + const committed = (yield* db + .select({ id: EventTable.id }) + .from(EventTable) + .where(eq(EventTable.type, "session.next.tool.called.1")) + .all() + .pipe(Effect.orDie)).length + + // Tool.Called was durably committed before the side effect started. + expect(committed).toBe(1) + yield* Deferred.succeed(gate, undefined) + yield* Fiber.await(run) + + // Tool turn: 3 batches (step+input start, input-end+called flushed before + // execution, tool success + step ended) plus 3 batches for the + // continuation text turn; the tool input delta and text delta are the only + // live publishes. + expect(counts.publishMany).toBe(6) + expect(counts.publish).toBe(3) + + // Incremental history is equivalent to the full read: each turn's request + // carries the complete prior conversation. + expect(requests).toHaveLength(3) + expect(requests[1]?.messages.map((message) => message.role)).toEqual(["user", "assistant", "user"]) + expect(requests[2]?.messages.map((message) => message.role)).toEqual([ + "user", + "assistant", + "user", + "assistant", + "tool", + ]) + expect(yield* session.context(sessionID)).toMatchObject([ + { type: "user", text: "First" }, + { type: "assistant", finish: "stop", content: [{ type: "text", text: "Hello" }] }, + { type: "user", text: "Second" }, + { type: "assistant", content: [{ type: "tool", id: "call-echo", name: "echo", state: { status: "completed" } }] }, + { type: "assistant", finish: "stop", content: [{ type: "text", text: "Done" }] }, + ]) + + // Durable event sequence matches the pre-batching order (prompt admission + // events are published by the session layer, not the runner). + const types = yield* durableEventTypes(sessionID) + const runnerEvents = types.filter((type) => !type.includes("prompt")) + expect(runnerEvents).toEqual([ + "session.next.step.started.1", + "session.next.text.started.1", + "session.next.text.ended.1", + "session.next.step.ended.2", + "session.next.step.started.1", + "session.next.tool.input.started.1", + "session.next.tool.input.ended.1", + "session.next.tool.called.1", + "session.next.tool.success.1", + "session.next.step.ended.2", + "session.next.step.started.1", + "session.next.text.started.1", + "session.next.text.ended.1", + "session.next.step.ended.2", + ]) + + // Unchanged tree: every step reuses the previous tree ID and the diff + // computation is skipped (files is the empty diff of identical trees). + expect(probe.filesCalls).toBe(0) + const ended = yield* stepEndedData(sessionID) + expect(ended).toHaveLength(3) + for (const data of ended) { + expect(data.snapshot).toBe("tree-1") + expect(data.files).toEqual([]) + } + }), + ) + + it.effect("resets the history cursor after compaction", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + const eventService = yield* EventV2.Service + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) + response = textTurn("text-first", "Hello") + yield* session.resume(sessionID) + const compactionID = SessionMessage.ID.create() + yield* eventService.publish(SessionEvent.Compaction.Started, { + sessionID, + messageID: compactionID, + timestamp: DateTime.makeUnsafe(1), + reason: "manual", + }) + yield* eventService.publish(SessionEvent.Compaction.Ended, { + sessionID, + messageID: compactionID, + timestamp: DateTime.makeUnsafe(2), + reason: "manual", + text: "summary", + recent: "", + }) + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) + requests.length = 0 + response = textTurn("text-second", "Again") + yield* session.resume(sessionID) + + expect(requests).toHaveLength(1) + const userTexts = requests[0]!.messages + .filter((message) => message.role === "user") + .flatMap((message) => + message.content.filter((content): content is { type: "text"; text: string } => content.type === "text").map( + (content) => content.text, + ), + ) + expect(userTexts[0]).toContain("") + expect(userTexts[0]).toContain("summary") + expect(userTexts[1]).toBe("Second") + // The compaction moved the read window: pre-compaction messages are gone + // from the request, proving the cursor reset re-read from the compaction. + expect(userTexts.join(" ")).not.toContain("First") + }), + ) + + it.effect("reuses snapshot IDs and skips unchanged-tree diffs across steps", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) + response = textTurn("text-first", "Hello") + yield* session.resume(sessionID) + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "Second" }), resume: false }) + response = textTurn("text-second", "Again") + yield* session.resume(sessionID) + + expect(probe.captureCalls).toBe(4) + expect(probe.filesCalls).toBe(0) + const ended = yield* stepEndedData(sessionID) + expect(ended.map((data) => data.snapshot)).toEqual(["tree-1", "tree-1"]) + expect(ended.map((data) => data.files)).toEqual([[], []]) + }), + ) + + it.effect("computes real files for a changed tree", () => + Effect.gen(function* () { + yield* setup + probe.captures = ["tree-1", "tree-2"] + const session = yield* SessionV2.Service + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) + response = textTurn("text-first", "Hello") + yield* session.resume(sessionID) + + expect(probe.filesCalls).toBe(1) + expect(probe.filesPairs).toEqual([["tree-1", "tree-2"]]) + const [ended] = yield* stepEndedData(sessionID) + expect(ended.snapshot).toBe("tree-2") + expect(ended.files).toEqual([]) + }), + ) + + it.effect("fails a hung provider turn through the provider failure path after the deadline", () => + Effect.gen(function* () { + yield* setup + const session = yield* SessionV2.Service + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) + responseStream = Stream.never + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* waitUntil(Effect.sync(() => requests.length >= 1), "provider stream never started") + yield* TestClock.adjust(Duration.minutes(11)) + const exit = yield* Fiber.await(run) + + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) { + const error = Cause.squash(exit.cause) + expect(error).toBeInstanceOf(LLMError) + if (error instanceof LLMError) { + expect(error.reason._tag).toBe("Transport") + if (error.reason._tag === "Transport") { + expect(error.reason.message).toBe("Provider turn timed out") + expect(error.reason.kind).toBe("Timeout") + } + } + } + expect(requests).toHaveLength(1) + expect(yield* session.context(sessionID)).toMatchObject([ + { type: "user", text: "First" }, + { type: "assistant", finish: "error", error: { type: "unknown", message: "Provider turn timed out" } }, + ]) + }), + ) + + it.effect("applies the configured agent timeout and bounds a hung tool wait", () => + Effect.gen(function* () { + yield* setup + // 1-second turn deadline via the `agents.build.timeout` config field + // (seconds); the DAG-default mirror is 10 minutes when unset. + configEntries = [new Config.Document({ type: "document", info: { agents: { build: new ConfigAgent.Info({ timeout: 1 }) } } })] + const session = yield* SessionV2.Service + yield* session.prompt({ sessionID, prompt: Prompt.make({ text: "First" }), resume: false }) + response = toolTurn + const gate = yield* Deferred.make() + toolExecutionGate = gate + const run = yield* session.resume(sessionID).pipe(Effect.forkChild) + yield* waitUntil(Effect.sync(() => requests.length >= 1), "provider stream never started") + expect(Duration.toSeconds(requests[0]!.http!.timeout!)).toBe(1) + // The tool call started and is stuck on the never-released gate; the + // provider stream itself has finished (only the tool wait remains). + yield* waitUntil(Effect.sync(() => executions.length >= 1), "echo tool never started") + yield* TestClock.adjust(Duration.seconds(2)) + const exit = yield* Fiber.await(run) + + expect(Exit.isFailure(exit)).toBe(true) + if (Exit.isFailure(exit)) { + const error = Cause.squash(exit.cause) + expect(error).toBeInstanceOf(LLMError) + if (error instanceof LLMError) { + expect(error.reason._tag).toBe("Transport") + if (error.reason._tag === "Transport") { + expect(error.reason.message).toBe("Tool execution timed out") + expect(error.reason.kind).toBe("Timeout") + } + } + } + expect(yield* session.context(sessionID)).toMatchObject([ + { type: "user", text: "First" }, + { + type: "assistant", + content: [ + { + type: "tool", + id: "call-echo", + name: "echo", + state: { status: "error", error: { type: "unknown", message: "Tool execution failed: SessionRunner.stream: Tool execution timed out" } }, + }, + ], + }, + ]) + }), + ) +})