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, + ) +})