From 79a190eeccaa3dd24ab1fb7a6369b91de8214fb6 Mon Sep 17 00:00:00 2001 From: Csaba Fabian Date: Thu, 22 Oct 2026 10:37:50 +0000 Subject: [PATCH] A long chat is no longer read in full on every prompt Three readers loaded every message and part of a session on each prompt: the prompt loop's history (MessageV2.stream, on every model step, before filterCompacted kept only what follows the newest compaction), prune after each turn, and the ACP usage update, which fetched the whole session over HTTP to sum its cost. On a chat with 65 MB of history a prompt peaked at 1.0-1.6 GB and took 4-6 s. - filterCompactedEffect pages newest first through MessageV2.scanNewestFirst and stops where filterCompacted stops; both run the same walk (compactedScanner). - prune walks through Session.scanNewestFirst and stops at the newest summary or the first cleared output, as its loop already did. - The usage update reads the newest 20 messages and sends no cost. ACP makes cost optional and DocuMentor reads only used and size. test/pg/bounded-history.test.ts holds the paged read to the full one for sessions without a compaction, with one, with a retained tail across a page boundary, with two, and with an unfinished summary, and counts the pages each reads. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_0148RPMT9kUkZ6BL4TtebZJe --- documentor/FORK.md | 4 + packages/opencode/src/acp/service.ts | 9 +- packages/opencode/src/session/compaction.ts | 52 ++-- packages/opencode/src/session/message-v2.ts | 62 ++++- packages/opencode/src/session/session.ts | 13 + .../opencode/test/pg/bounded-history.test.ts | 241 ++++++++++++++++++ 6 files changed, 344 insertions(+), 37 deletions(-) create mode 100644 packages/opencode/test/pg/bounded-history.test.ts diff --git a/documentor/FORK.md b/documentor/FORK.md index a9b12f9533af..9c10817307e0 100644 --- a/documentor/FORK.md +++ b/documentor/FORK.md @@ -26,12 +26,16 @@ what a rebase meets. | `packages/opencode/src/session/processor.ts` | One lock per processor around each read-then-write of a tool part; `tool-input-delta` and `tool-input-end` do not read a part the processor already tracks | The processor and the running tool both update a tool part. SQLite made each update atomic; a network database lets them interleave, and the tool's metadata was lost. Each read is a round trip, and a large edit streams thousands of deltas | | `packages/opencode/src/acp/service.ts` | `prompt` passes a `messageId` that starts with `msg` to v1 as `messageID`, for a normal prompt and a slash command; an empty prompt passes none. `newSession` passes a `_meta.opencode.sessionId` that starts with `ses` to `session.create` as `id` | DocuMentor names the user message a prompt creates and the session a new chat creates, so their optimistic, live and durable copies share one id. ACP clients that send UUIDs, or no `_meta`, keep opencode-minted ids | | `packages/opencode/src/acp/service.ts` | `prompt` reads `_meta.documentor = { author, answerable, noReply, system, partMetadata }` and forwards each well-formed field to v1, `partMetadata` as the first text part's `metadata`; a prompt with `noReply` or `answerable: false` is sent without the idle wait and answers `end_turn` at once; a request carrying `_meta.documentor` is never read as a slash command; an answer that is a user message is `end_turn` without usage; `newSession` reads `_meta.documentor.metadata` into `session.create` | ACP is DocuMentor's only channel to the session; a stored-only message produces no idle event; a message's mentions live on its text part | +| `packages/opencode/src/acp/service.ts` | The `usage_update` after a prompt reads the newest 20 messages and carries no `cost` | The session total needed every message of the chat, fetched over HTTP on each prompt. `cost` is optional in ACP, and DocuMentor reads only `used` and `size` | | `packages/opencode/src/session/prompt.ts` | `prompt` with no parts and no `noReply`, on a session that already has a user message, stores nothing and runs `loop`; on a session with no user message it stores a zero-part row as upstream does. `createUserMessage` stores the message and its parts with `Session.saveMessage` | DocuMentor's abort-then-answer: after a cancel, an empty ACP prompt answers the user messages that arrived during the cancelled step, which opencode already holds. A v1 client sending `parts: []` gets the same resume; upstream clients send none. A reader never sees a user message without all of its parts | | `packages/opencode/src/session/prompt.ts` | `PromptInput` accepts `author` and `answerable`; `prompt` returns the stored message without a loop when `answerable` is `false`, as for `noReply`; `runLoop` returns the newest assistant message (the newest message only when the session has no assistant message) instead of throwing when no answerable user message exists; a prompt of any kind, a stored-only message included, commits a pending revert; the loop's model input is built with `attributeAuthors` when the session's metadata is `{ documentor: { kind: "discussion" } }` | A discussion stores human messages as user messages; only an `@mentor` message starts a turn, and that turn must see who said what | | `packages/schema/src/v1/session.ts`, `packages/core/src/v1/session.ts` | A user message's info carries `author?: {id, name}` and `answerable?: boolean`; `UserAuthor` is exported | DocuMentor records who typed each message in opencode itself, and marks a discussion message as context that no turn answers | | `packages/opencode/src/session/message-v2.ts` | `latest` skips user messages whose `answerable` is `false`; `toModelMessagesEffect` takes `attributeAuthors` and prefixes the first non-synthetic text part of an authored user message with `Name: `; `authorPrefix` and `isDiscussion` are exported | The message a turn answers is the newest answerable one; attribution is switched by the session kind so ordinary chats send today's prompts | +| `packages/opencode/src/session/message-v2.ts` | `filterCompactedEffect` reads pages newest first through `scanNewestFirst` and stops where `filterCompacted` stops; both run the same walk, `compactedScanner` | The prompt loop calls it on every step. It read the whole session and then kept only what follows the newest compaction: a chat with 65 MB of history peaked above 1 GB per prompt | | `packages/opencode/src/session/compaction.ts` | `serialize` prefixes an authored user message with `Name: ` when the session is a discussion (its `process` passes `MessageV2.isDiscussion(session.metadata)`); `prune` and `turns()` count only user messages whose `answerable` is not `false` | A compacted discussion keeps who said what; side remarks do not shrink the protected or retained window | +| `packages/opencode/src/session/compaction.ts` | `prune` walks the session through `Session.scanNewestFirst` instead of loading it whole | The walk stops at the newest summary or the first cleared output, but every message was loaded after each turn | | `packages/opencode/src/session/session.ts` | `CreateInput` and `create` take an optional `id`; `create` refuses an id that names an existing session. `saveMessage` publishes a message and its parts in one transaction. `fork` publishes its copy with `publishMany` and removes the new session again if the copy does not commit | DocuMentor shows a new chat under its final id before opencode answers `session/new`. Each part was its own commit: on a network database a reader between them saw the context part of a user message without its text. A fork of a long chat was one transaction per message and per part: tens of seconds against the remote database | +| `packages/opencode/src/session/session.ts` | `scanNewestFirst` hands messages to a visitor newest first, a page at a time, until it answers false | What `prune` reads, without loading a long chat in full | | `packages/core/src/event.ts` | Adds `Publication`, `publication` and `publishMany`: durable events of one aggregate projected, written and notified in one transaction. A copy of the non-replay path of `commitDurableEvent`, not a refactor of it | Each `publish` is about nine round trips; `Session.fork` copies a whole chat. Kept as a separate copy so upstream changes to `commitDurableEvent` rebase cleanly | | `packages/opencode/src/event-v2-bridge.ts` | Routes `publishMany` with the same instance location as `publish` | The bridge implements `EventV2.Interface` | | `packages/core/test/session-runner-tool-events.test.ts` | Its `EventV2.Service` double has a `publishMany` | The interface requires it | diff --git a/packages/opencode/src/acp/service.ts b/packages/opencode/src/acp/service.ts index 32214c1f51a6..c72a6c7fcc5d 100644 --- a/packages/opencode/src/acp/service.ts +++ b/packages/opencode/src/acp/service.ts @@ -46,6 +46,9 @@ import { ModelV2 } from "@opencode-ai/core/model" import { Provider } from "@/provider/provider" import type { Command } from "@/command" +/** Messages read for a usage update: enough to hold the newest assistant message. */ +const USAGE_UPDATE_MESSAGES = 20 + export const AuthMethodID = "opencode-login" export type Error = ACPError.Error @@ -655,6 +658,10 @@ function makeUsageService(sdk: OpencodeClient) { }, ) + // documentor: reads the newest page only and reports no cost. The latest + // assistant message is always on it, and the session total would need every + // message of a long chat on each prompt; the ACP field is optional and + // DocuMentor reads only `used` and `size`. const sendUpdate: UsageService.Interface["sendUpdate"] = Effect.fn("ACP.promptUsage.sendUpdate")(function* (params) { const messages = yield* request( () => @@ -662,6 +669,7 @@ function makeUsageService(sdk: OpencodeClient) { { sessionID: params.sessionID, directory: params.directory, + limit: USAGE_UPDATE_MESSAGES, }, { throwOnError: true }, ), @@ -692,7 +700,6 @@ function makeUsageService(sdk: OpencodeClient) { sessionUpdate: "usage_update", used: UsageService.contextTokens(message), size, - cost: { amount: UsageService.totalSessionCost(messages), currency: "USD" }, }, }) .catch(() => {}), diff --git a/packages/opencode/src/session/compaction.ts b/packages/opencode/src/session/compaction.ts index aaaa0e6dedd1..8a71a6ba0a28 100644 --- a/packages/opencode/src/session/compaction.ts +++ b/packages/opencode/src/session/compaction.ts @@ -277,34 +277,40 @@ const layer = Layer.effect( if (!cfg.compaction?.prune) return yield* Effect.logInfo("pruning") - const msgs = yield* session - .messages({ sessionID: input.sessionID }) - .pipe(Effect.catchIf(NotFoundError.isInstance, () => Effect.succeed(undefined))) - if (!msgs) return - let total = 0 let pruned = 0 const toPrune: SessionV1.ToolPart[] = [] let turns = 0 - loop: for (let msgIndex = msgs.length - 1; msgIndex >= 0; msgIndex--) { - const msg = msgs[msgIndex] - if (msg.info.role === "user" && msg.info.answerable !== false) turns++ - if (turns < 2) continue - if (msg.info.role === "assistant" && msg.info.summary) break loop - for (let partIndex = msg.parts.length - 1; partIndex >= 0; partIndex--) { - const part = msg.parts[partIndex] - if (part.type !== "tool") continue - if (part.state.status !== "completed") continue - if (PRUNE_PROTECTED_TOOLS.includes(part.tool)) continue - if (part.state.time.compacted) break loop - const estimate = Token.estimate(part.state.output) - total += estimate - if (total <= PRUNE_PROTECT) continue - pruned += estimate - toPrune.push(part) - } - } + // documentor: reads newest first and stops where the walk stops, instead of loading the + // whole session after every turn. + const found = yield* session + .scanNewestFirst({ + sessionID: input.sessionID, + visit: (msg) => { + if (msg.info.role === "user" && msg.info.answerable !== false) turns++ + if (turns < 2) return true + if (msg.info.role === "assistant" && msg.info.summary) return false + for (let partIndex = msg.parts.length - 1; partIndex >= 0; partIndex--) { + const part = msg.parts[partIndex] + if (part.type !== "tool") continue + if (part.state.status !== "completed") continue + if (PRUNE_PROTECTED_TOOLS.includes(part.tool)) continue + if (part.state.time.compacted) return false + const estimate = Token.estimate(part.state.output) + total += estimate + if (total <= PRUNE_PROTECT) continue + pruned += estimate + toPrune.push(part) + } + return true + }, + }) + .pipe( + Effect.as(true), + Effect.catchIf(NotFoundError.isInstance, () => Effect.succeed(false)), + ) + if (!found) return yield* Effect.logInfo("found", { pruned, total }) if (pruned > PRUNE_MINIMUM) { diff --git a/packages/opencode/src/session/message-v2.ts b/packages/opencode/src/session/message-v2.ts index d35123bb078e..b4cc829c8033 100644 --- a/packages/opencode/src/session/message-v2.ts +++ b/packages/opencode/src/session/message-v2.ts @@ -550,29 +550,39 @@ export const get = Effect.fn("MessageV2.get")(function* (input: { sessionID: Ses } }) -export function filterCompacted(msgs: Iterable) { +/** + * Reads messages newest first and keeps what the model still sees: everything + * back to the newest completed compaction, plus the tail that compaction + * retains. `push` answers false once no older message can matter. + */ +function compactedScanner() { const result = [] as WithParts[] const completed = new Set() let retain: MessageID | undefined - for (const msg of msgs) { + const push = (msg: WithParts): boolean => { result.push(msg) - if (retain) { - if (msg.info.id === retain) break - continue - } + if (retain) return msg.info.id !== retain if (msg.info.role === "user" && completed.has(msg.info.id)) { const part = msg.parts.find((item): item is CompactionPart => item.type === "compaction") - if (!part) continue - if (!part.tail_start_id) break + if (!part) return true + if (!part.tail_start_id) return false retain = part.tail_start_id - if (msg.info.id === retain) break - continue + return msg.info.id !== retain } - if (msg.info.role === "user" && completed.has(msg.info.id) && msg.parts.some((part) => part.type === "compaction")) - break if (msg.info.role === "assistant" && msg.info.summary && msg.info.finish && !msg.info.error) completed.add(msg.info.parentID) + return true } + return { push, finish: () => orderCompacted(result) } +} + +export function filterCompacted(msgs: Iterable) { + const scanner = compactedScanner() + for (const msg of msgs) if (!scanner.push(msg)) break + return scanner.finish() +} + +function orderCompacted(result: WithParts[]) { result.reverse() const compactionIndex = result.findLastIndex( (msg) => @@ -603,8 +613,34 @@ export function filterCompacted(msgs: Iterable) { return result } +/** + * documentor: hands a session's messages to `visit` newest first, a page at a time, until it + * answers false or no older message is left. Fails with NotFoundError when the session does not exist. + */ +export const scanNewestFirst = Effect.fnUntraced(function* (input: { + sessionID: SessionID + visit: (message: WithParts) => boolean +}) { + let before: string | undefined + while (true) { + const next = yield* page({ sessionID: input.sessionID, limit: 50, before }) + for (let i = next.items.length - 1; i >= 0; i--) { + const item = next.items[i] + if (item && !input.visit(item)) return + } + if (!next.more || !next.cursor) return + before = next.cursor + } +}) + +// documentor: reads back to the newest compaction and no further, so a long chat is not +// loaded in full on every step of the prompt loop. export const filterCompactedEffect = Effect.fnUntraced(function* (sessionID: SessionID) { - return filterCompacted(yield* stream(sessionID)) + const scanner = compactedScanner() + yield* scanNewestFirst({ sessionID, visit: scanner.push }).pipe( + Effect.catchIf(NotFoundError.isInstance, () => Effect.void), + ) + return scanner.finish() }) // filterCompacted reorders messages for model consumption diff --git a/packages/opencode/src/session/session.ts b/packages/opencode/src/session/session.ts index 7e1abbfc35cc..30bd3d93827d 100644 --- a/packages/opencode/src/session/session.ts +++ b/packages/opencode/src/session/session.ts @@ -449,6 +449,14 @@ export interface Interface { readonly setWorkspace: (input: { sessionID: SessionID; workspaceID: Info["workspaceID"] }) => Effect.Effect readonly diff: (sessionID: SessionID) => Effect.Effect readonly messages: (input: { sessionID: SessionID; limit?: number }) => Effect.Effect + /** + * documentor: hands messages to `visit` newest first, a page at a time, until it answers false + * or the session has no older message, so a caller that needs only recent history stops there. + */ + readonly scanNewestFirst: (input: { + sessionID: SessionID + visit: (message: SessionV1.WithParts) => boolean + }) => Effect.Effect readonly children: (parentID: SessionID) => Effect.Effect readonly remove: (sessionID: SessionID) => Effect.Effect readonly updateMessage: (msg: T) => Effect.Effect @@ -899,6 +907,10 @@ const layer: Layer.Layer< return result.reverse() }) + const scanNewestFirst: Interface["scanNewestFirst"] = Effect.fn("Session.scanNewestFirst")((input) => + MessageV2.scanNewestFirst(input).pipe(Effect.provideService(Database.Service, database)), + ) + const removeMessage = Effect.fn("Session.removeMessage")(function* (input: { sessionID: SessionID messageID: MessageID @@ -971,6 +983,7 @@ const layer: Layer.Layer< setWorkspace, diff, messages, + scanNewestFirst, children, remove, updateMessage, diff --git a/packages/opencode/test/pg/bounded-history.test.ts b/packages/opencode/test/pg/bounded-history.test.ts new file mode 100644 index 000000000000..4a46aa3417b7 --- /dev/null +++ b/packages/opencode/test/pg/bounded-history.test.ts @@ -0,0 +1,241 @@ +import { describe, expect } from "bun:test" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import type { SessionV1 } from "@opencode-ai/core/v1/session" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { ModelV2 } from "@opencode-ai/core/model" +import { Effect, Tracer } from "effect" +import { Session as SessionNs } from "@/session/session" +import { MessageV2 } from "../../src/session/message-v2" +import { MessageID, PartID, type SessionID } from "../../src/session/schema" +import { testEffect } from "../lib/effect" + +// The prompt loop reads history back to the newest compaction, a page at a time. These cases +// hold it to the full read it replaces (`filterCompacted` over `stream`) and count the pages. + +const it = testEffect(LayerNode.compile(LayerNode.group([SessionNs.node, MessageV2.node, SessionProjector.node]))) + +const withSession = (fn: (sessionID: SessionID) => Effect.Effect) => + Effect.acquireUseRelease( + Effect.gen(function* () { + const session = yield* SessionNs.Service + return (yield* session.create({})).id + }), + fn, + (sessionID) => SessionNs.Service.use((session) => session.remove(sessionID)).pipe(Effect.ignore), + ) + +let clock = Date.now() + +const addUser = Effect.fn("Test.addUser")(function* (sessionID: SessionID) { + const session = yield* SessionNs.Service + const id = MessageID.ascending() + yield* session.updateMessage({ + id, + sessionID, + role: "user", + time: { created: clock++ }, + agent: "test", + model: { providerID: "test", modelID: "test" }, + } as unknown as SessionV1.Info) + yield* session.updatePart({ id: PartID.ascending(), sessionID, messageID: id, type: "text", text: id }) + return id +}) + +const addUsers = Effect.fn("Test.addUsers")(function* (sessionID: SessionID, count: number) { + const ids = [] as MessageID[] + for (let i = 0; i < count; i++) ids.push(yield* addUser(sessionID)) + return ids +}) + +const addSummary = Effect.fn("Test.addSummary")(function* ( + sessionID: SessionID, + parentID: MessageID, + finished: boolean, +) { + const session = yield* SessionNs.Service + yield* session.updateMessage({ + id: MessageID.ascending(), + sessionID, + role: "assistant", + time: { created: clock++ }, + parentID, + modelID: ModelV2.ID.make("test"), + providerID: ProviderV2.ID.make("test"), + mode: "", + agent: "default", + path: { cwd: "/", root: "/" }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + summary: true, + finish: finished ? "end_turn" : undefined, + } as unknown as SessionV1.Info) +}) + +/** A user message carrying a compaction part, answered by its summary. */ +const addCompaction = Effect.fn("Test.addCompaction")(function* ( + sessionID: SessionID, + options: { tail?: MessageID; finished?: boolean } = {}, +) { + const session = yield* SessionNs.Service + const id = yield* addUser(sessionID) + yield* session.updatePart({ + id: PartID.ascending(), + sessionID, + messageID: id, + type: "compaction", + auto: true, + tail_start_id: options.tail, + } as any) + yield* addSummary(sessionID, id, options.finished ?? true) + return id +}) + +/** Runs `effect` and counts the history pages it reads. */ +const countingPages = (effect: Effect.Effect) => + Effect.gen(function* () { + const native = yield* Effect.tracer + let pages = 0 + const tracer = Tracer.make({ + span(options) { + if (options.name === "MessageV2.page") pages++ + return native.span(options) + }, + }) + const result = yield* effect.pipe(Effect.withTracer(tracer)) + return { result, pages } + }) + +// Each case writes up to a hundred messages through the event log; a loaded runner needs longer than the default. +const TIMEOUT = 60_000 + +const ids = (messages: readonly SessionV1.WithParts[]) => messages.map((message) => message.info.id) + +/** The bounded read matches the full one, and reads `pages` pages where the full read needs `fullPages`. */ +const expectBounded = (sessionID: SessionID) => + Effect.gen(function* () { + const bounded = yield* countingPages(MessageV2.filterCompactedEffect(sessionID)) + const full = yield* countingPages( + MessageV2.stream(sessionID).pipe(Effect.map((all) => MessageV2.filterCompacted(all))), + ) + expect(ids(bounded.result)).toEqual(ids(full.result)) + return { pages: bounded.pages, fullPages: full.pages, kept: bounded.result.length } + }) + +describe("documentor: MessageV2.filterCompactedEffect reads back to the newest compaction", () => { + it.instance( + "reads every page of a session without a compaction", + () => + withSession((sessionID) => + Effect.gen(function* () { + yield* addUsers(sessionID, 60) + const { pages, fullPages, kept } = yield* expectBounded(sessionID) + expect(kept).toBe(60) + expect(pages).toBe(fullPages) + }), + ), + TIMEOUT, + ) + + it.instance( + "stops at a compaction that retains no tail", + () => + withSession((sessionID) => + Effect.gen(function* () { + yield* addUsers(sessionID, 60) + yield* addCompaction(sessionID) + yield* addUsers(sessionID, 3) + const { pages, fullPages, kept } = yield* expectBounded(sessionID) + expect(kept).toBe(5) + expect(pages).toBe(1) + expect(fullPages).toBe(2) + }), + ), + TIMEOUT, + ) + + it.instance( + "keeps a retained tail that spans a page boundary", + () => + withSession((sessionID) => + Effect.gen(function* () { + yield* addUsers(sessionID, 40) + const tail = yield* addUsers(sessionID, 60) + yield* addCompaction(sessionID, { tail: tail[0] }) + yield* addUsers(sessionID, 3) + const { pages, fullPages, kept } = yield* expectBounded(sessionID) + expect(kept).toBe(65) + expect(pages).toBe(2) + expect(fullPages).toBe(3) + }), + ), + TIMEOUT, + ) + + it.instance( + "stops at the newer of two compactions", + () => + withSession((sessionID) => + Effect.gen(function* () { + yield* addUsers(sessionID, 40) + yield* addCompaction(sessionID) + yield* addUsers(sessionID, 40) + yield* addCompaction(sessionID) + yield* addUsers(sessionID, 2) + const { pages, fullPages, kept } = yield* expectBounded(sessionID) + expect(kept).toBe(4) + expect(pages).toBe(1) + expect(fullPages).toBe(2) + }), + ), + TIMEOUT, + ) + + it.instance( + "reads past a compaction whose summary never finished", + () => + withSession((sessionID) => + Effect.gen(function* () { + yield* addUsers(sessionID, 60) + yield* addCompaction(sessionID, { finished: false }) + yield* addUsers(sessionID, 2) + const { pages, fullPages, kept } = yield* expectBounded(sessionID) + expect(kept).toBe(64) + expect(pages).toBe(fullPages) + }), + ), + TIMEOUT, + ) + + it.instance("returns nothing for a session that does not exist", () => + Effect.gen(function* () { + expect(yield* MessageV2.filterCompactedEffect("ses_missing" as SessionID)).toEqual([]) + }), + ) +}) + +describe("documentor: Session.scanNewestFirst", () => { + it.instance( + "stops reading pages where the visitor stops", + () => + withSession((sessionID) => + Effect.gen(function* () { + const written = yield* addUsers(sessionID, 70) + const session = yield* SessionNs.Service + const seen = [] as MessageID[] + const { pages } = yield* countingPages( + session.scanNewestFirst({ + sessionID, + visit: (message) => { + seen.push(message.info.id) + return seen.length < 10 + }, + }), + ) + expect(seen).toEqual(written.slice(-10).reverse()) + expect(pages).toBe(1) + }), + ), + TIMEOUT, + ) +})