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