From 6667ec08e88b407ff88dd868549da15495515695 Mon Sep 17 00:00:00 2001 From: Fabian Csy Date: Mon, 19 Oct 2026 14:01:12 +0000 Subject: [PATCH] Session.fork publishes its copy in one transaction Against a remote Postgres each durable publish is its own transaction of eight or nine round trips, so forking a 234-message chat (936 rows) took about 30 s at a 2 ms round trip. EventV2.publishMany publishes durable events of one aggregate in a single transaction: one lock, one sequence read, the projectors per event as before, then one sequence write and chunked event inserts. Listeners are notified per event after the commit. Same rows, same events, measured interleaved against the old path at a 2 ms round trip: 47-77 s before, 8.3-8.7 s after; 6-7 s with 16 KB parts. The copy commits all or nothing, and fork removes the new session if it does not. publishMany is a copy of commitDurableEvent's non-replay path rather than a refactor of it, so upstream changes there rebase cleanly. Co-Authored-By: Claude Opus 5.5 (1M context) --- documentor/FORK.md | 59 ++--- packages/core/src/event.ts | 122 +++++++++++ packages/core/test/event-publish-many.test.ts | 204 ++++++++++++++++++ .../test/session-runner-tool-events.test.ts | 1 + packages/opencode/src/event-v2-bridge.ts | 18 +- packages/opencode/src/session/session.ts | 70 +++--- .../opencode/test/session/fork-batch.test.ts | 150 +++++++++++++ 7 files changed, 570 insertions(+), 54 deletions(-) create mode 100644 packages/core/test/event-publish-many.test.ts create mode 100644 packages/opencode/test/session/fork-batch.test.ts diff --git a/documentor/FORK.md b/documentor/FORK.md index da50d81c43b7..0d8461f0a024 100644 --- a/documentor/FORK.md +++ b/documentor/FORK.md @@ -13,34 +13,37 @@ what a rebase meets. ## Upstream files changed -| File | What differs | Why | -| ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | -| The 11 table files: `packages/core/src/{account,credential,event,permission,project,session,share}/sql.ts`, `control-plane/workspace.sql.ts`, `data-migration.sql.ts`, `database/schema.sql.ts`, `database/path.ts` | Postgres table builder; `count`, `jsonb` and `text` columns from `effect-drizzle-pg`; time columns are `double precision` | Written by `documentor/scripts/convert-schema.py`, never by hand | -| `packages/core/src/database/schema.gen.ts`, `packages/core/schema.json`, `packages/core/drizzle.config.ts` | The Postgres baseline, as one statement batch | Generated; see "Regenerating the baseline" | -| `packages/core/src/database/database.ts` | Connects with `@effect/sql-pg`; verifies collation and migrations and changes nothing; a throwaway schema per layer when `OPENCODE_DATABASE_EPHEMERAL=1` | The service itself | -| `packages/core/src/database/migration.ts` | `migrate`, `migrateAll`, `verify`, `pending`, `BASELINE_IDS`, `ported` | Schema work is a command, not a start-up side effect | -| `packages/core/src/project/directories.ts` | `Transaction` is `Database.Transaction` | It was typed against the SQLite client | -| `packages/core/src/flag/flag.ts` | No `OPENCODE_DB` | Nothing reads it | -| `packages/core/package.json` | Depends on `@effect/sql-pg`, `pg`, `@opencode-ai/effect-drizzle-pg`; tests run with `--timeout 30000` | Each test database is a real schema, about 110 ms to build | -| `packages/opencode/src/cli/cmd/db.ts` | `db `, `db migrate [--all]`, `db provision`, `db grant-read`, all outside the application runtime | The runtime's database service refuses the schemas these commands exist to create and migrate | -| `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/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/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/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 | 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 | -| `packages/core/src/v1/config/config.ts`, `packages/opencode/src/session/llm.ts`, `packages/opencode/src/agent/agent.ts` | `experimental.openTelemetryChatContent` is a config key; both `experimental_telemetry` blocks pass it as the AI SDK's `recordInputs` and `recordOutputs`, so unless it is `true` the `ai.*` spans carry no prompt and no response text | The prompts DocuMentor sends hold personal data about third parties. Upstream's bare `openTelemetry` boolean records them on every span, and DocuMentor's collector had to strip them downstream | -| `packages/sdk/js/src/v2/gen/*` | `session.create` takes `id`; `session.prompt` and `session.promptAsync` take `author` and `answerable`, `UserMessage` declares both, and `Config.experimental` declares `openTelemetryChatContent` | Generated by `packages/sdk/js/script/build.ts` from the server's OpenAPI | -| `packages/opencode/src/acp/event.ts` | `agent_message_chunk`, `agent_thought_chunk`, `tool_call` and `tool_call_update`, live and replayed, carry `_meta.opencode = { messageId, parentId }`: the assistant message and the user message it answers. The subscription caches each assistant message's `parentID` from `message.updated` for the sessions it tracks. On a miss it fetches the message once, shared by concurrent updates; a failed fetch is remembered until the session goes idle, and meanwhile the updates carry `parentId: null`. A session's cache is forgotten when it goes idle, and the whole cache when the event stream disconnects | DocuMentor detects from the ACP stream alone, in order, the moment the agent starts answering a message sent while the turn ran | -| `packages/opencode/src/acp/event.ts` | A replayed `user_message_chunk` carries `_meta.opencode = { messageId, parentId: null, author?, answerable?, partMetadata? }` | DocuMentor rebuilds a discussion from the replay with authors and mention metadata | -| Fork cases in `packages/opencode/test/acp/event.test.ts`, `test/acp/service-session.test.ts`, `test/session/prompt.test.ts`, `test/session/message-v2.test.ts`, `test/session/compaction.test.ts`, `test/session/session.test.ts` and `test/session/llm.test.ts` | Tests named `documentor: …` | They pin the ACP, session and telemetry rows above; a rebase keeps them | -| `packages/core/src/integration.ts` | An OAuth attempt's credential is stored before the attempt reads as complete, inside one update of the attempts map | The attempt was marked complete first and the credential written after. On SQLite nothing could look in between; on a network database a caller that saw "complete" could list no credential | -| `patches/effect@4.0.0-beta.83.patch` | Adds Effect's own fix for re-entrant interruption, first released in `4.0.0-beta.100` | A fiber that woke another fiber which interrupted it synchronously had its continuation stack corrupted once its cleanup awaited anything. With Postgres that leaked a pooled connection. Drop these hunks when upstream moves to `beta.100` or later | -| `turbo.json` | `OPENCODE_TEST_DATABASE_URL` passes through to every task; `@opencode-ai/effect-drizzle-pg#test` is a task | turbo strips variables it is not told about | -| Test helpers: `packages/core/test/preload.ts`, `packages/opencode/test/preload.ts`, `packages/opencode/test/fixture/db.ts`, `packages/opencode/test/server/httpapi-exercise/environment.ts`, `packages/sdk-next/test/embedded.test.ts` | Ask for a throwaway schema; no database file to delete | | +| File | What differs | Why | +| ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| The 11 table files: `packages/core/src/{account,credential,event,permission,project,session,share}/sql.ts`, `control-plane/workspace.sql.ts`, `data-migration.sql.ts`, `database/schema.sql.ts`, `database/path.ts` | Postgres table builder; `count`, `jsonb` and `text` columns from `effect-drizzle-pg`; time columns are `double precision` | Written by `documentor/scripts/convert-schema.py`, never by hand | +| `packages/core/src/database/schema.gen.ts`, `packages/core/schema.json`, `packages/core/drizzle.config.ts` | The Postgres baseline, as one statement batch | Generated; see "Regenerating the baseline" | +| `packages/core/src/database/database.ts` | Connects with `@effect/sql-pg`; verifies collation and migrations and changes nothing; a throwaway schema per layer when `OPENCODE_DATABASE_EPHEMERAL=1` | The service itself | +| `packages/core/src/database/migration.ts` | `migrate`, `migrateAll`, `verify`, `pending`, `BASELINE_IDS`, `ported` | Schema work is a command, not a start-up side effect | +| `packages/core/src/project/directories.ts` | `Transaction` is `Database.Transaction` | It was typed against the SQLite client | +| `packages/core/src/flag/flag.ts` | No `OPENCODE_DB` | Nothing reads it | +| `packages/core/package.json` | Depends on `@effect/sql-pg`, `pg`, `@opencode-ai/effect-drizzle-pg`; tests run with `--timeout 30000` | Each test database is a real schema, about 110 ms to build | +| `packages/opencode/src/cli/cmd/db.ts` | `db `, `db migrate [--all]`, `db provision`, `db grant-read`, all outside the application runtime | The runtime's database service refuses the schemas these commands exist to create and migrate | +| `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/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/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/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/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 | +| `packages/core/src/v1/config/config.ts`, `packages/opencode/src/session/llm.ts`, `packages/opencode/src/agent/agent.ts` | `experimental.openTelemetryChatContent` is a config key; both `experimental_telemetry` blocks pass it as the AI SDK's `recordInputs` and `recordOutputs`, so unless it is `true` the `ai.*` spans carry no prompt and no response text | The prompts DocuMentor sends hold personal data about third parties. Upstream's bare `openTelemetry` boolean records them on every span, and DocuMentor's collector had to strip them downstream | +| `packages/sdk/js/src/v2/gen/*` | `session.create` takes `id`; `session.prompt` and `session.promptAsync` take `author` and `answerable`, `UserMessage` declares both, and `Config.experimental` declares `openTelemetryChatContent` | Generated by `packages/sdk/js/script/build.ts` from the server's OpenAPI | +| `packages/opencode/src/acp/event.ts` | `agent_message_chunk`, `agent_thought_chunk`, `tool_call` and `tool_call_update`, live and replayed, carry `_meta.opencode = { messageId, parentId }`: the assistant message and the user message it answers. The subscription caches each assistant message's `parentID` from `message.updated` for the sessions it tracks. On a miss it fetches the message once, shared by concurrent updates; a failed fetch is remembered until the session goes idle, and meanwhile the updates carry `parentId: null`. A session's cache is forgotten when it goes idle, and the whole cache when the event stream disconnects | DocuMentor detects from the ACP stream alone, in order, the moment the agent starts answering a message sent while the turn ran | +| `packages/opencode/src/acp/event.ts` | A replayed `user_message_chunk` carries `_meta.opencode = { messageId, parentId: null, author?, answerable?, partMetadata? }` | DocuMentor rebuilds a discussion from the replay with authors and mention metadata | +| Fork cases in `packages/opencode/test/acp/event.test.ts`, `test/acp/service-session.test.ts`, `test/session/prompt.test.ts`, `test/session/message-v2.test.ts`, `test/session/compaction.test.ts`, `test/session/session.test.ts` `test/session/llm.test.ts` and `test/session/fork-batch.test.ts`, and `packages/core/test/event-publish-many.test.ts` | Tests named `documentor: …` | They pin the ACP, session and telemetry rows above; a rebase keeps them | +| `packages/core/src/integration.ts` | An OAuth attempt's credential is stored before the attempt reads as complete, inside one update of the attempts map | The attempt was marked complete first and the credential written after. On SQLite nothing could look in between; on a network database a caller that saw "complete" could list no credential | +| `patches/effect@4.0.0-beta.83.patch` | Adds Effect's own fix for re-entrant interruption, first released in `4.0.0-beta.100` | A fiber that woke another fiber which interrupted it synchronously had its continuation stack corrupted once its cleanup awaited anything. With Postgres that leaked a pooled connection. Drop these hunks when upstream moves to `beta.100` or later | +| `turbo.json` | `OPENCODE_TEST_DATABASE_URL` passes through to every task; `@opencode-ai/effect-drizzle-pg#test` is a task | turbo strips variables it is not told about | +| Test helpers: `packages/core/test/preload.ts`, `packages/opencode/test/preload.ts`, `packages/opencode/test/fixture/db.ts`, `packages/opencode/test/server/httpapi-exercise/environment.ts`, `packages/sdk-next/test/embedded.test.ts` | Ask for a throwaway schema; no database file to delete | | Upstream tests that are changed or deleted are listed, with reasons, in [`upstream-test-exceptions.md`](upstream-test-exceptions.md). diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index c92ac0ac2ce3..b22ab3309a5c 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -123,12 +123,35 @@ export interface PublishOptions { readonly commit?: (seq: number) => Effect.Effect } +/** documentor: one durable event of a {@link Interface.publishMany} batch. */ +export interface Publication { + readonly definition: D + readonly data: Data +} + +/** documentor: pairs a definition with its data, typed, for {@link Interface.publishMany}. */ +export const publication = (definition: D, data: Data): Publication => + ({ definition, data }) as Publication + +/** documentor: events per INSERT in publishMany. Parts carry tool output, so a long chat's rows can be large. */ +const PUBLISH_MANY_CHUNK = 200 + export interface Interface { readonly publish: ( definition: D, data: Data, options?: PublishOptions, ) => Effect.Effect> + /** + * documentor: publishes durable events of one aggregate in a single transaction, in order. + * Each event is projected and notified exactly as `publish` would; only the round trips + * of the per-event transaction (lock, sequence read, sequence and event writes) are shared. + * All or nothing: a failing projector commits none of them. + */ + readonly publishMany: ( + publications: readonly Publication[], + options?: { readonly location?: Location.Ref }, + ) => Effect.Effect readonly subscribe: (definition: D) => Stream.Stream> readonly all: () => Stream.Stream readonly durable: (input: { readonly aggregateID: string; readonly after?: number }) => Stream.Stream @@ -438,6 +461,104 @@ export const layerWith = (options?: LayerOptions) => }) } + // documentor: commitDurableEvent's non-replay path for a batch of one aggregate. A separate + // copy rather than a shared helper, so upstream changes to that function rebase cleanly. + function publishMany(publications: readonly Publication[], options?: { readonly location?: Location.Ref }) { + return Effect.gen(function* () { + if (publications.length === 0) return [] as Payload[] + const serviceLocation = Option.getOrUndefined(yield* Effect.serviceOption(Location.Service)) + const location = + options?.location ?? + (serviceLocation + ? { directory: serviceLocation.directory, workspaceID: serviceLocation.workspaceID } + : undefined) + let aggregateID: string | undefined + const pending = new Array<{ definition: Definition; event: Payload }>() + for (const { definition, data } of publications) { + const durable = definition.durable + const aggregate = durable ? (data as Record)[durable.aggregate] : undefined + if (!durable || typeof aggregate !== "string") + return yield* Effect.die( + new InvalidDurableEventError({ + type: definition.type, + message: "publishMany takes durable events with a string aggregate field only", + }), + ) + if (aggregateID !== undefined && aggregate !== aggregateID) + return yield* Effect.die( + new InvalidDurableEventError({ + type: definition.type, + message: `Aggregate mismatch: expected ${aggregateID}, got ${aggregate}`, + }), + ) + aggregateID = aggregate + pending.push({ + definition, + event: { id: ID.create(), type: definition.type, ...(location ? { location } : {}), data } as Payload, + }) + } + const aggregate = aggregateID! + const committed = yield* Effect.uninterruptible( + Effect.gen(function* () { + const events = yield* db + .transaction( + () => + Effect.gen(function* () { + const latest = yield* latestSequence(db, aggregate) + const rows = new Array() + const events = new Array() + for (const [index, { definition, event }] of pending.entries()) { + const durable = definition.durable! + const seq = latest + 1 + index + const encoded = Schema.encodeUnknownSync(definition.data)(event.data) as Record + const committed = { + ...event, + durable: { aggregateID: aggregate, seq, version: durable.version }, + } as Payload + for (const projector of projectors.get(event.type) ?? []) { + yield* projector(committed) + } + rows.push({ + id: event.id, + aggregate_id: aggregate, + seq, + type: versionedType(definition.type, durable.version), + data: encoded, + }) + events.push(committed) + } + const last = latest + pending.length + yield* db + .insert(EventSequenceTable) + .values([{ aggregate_id: aggregate, seq: last }]) + .onConflictDoUpdate({ target: EventSequenceTable.aggregate_id, set: { seq: last } }) + .run() + .pipe(Effect.orDie) + for (let start = 0; start < rows.length; start += PUBLISH_MANY_CHUNK) { + yield* db + .insert(EventTable) + .values(rows.slice(start, start + PUBLISH_MANY_CHUNK)) + .run() + .pipe(Effect.orDie) + } + return events + }), + { behavior: "immediate" }, + ) + .pipe(Effect.orDie) + yield* Effect.forEach(pubsub.durable.get(aggregate) ?? [], (wake) => PubSub.publish(wake, undefined), { + discard: true, + }) + return events + }), + ) + for (const event of committed) { + yield* notify(event, true) + } + return committed + }) + } + function replay( event: SerializedEvent, options?: { readonly publish?: boolean; readonly ownerID?: string; readonly strictOwner?: boolean }, @@ -621,6 +742,7 @@ export const layerWith = (options?: LayerOptions) => return Service.of({ publish, + publishMany, subscribe, all: streamAll, durable, diff --git a/packages/core/test/event-publish-many.test.ts b/packages/core/test/event-publish-many.test.ts new file mode 100644 index 000000000000..b0d361385712 --- /dev/null +++ b/packages/core/test/event-publish-many.test.ts @@ -0,0 +1,204 @@ +import { describe, expect } from "bun:test" +import { Effect, Exit, Fiber, Schema, Stream } from "effect" +import { EventV2 } from "@opencode-ai/core/event" +import { Database } from "@opencode-ai/core/database/database" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql" +import { asc, eq } from "drizzle-orm" +import { SessionV1 } from "@opencode-ai/schema/session-v1" +import { Session } from "@opencode-ai/schema/session" +import { testEffect } from "./lib/effect" + +const Note = EventV2.define({ + type: "test.batch.note", + durable: { version: 1, aggregate: "id" }, + schema: { id: Schema.String, text: Schema.String }, +}) + +const Tag = EventV2.define({ + type: "test.batch.tag", + durable: { version: 2, aggregate: "id" }, + schema: { id: Schema.String, label: Schema.String }, +}) + +const Live = EventV2.define({ + type: "test.batch.live", + schema: { id: Schema.String }, +}) + +const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node]))) + +const rowsOf = (aggregateID: string) => + Effect.gen(function* () { + const { db } = yield* Database.Service + return yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, aggregateID)) + .orderBy(asc(EventTable.seq)) + .all() + .pipe(Effect.orDie) + }) + +const batch = (aggregateID: string, count: number) => + Array.from({ length: count }, (_, index) => + index % 3 === 2 + ? EventV2.publication(Tag, { id: aggregateID, label: `tag ${index}` }) + : EventV2.publication(Note, { id: aggregateID, text: `note ${index}` }), + ) + +describe("documentor: EventV2.publishMany", () => { + it.effect("stores the same rows as publishing the events one by one", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const batched = EventV2.ID.create() + const sequential = EventV2.ID.create() + + yield* events.publishMany(batch(batched, 7)) + for (const { definition, data } of batch(sequential, 7)) { + yield* events.publish(definition, data as never) + } + + const shape = (rows: (typeof EventTable.$inferSelect)[], aggregateID: string) => + rows.map((row) => ({ seq: row.seq, type: row.type, data: { ...row.data, id: row.data.id === aggregateID } })) + expect(shape(yield* rowsOf(batched), batched)).toEqual(shape(yield* rowsOf(sequential), sequential)) + expect((yield* rowsOf(batched)).map((row) => row.seq)).toEqual([0, 1, 2, 3, 4, 5, 6]) + }), + ) + + it.effect("continues the aggregate's sequence and is followed by single publishes", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const { db } = yield* Database.Service + const aggregateID = EventV2.ID.create() + + yield* events.publish(Note, { id: aggregateID, text: "before" }) + const published = yield* events.publishMany(batch(aggregateID, 3)) + yield* events.publish(Note, { id: aggregateID, text: "after" }) + + expect(published.map((event) => event.durable?.seq)).toEqual([1, 2, 3]) + expect((yield* rowsOf(aggregateID)).map((row) => row.seq)).toEqual([0, 1, 2, 3, 4]) + const sequence = yield* db + .select() + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, aggregateID)) + .get() + .pipe(Effect.orDie) + expect(sequence?.seq).toBe(4) + }), + ) + + it.effect("projects every event in order, then notifies each one after the commit", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = EventV2.ID.create() + const order = new Array() + yield* events.project(Note, (event) => Effect.sync(() => order.push(`project ${event.data.text}`))) + yield* events.listen((event) => + Effect.sync(() => { + if ((event.data as { id?: string }).id === aggregateID) order.push(`notify ${event.durable?.seq}`) + }), + ) + const typed = yield* events.subscribe(Note).pipe(Stream.take(2), Stream.runCollect, Effect.forkScoped) + yield* Effect.yieldNow + + yield* events.publishMany([ + EventV2.publication(Note, { id: aggregateID, text: "a" }), + EventV2.publication(Note, { id: aggregateID, text: "b" }), + ]) + + expect(order).toEqual(["project a", "project b", "notify 0", "notify 1"]) + expect(Array.from(yield* Fiber.join(typed)).map((event) => event.data.text)).toEqual(["a", "b"]) + }), + ) + + it.effect("commits nothing and notifies no one when a projector fails mid-batch", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const { db } = yield* Database.Service + const aggregateID = EventV2.ID.create() + const notified = new Array() + yield* db.run("CREATE TABLE IF NOT EXISTS publish_many_probe (value text NOT NULL)") + yield* db.run("DELETE FROM publish_many_probe") + yield* events.project(Note, (event) => + event.data.text === "boom" + ? Effect.die("projector failed") + : db + .run(`INSERT INTO publish_many_probe (value) VALUES ('${event.data.text}')`) + .pipe(Effect.orDie, Effect.asVoid), + ) + yield* events.listen((event) => Effect.sync(() => notified.push(event.id))) + + const exit = yield* events + .publishMany([ + EventV2.publication(Note, { id: aggregateID, text: "kept?" }), + EventV2.publication(Note, { id: aggregateID, text: "boom" }), + ]) + .pipe(Effect.exit) + + expect(Exit.isFailure(exit)).toBeTrue() + expect(yield* db.all("SELECT value FROM publish_many_probe")).toEqual([]) + expect(yield* rowsOf(aggregateID)).toEqual([]) + expect( + yield* db.select().from(EventSequenceTable).where(eq(EventSequenceTable.aggregate_id, aggregateID)).all(), + ).toEqual([]) + expect(notified).toEqual([]) + }), + ) + + it.effect("refuses a batch that spans aggregates or holds a live-only event", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const first = EventV2.ID.create() + const second = EventV2.ID.create() + + const mixed = yield* events + .publishMany([ + EventV2.publication(Note, { id: first, text: "a" }), + EventV2.publication(Note, { id: second, text: "b" }), + ]) + .pipe(Effect.exit) + const live = yield* events.publishMany([EventV2.publication(Live, { id: first })]).pipe(Effect.exit) + + expect(String(mixed)).toContain("Aggregate mismatch") + expect(String(live)).toContain("durable events") + expect(yield* rowsOf(first)).toEqual([]) + expect(yield* events.publishMany([])).toEqual([]) + }), + ) + + it.effect("wakes a durable aggregate stream that is tailing", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + // The stream decodes through the durable manifest, which knows real event types only. + const aggregateID = Session.ID.create() + const fiber = yield* events.durable({ aggregateID }).pipe(Stream.take(4), Stream.runCollect, Effect.forkScoped) + yield* Effect.yieldNow + + yield* events.publishMany( + Array.from({ length: 4 }, (_, index) => + EventV2.publication(SessionV1.Event.MessageRemoved, { + sessionID: aggregateID, + messageID: SessionV1.MessageID.ascending(`msg_${index}`), + }), + ), + ) + + expect(Array.from(yield* Fiber.join(fiber)).map((event) => event.durable?.seq)).toEqual([0, 1, 2, 3]) + }), + ) + + it.effect("writes a batch larger than one insert chunk", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const aggregateID = EventV2.ID.create() + + yield* events.publishMany(batch(aggregateID, 450)) + + const rows = yield* rowsOf(aggregateID) + expect(rows).toHaveLength(450) + expect(rows.map((row) => row.seq)).toEqual(Array.from({ length: 450 }, (_, index) => index)) + }), + ) +}) diff --git a/packages/core/test/session-runner-tool-events.test.ts b/packages/core/test/session-runner-tool-events.test.ts index f96ea4dea2b0..501d2d75bad5 100644 --- a/packages/core/test/session-runner-tool-events.test.ts +++ b/packages/core/test/session-runner-tool-events.test.ts @@ -26,6 +26,7 @@ const capture = () => { }) return event }), + publishMany: () => Effect.succeed([]), subscribe: () => Stream.empty, all: () => Stream.empty, durable: () => Stream.empty, diff --git a/packages/opencode/src/event-v2-bridge.ts b/packages/opencode/src/event-v2-bridge.ts index 8e65f2752df7..764f229704df 100644 --- a/packages/opencode/src/event-v2-bridge.ts +++ b/packages/opencode/src/event-v2-bridge.ts @@ -32,6 +32,22 @@ const layer = Layer.effect( }) }) + // documentor: the same location routing as `publish`, for a batch. + const publishMany: EventV2.Interface["publishMany"] = (publications, options) => + Effect.gen(function* () { + if (options?.location) return yield* events.publishMany(publications, options) + const ctx = yield* InstanceRef + if (!ctx) return yield* events.publishMany(publications, options) + const workspaceID = yield* WorkspaceRef + return yield* events.publishMany(publications, { + location: new Location.Info({ + directory: AbsolutePath.make(ctx.directory), + ...(workspaceID ? { workspaceID } : {}), + project: { id: Project.ID.make(ctx.project.id), directory: AbsolutePath.make(ctx.worktree) }, + }), + }) + }) + const unsubscribe = yield* events.listen((event) => Effect.gen(function* () { const ctx = yield* InstanceRef @@ -62,7 +78,7 @@ const layer = Layer.effect( ) yield* Effect.addFinalizer(() => unsubscribe) - return Service.of({ ...events, publish }) + return Service.of({ ...events, publish, publishMany }) }), ) diff --git a/packages/opencode/src/session/session.ts b/packages/opencode/src/session/session.ts index cb3544c8fab5..7e1abbfc35cc 100644 --- a/packages/opencode/src/session/session.ts +++ b/packages/opencode/src/session/session.ts @@ -10,6 +10,7 @@ import type { ProviderMetadata, Usage } from "@opencode-ai/llm" import { InstallationVersion } from "@opencode-ai/core/installation/version" import { Database } from "@opencode-ai/core/database/database" import { EventV2Bridge } from "@/event-v2-bridge" +import { EventV2 } from "@opencode-ai/core/event" import { SessionV2 } from "@opencode-ai/core/session" import * as SessionExecutionLocal from "@opencode-ai/core/session/execution/local" import { locationServiceMapLayer } from "@opencode-ai/core/location-services" @@ -723,35 +724,54 @@ const layer: Layer.Layer< title, metadata: structuredClone(original.metadata), }) - const msgs = yield* messages({ sessionID: input.sessionID }) - const idMap = new Map() - const target = input.messageID ? msgs.findIndex((msg) => msg.info.id === input.messageID) : msgs.length - - for (const msg of msgs.slice(0, target < 0 ? msgs.length : target)) { - const newID = MessageID.ascending() - idMap.set(msg.info.id, newID) - - const parentID = msg.info.role === "assistant" && msg.info.parentID ? idMap.get(msg.info.parentID) : undefined - const cloned = yield* updateMessage({ - ...msg.info, - sessionID: session.id, - id: newID, - ...(parentID && { parentID }), - }) - - for (const part of msg.parts) { - const p: SessionV1.Part = { - ...part, - id: PartID.ascending(), - messageID: cloned.id, + // documentor: the copy is published as one batch. Against a remote Postgres a single publish + // is about nine round trips, so a long chat copied event by event takes tens of seconds. The + // batch commits all or nothing; the session created above is removed again if it does not. + yield* Effect.gen(function* () { + const msgs = yield* messages({ sessionID: input.sessionID }) + const idMap = new Map() + const target = input.messageID ? msgs.findIndex((msg) => msg.info.id === input.messageID) : msgs.length + const publications = new Array() + + for (const msg of msgs.slice(0, target < 0 ? msgs.length : target)) { + const newID = MessageID.ascending() + idMap.set(msg.info.id, newID) + + const parentID = msg.info.role === "assistant" && msg.info.parentID ? idMap.get(msg.info.parentID) : undefined + const cloned: SessionV1.Info = { + ...msg.info, sessionID: session.id, + id: newID, + ...(parentID && { parentID }), } - if (p.type === "compaction" && p.tail_start_id) { - p.tail_start_id = idMap.get(p.tail_start_id) + publications.push( + EventV2.publication(SessionV1.Event.MessageUpdated, { sessionID: cloned.sessionID, info: cloned }), + ) + + for (const part of msg.parts) { + const p: SessionV1.Part = { + ...part, + id: PartID.ascending(), + messageID: cloned.id, + sessionID: session.id, + } + if (p.type === "compaction" && p.tail_start_id) { + p.tail_start_id = idMap.get(p.tail_start_id) + } + publications.push( + EventV2.publication(SessionV1.Event.PartUpdated, { + sessionID: p.sessionID, + part: structuredClone(p), + time: Date.now(), + }), + ) } - yield* updatePart(p) } - } + yield* events.publishMany(publications) + }).pipe( + Effect.withSpan("Session.fork.copy"), + Effect.onError(() => remove(session.id).pipe(Effect.ignore)), + ) return session }) diff --git a/packages/opencode/test/session/fork-batch.test.ts b/packages/opencode/test/session/fork-batch.test.ts new file mode 100644 index 000000000000..b4793cd65e58 --- /dev/null +++ b/packages/opencode/test/session/fork-batch.test.ts @@ -0,0 +1,150 @@ +import { describe, expect } from "bun:test" +import { eq, sql } from "drizzle-orm" +import { Effect, Exit } from "effect" +import { Database } from "@opencode-ai/core/database/database" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { EventV2 } from "@opencode-ai/core/event" +import { EventTable } from "@opencode-ai/core/event/sql" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { SessionTable } from "@opencode-ai/core/session/sql" +import { SessionV1 } from "@opencode-ai/core/v1/session" +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" + +const it = testEffect( + LayerNode.compile( + LayerNode.group([SessionNs.node, MessageV2.node, SessionProjector.node, Database.node, EventV2.node]), + ), +) + +/** Writes `count` messages, alternating user and assistant, each assistant reply with three parts. */ +const seed = (sessionID: SessionID, count: number) => + Effect.gen(function* () { + const session = yield* SessionNs.Service + let parent: MessageID | undefined + for (let index = 0; index < count; index++) { + const id = MessageID.ascending() + const user = index % 2 === 0 + const info = user + ? { + id, + sessionID, + role: "user", + time: { created: Date.now() }, + agent: "test", + model: { providerID: "test", modelID: "test" }, + tools: {}, + mode: "", + } + : { + id, + sessionID, + role: "assistant", + time: { created: Date.now() }, + parentID: parent, + modelID: "test", + providerID: "test", + mode: "", + agent: "default", + path: { cwd: "/", root: "/" }, + cost: 0, + tokens: { input: 0, output: 0, reasoning: 0, cache: { read: 0, write: 0 } }, + } + const parts = Array.from({ length: user ? 1 : 3 }, (_, part) => ({ + id: PartID.ascending(), + sessionID, + messageID: id, + type: "text" as const, + text: `message ${index} part ${part}`, + })) + yield* session.saveMessage(info as unknown as SessionV1.Info, parts) + if (user) parent = id + } + }) + +const currentSchema = Effect.gen(function* () { + const { db } = yield* Database.Service + return (yield* db.get<{ name: string }>(sql`select current_schema() as name`))!.name +}) + +/** Aggregates whose sequence numbers are not exactly 0..n. Empty means every log is gap-free. */ +const broken = (schema: string) => + sql.raw(` + select aggregate_id from "${schema}".event group by aggregate_id + having count(*) <> max(seq) + 1 or count(distinct seq) <> count(*) + `) + +describe("documentor: fork publishes its copy as one batch", () => { + it.instance("copies every message and part with a gap-free event log", () => + Effect.gen(function* () { + const session = yield* SessionNs.Service + const { db } = yield* Database.Service + const origin = yield* session.create({}) + yield* seed(origin.id, 10) + + const fork = yield* session.fork({ sessionID: origin.id }) + + const source = yield* session.messages({ sessionID: origin.id }) + const copied = yield* session.messages({ sessionID: fork.id }) + const text = (msgs: typeof source) => + msgs.map((msg) => [msg.info.role, msg.parts.map((part) => (part as { text?: string }).text)]) + expect(text(copied)).toEqual(text(source)) + const byID = new Map(copied.map((msg) => [msg.info.id, msg])) + for (const msg of copied) { + if (msg.info.role === "assistant") expect(byID.get(msg.info.parentID)?.info.role).toBe("user") + } + const forkEvents = yield* db.select().from(EventTable).where(eq(EventTable.aggregate_id, fork.id)).all() + expect(forkEvents.length).toBeGreaterThanOrEqual(1 + 10 + 20) + expect(yield* db.all(broken(yield* currentSchema))).toEqual([]) + }), + ) + + it.instance("lets another chat in the same schema write while a fork is copying", () => + Effect.gen(function* () { + const session = yield* SessionNs.Service + const origin = yield* session.create({}) + const other = yield* session.create({}) + yield* seed(origin.id, 60) + + const [fork] = yield* Effect.all( + [ + session.fork({ sessionID: origin.id }), + Effect.all( + Array.from({ length: 10 }, (_, index) => + session.setTitle({ sessionID: other.id, title: `title ${index}` }), + ), + { concurrency: 4 }, + ), + ], + { concurrency: "unbounded" }, + ) + + expect((yield* session.messages({ sessionID: fork.id })).length).toBe(60) + expect((yield* session.get(other.id)).title).toMatch(/^title \d$/) + const { db } = yield* Database.Service + expect(yield* db.all(broken(yield* currentSchema))).toEqual([]) + }), + ) + + it.instance("removes the forked session again when the copy cannot commit", () => + Effect.gen(function* () { + const session = yield* SessionNs.Service + const events = yield* EventV2.Service + const { db } = yield* Database.Service + const origin = yield* session.create({}) + yield* seed(origin.id, 4) + yield* events.project(SessionV1.Event.PartUpdated, (event) => + event.data.sessionID === origin.id ? Effect.void : Effect.die("projector failed"), + ) + const sessionsBefore = (yield* db.select({ id: SessionTable.id }).from(SessionTable).all()).length + + const exit = yield* session.fork({ sessionID: origin.id }).pipe(Effect.exit) + + expect(Exit.isFailure(exit)).toBeTrue() + expect((yield* db.select({ id: SessionTable.id }).from(SessionTable).all()).length).toBe(sessionsBefore) + expect((yield* session.messages({ sessionID: origin.id })).length).toBe(4) + }), + ) +})