diff --git a/packages/core/src/event.ts b/packages/core/src/event.ts index c92ac0ac2ce3..e20f980242e5 100644 --- a/packages/core/src/event.ts +++ b/packages/core/src/event.ts @@ -3,7 +3,7 @@ export * as EventV2 from "./event" import { Cause, Context, Effect, Layer, Option, PubSub, Queue, Schema, Stream } from "effect" import { Event } from "@opencode-ai/schema/event" import type { Data, Definition, Payload } from "@opencode-ai/schema/event" -import { and, asc, eq, gt, inArray } from "drizzle-orm" +import { and, asc, eq, gt, inArray, sql } from "drizzle-orm" import { Database } from "./database/database" import { EventSequenceTable, EventTable } from "./event/sql" import { Location } from "./location" @@ -121,6 +121,13 @@ export interface PublishOptions { readonly location?: Location.Ref /** Local operational projection committed atomically with a new durable event. Not replayed or serialized. */ readonly commit?: (seq: number) => Effect.Effect + /** + * When false, the durable event is projected locally but NOT persisted to the + * event table or sequence. The payload is still notified to in-process + * listeners (SSE, UI) but carries no `durable` envelope, so cross-instance + * sync does not observe it. Defaults to true (full event sourcing). + */ + readonly persist?: boolean } export interface Interface { @@ -212,6 +219,7 @@ export const layerWith = (options?: LayerOptions) => readonly strictOwner?: boolean }, commit?: (seq: number) => Effect.Effect, + persist = true, ) { return Effect.gen(function* () { const durable = definition?.durable @@ -234,6 +242,40 @@ export const layerWith = (options?: LayerOptions) => ) } const list = projectors.get(event.type) ?? [] + if (!persist) { + // Local-only publish: project the event into the operational + // tables (MessageTable/PartTable/SessionTable) atomically, but + // do not append to the durable event log or advance the + // aggregate sequence. Returning undefined signals the caller to + // notify listeners with no `durable` envelope, so cross-instance + // sync never observes this event. The `commit` hook is not + // invoked either: it is documented as requiring a committed seq, + // and no caller combines `commit` with `persist:false`. The + // projector receives `durable.seq = -1` as a placeholder; none + // of the current projectors read seq (they upsert by entity id), + // so the value is inert — kept only to satisfy the Payload type. + return yield* Effect.uninterruptible( + Effect.gen(function* () { + yield* db + .transaction( + () => + Effect.gen(function* () { + const committed = { + ...event, + durable: { aggregateID, seq: -1, version: durable.version }, + } as Payload + for (const projector of list) { + yield* projector(committed) + } + return + }), + { behavior: "immediate" }, + ) + .pipe(Effect.orDie) + return undefined + }), + ) + } return yield* Effect.uninterruptible( Effect.gen(function* () { const committed = yield* db @@ -366,7 +408,12 @@ export const layerWith = (options?: LayerOptions) => }) } - function publishEvent(definition: D, event: Payload, commit?: PublishOptions["commit"]) { + function publishEvent( + definition: D, + event: Payload, + commit?: PublishOptions["commit"], + persist = true, + ) { return Effect.gen(function* () { if (!definition?.durable && commit) return yield* Effect.die( @@ -376,7 +423,7 @@ export const layerWith = (options?: LayerOptions) => }), ) if (definition?.durable) { - const committed = yield* commitDurableEvent(definition, event as Payload, undefined, commit) + const committed = yield* commitDurableEvent(definition, event as Payload, undefined, commit, persist) if (committed) { event = { ...event, @@ -434,6 +481,7 @@ export const layerWith = (options?: LayerOptions) => data, } as Payload, options?.commit, + options?.persist ?? true, ) }) } @@ -636,3 +684,84 @@ export const layerWith = (options?: LayerOptions) => const layer = layerWith() export const node = makeGlobalNode({ service: Service, layer: layer, deps: [Database.node] }) + +export const SNAPSHOT_TYPES = ["message.updated", "message.part.updated"] as const + +/** + * Compact snapshot-like durable events, keeping only the latest occurrence per + * (aggregate, type, entity) and deleting intermediate full-state copies. + * + * These events carry the complete message/part payload on every update, so a + * single message renders N rows whose payloads are total supersets of their + * predecessors. Replaying the retained latest row reproduces the identical + * final projection (the projector upserts by id), while intermediate rows are + * pure write amplification. + * + * Deleting rows leaves `seq` gaps; sequence stream readers (`readAfter`, + * `history`) use `seq > after` so gaps are transparent. Replay packets are + * re-cost in the sender and validated against their own emitted `seq` order, + * not against DB adjacency, so gaps are also safe for sync replay. + * + * Non-snapshot lifecycle rows (`session.created`, `message.removed`, + * `message.part.delta`, ...) are never touched, and `event_sequence` is left + * at its current high-water mark. + */ +export const compactSnapshotEvents = Effect.fn("EventV2.compactSnapshotEvents")(function* ( + db: Database.Interface["db"], +) { + const snapshotTypes = SNAPSHOT_TYPES.map((type) => versionedType(type, 1)) + const stats = yield* db + .select({ + rows: sql`count(*)`, + bytes: sql`sum(length(data))`, + }) + .from(EventTable) + .where(inArray(EventTable.type, snapshotTypes)) + .get() + .pipe(Effect.orDie) + yield* db + .run( + sql.raw(` + DELETE FROM "event" + WHERE "type" IN ('message.updated.1', 'message.part.updated.1') + AND "id" NOT IN ( + SELECT "id" FROM ( + SELECT + "id", + ROW_NUMBER() OVER ( + PARTITION BY "aggregate_id", "type", "entity" + ORDER BY "seq" DESC + ) AS "rn" + FROM ( + SELECT + "id", + "aggregate_id", + "type", + "seq", + CASE "type" + WHEN 'message.updated.1' THEN json_extract("data", '$.info.id') + WHEN 'message.part.updated.1' THEN json_extract("data", '$.part.id') + END AS "entity" + FROM "event" + WHERE "type" IN ('message.updated.1', 'message.part.updated.1') + ) + ) + WHERE "rn" = 1 + ) + `), + ) + .pipe(Effect.orDie) + const remaining = yield* db + .select({ + rows: sql`count(*)`, + bytes: sql`sum(length(data))`, + }) + .from(EventTable) + .where(inArray(EventTable.type, snapshotTypes)) + .get() + .pipe(Effect.orDie) + return { + removed: (stats?.rows ?? 0) - (remaining?.rows ?? 0), + bytes: (stats?.bytes ?? 0) - (remaining?.bytes ?? 0), + } +}) diff --git a/packages/core/test/event-compact.test.ts b/packages/core/test/event-compact.test.ts new file mode 100644 index 000000000000..bf4d0593e10f --- /dev/null +++ b/packages/core/test/event-compact.test.ts @@ -0,0 +1,92 @@ +import { describe, expect } from "bun:test" +import { Effect, Layer } from "effect" +import { EventV2 } from "@opencode-ai/core/event" +import { Database } from "@opencode-ai/core/database/database" +import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql" +import { Location } from "@opencode-ai/core/location" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { WorkspaceV2 } from "@opencode-ai/core/workspace" +import { eq } from "drizzle-orm" +import { location } from "./fixture/location" +import { testEffect } from "./lib/effect" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" + +const locationLayer = Layer.succeed( + Location.Service, + Location.Service.of( + location({ directory: AbsolutePath.make("project"), workspaceID: WorkspaceV2.ID.make("wrk_test") }), + ), +) + +const it = testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [[Location.node, locationLayer]]), +) + +const insert = (db: Database.Interface["db"]) => + (rows: { id: string; aggregateID: string; seq: number; type: string; data: Record }[]) => + db + .insert(EventTable) + .values( + rows.map((row) => ({ + id: row.id, + aggregate_id: row.aggregateID, + seq: row.seq, + type: row.type, + data: row.data, + })) as never, + ) + .run() + .pipe(Effect.orDie) + +describe("EventV2.compactSnapshotEvents", () => { + it.effect("keeps only the latest message.updated and part.updated per entity", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db.insert(EventSequenceTable).values([{ aggregate_id: "ses_a", seq: 10 }]).run().pipe(Effect.orDie) + yield* insert(db)([ + { id: "e1", aggregateID: "ses_a", seq: 1, type: "message.updated.1", data: { info: { id: "msg_m1", text: "v1" } } }, + { id: "e9", aggregateID: "ses_a", seq: 9, type: "session.created.1", data: { sessionID: "ses_a" } }, + { id: "e10", aggregateID: "ses_a", seq: 10, type: "session.updated.1", data: { sessionID: "ses_a" } }, + { id: "e2", aggregateID: "ses_a", seq: 2, type: "message.updated.1", data: { info: { id: "msg_m1", text: "v2" } } }, + { id: "e3", aggregateID: "ses_a", seq: 3, type: "message.updated.1", data: { info: { id: "msg_m1", text: "v3" } } }, + { id: "e4", aggregateID: "ses_a", seq: 4, type: "message.updated.1", data: { info: { id: "msg_m2", text: "x" } } }, + { id: "e5", aggregateID: "ses_a", seq: 5, type: "message.part.updated.1", data: { part: { id: "prt_p1", text: "a" } } }, + { id: "e6", aggregateID: "ses_a", seq: 6, type: "message.part.updated.1", data: { part: { id: "prt_p1", text: "ab" } } }, + { id: "e7", aggregateID: "ses_a", seq: 7, type: "message.part.updated.1", data: { part: { id: "prt_p1", text: "abc" } } }, + { id: "e8", aggregateID: "ses_a", seq: 8, type: "message.removed.1", data: { sessionID: "ses_a", messageID: "msg_m9" } }, + ]) + + const result = yield* EventV2.compactSnapshotEvents(db) + expect(result.removed).toBe(4) + + const rows = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, "ses_a")) + .all() + .pipe(Effect.orDie) + const updated = rows.filter((row) => row.type === "message.updated.1") + const parts = rows.filter((row) => row.type === "message.part.updated.1") + const removed = rows.filter((row) => row.type === "message.removed.1") + expect(updated).toHaveLength(2) + expect(parts).toHaveLength(1) + const texts = updated.map((row) => (row.data as { info?: { text?: string } }).info?.text) + expect(texts).toContain("x") + expect(texts).toContain("v3") + expect(removed).toHaveLength(1) + }), + ) + + it.effect("removes nothing when no duplicate snapshots exist", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + yield* db.insert(EventSequenceTable).values([{ aggregate_id: "ses_b", seq: 1 }]).run().pipe(Effect.orDie) + yield* insert(db)([ + { id: "e1", aggregateID: "ses_b", seq: 1, type: "message.updated.1", data: { info: { id: "msg_m1", text: "only" } } }, + ]) + const result = yield* EventV2.compactSnapshotEvents(db) + expect(result.removed).toBe(0) + }), + ) +}) \ No newline at end of file diff --git a/packages/core/test/event-persist-gate.test.ts b/packages/core/test/event-persist-gate.test.ts new file mode 100644 index 000000000000..869a0caa7295 --- /dev/null +++ b/packages/core/test/event-persist-gate.test.ts @@ -0,0 +1,105 @@ +import { describe, expect } from "bun:test" +import { Effect, Layer } from "effect" +import { EventV2 } from "@opencode-ai/core/event" +import { SessionV1 } from "@opencode-ai/schema/session-v1" +import { Database } from "@opencode-ai/core/database/database" +import { Session } from "@opencode-ai/schema/session" +import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql" +import { Location } from "@opencode-ai/core/location" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { WorkspaceV2 } from "@opencode-ai/core/workspace" +import { eq } from "drizzle-orm" +import { location } from "./fixture/location" +import { testEffect } from "./lib/effect" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" + +const locationLayer = Layer.succeed( + Location.Service, + Location.Service.of( + location({ directory: AbsolutePath.make("project"), workspaceID: WorkspaceV2.ID.make("wrk_test") }), + ), +) +const it = testEffect( + AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, Location.node]), [[Location.node, locationLayer]]), +) + +const messageUpdated = ( + sid: Session.ID, + mid: SessionV1.MessageID, +): EventV2.Data => + ({ + sessionID: sid, + info: { + role: "user", + sessionID: sid, + id: mid, + time: { created: 1 }, + files: [], + agents: [], + text: "hello", + agent: "build", + model: { providerID: "openrouter", modelID: "test/model" }, + }, + }) as never + +describe("EventV2.publish persist gate", () => { + it.effect("persist:false skips the event log entirely", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const { db } = yield* Database.Service + const sid = Session.ID.create() + const mid = SessionV1.MessageID.ascending() + + const notified = yield* events.publish(SessionV1.Event.MessageUpdated, messageUpdated(sid, mid), { + persist: false, + }) + + const eventRows = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, sid)) + .all() + .pipe(Effect.orDie) + const seqRows = yield* db + .select() + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, sid)) + .all() + .pipe(Effect.orDie) + + expect(eventRows).toHaveLength(0) + expect(seqRows).toHaveLength(0) + // Payload still delivered to the caller (and thus to PubSub/SSE). + expect(notified.type).toBe("message.updated") + expect(notified.durable).toBeUndefined() + }), + ) + + it.effect("persist:true (default) writes the event log and advances the sequence", () => + Effect.gen(function* () { + const events = yield* EventV2.Service + const { db } = yield* Database.Service + const sid = Session.ID.create() + const mid = SessionV1.MessageID.ascending() + + yield* events.publish(SessionV1.Event.MessageUpdated, messageUpdated(sid, mid)) + + const eventRows = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, sid)) + .all() + .pipe(Effect.orDie) + const seqRows = yield* db + .select() + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, sid)) + .all() + .pipe(Effect.orDie) + expect(eventRows).toHaveLength(1) + expect(seqRows).toHaveLength(1) + expect(seqRows[0]?.seq).toBe(0) + }), + ) +}) \ No newline at end of file diff --git a/packages/opencode/src/cli/cancelled-error.ts b/packages/opencode/src/cli/cancelled-error.ts new file mode 100644 index 000000000000..78c48c4b1e9a --- /dev/null +++ b/packages/opencode/src/cli/cancelled-error.ts @@ -0,0 +1,3 @@ +import { Schema } from "effect" + +export class CancelledError extends Schema.TaggedErrorClass()("UICancelledError", {}) {} diff --git a/packages/opencode/src/cli/cmd/agent.ts b/packages/opencode/src/cli/cmd/agent.ts index c9c1d2c1670f..a1f0f1ee1580 100644 --- a/packages/opencode/src/cli/cmd/agent.ts +++ b/packages/opencode/src/cli/cmd/agent.ts @@ -1,6 +1,7 @@ import { cmd } from "./cmd" import * as prompts from "@clack/prompts" import { UI } from "../ui" +import { CancelledError } from "../cancelled-error" import { Global } from "@opencode-ai/core/global" import path from "path" import fs from "fs/promises" @@ -105,7 +106,7 @@ const AgentCreateCommand = effectCmd({ }, ], }) - if (prompts.isCancel(scopeResult)) throw new UI.CancelledError() + if (prompts.isCancel(scopeResult)) throw new CancelledError() scope = scopeResult } targetPath = path.join(scope === "global" ? Global.Path.config : path.join(ctx.worktree, ".opencode"), "agents") @@ -121,7 +122,7 @@ const AgentCreateCommand = effectCmd({ placeholder: "What should this agent do?", validate: (x) => (x && x.length > 0 ? undefined : "Required"), }) - if (prompts.isCancel(query)) throw new UI.CancelledError() + if (prompts.isCancel(query)) throw new CancelledError() description = query } @@ -132,7 +133,7 @@ const AgentCreateCommand = effectCmd({ const generated = await runLocalEffect(agentSvc.generate({ description, model })).catch((error) => { spinner.stop(`LLM failed to generate agent: ${error.message}`, 1) if (isFullyNonInteractive) process.exit(1) - throw new UI.CancelledError() + throw new CancelledError() }) spinner.stop(`Agent ${generated.identifier} generated`) @@ -149,7 +150,7 @@ const AgentCreateCommand = effectCmd({ })), initialValues: AVAILABLE_PERMISSIONS, }) - if (prompts.isCancel(result)) throw new UI.CancelledError() + if (prompts.isCancel(result)) throw new CancelledError() selected = result } @@ -179,7 +180,7 @@ const AgentCreateCommand = effectCmd({ ], initialValue: "all" as const, }) - if (prompts.isCancel(modeResult)) throw new UI.CancelledError() + if (prompts.isCancel(modeResult)) throw new CancelledError() mode = modeResult } @@ -216,7 +217,7 @@ const AgentCreateCommand = effectCmd({ process.exit(1) } prompts.log.error(`Agent file already exists: ${filePath}`) - throw new UI.CancelledError() + throw new CancelledError() } await Filesystem.write(filePath, content) diff --git a/packages/opencode/src/cli/cmd/db.ts b/packages/opencode/src/cli/cmd/db.ts index 9e7e37e18e91..80eb5150678f 100644 --- a/packages/opencode/src/cli/cmd/db.ts +++ b/packages/opencode/src/cli/cmd/db.ts @@ -3,7 +3,8 @@ import { spawn } from "child_process" import { Database } from "@opencode-ai/core/database/database" import { Effect } from "effect" import { sql } from "drizzle-orm" -import { effectCmd } from "../effect-cmd" +import { effectCmd, fail } from "../effect-cmd" +import { RuntimeFlags } from "@/effect/runtime-flags" const QueryCommand = effectCmd({ command: "$0 [query]", @@ -51,12 +52,43 @@ const PathCommand = effectCmd({ }), }) +const CompactCommand = effectCmd({ + command: "compact", + describe: + "delete duplicate snapshot events from the event log (local use only; incompatible with experimental workspaces sync)", + instance: false, + handler: Effect.fn("Cli.db.compact")(function* () { + const flags = yield* RuntimeFlags.Service + if (flags.experimentalWorkspaces) { + return yield* fail( + "db compact is not available while OPENCODE_EXPERIMENTAL_WORKSPACES is enabled: it leaves sequence gaps that break cross-instance sync history.", + ) + } + const { db } = yield* Database.Service + const EventV2 = yield* Effect.promise(() => import("@opencode-ai/core/event")) + const result = yield* EventV2.compactSnapshotEvents(db) + console.log( + `Removed ${result.removed} redundant snapshot events (${(result.bytes / 1024 / 1024).toFixed(1)} MiB of JSON payload).`, + ) + const sizeBefore = (yield* db.all(sql.raw(`PRAGMA page_count;`)).pipe(Effect.orDie)) as Array<{ page_count: number }> + yield* db.run(sql.raw(`VACUUM;`)).pipe(Effect.orDie) + const sizeAfter = (yield* db.all(sql.raw(`PRAGMA page_count;`)).pipe(Effect.orDie)) as Array<{ page_count: number }> + if (sizeBefore.length > 0 && sizeAfter.length > 0) { + const pagesBefore = Number(sizeBefore[0]?.page_count) + const pagesAfter = Number(sizeAfter[0]?.page_count) + console.log( + `DB pages: ${pagesBefore.toLocaleString()} -> ${pagesAfter.toLocaleString()} (-${(100 * (1 - pagesAfter / Math.max(pagesBefore, 1))).toFixed(1)}%)`, + ) + } + }), +}) + export const DbCommand = effectCmd({ command: "db", describe: "database tools", instance: false, builder: (yargs: Argv) => { - return yargs.command(QueryCommand).command(PathCommand).demandCommand() + return yargs.command(QueryCommand).command(PathCommand).command(CompactCommand).demandCommand() }, handler: Effect.fn("Cli.db")(function* () {}), }) diff --git a/packages/opencode/src/cli/cmd/export.ts b/packages/opencode/src/cli/cmd/export.ts index 8c3aa1618ae1..726fb15cf3c8 100644 --- a/packages/opencode/src/cli/cmd/export.ts +++ b/packages/opencode/src/cli/cmd/export.ts @@ -4,6 +4,7 @@ import { MessageV2 } from "../../session/message-v2" import { SessionID } from "../../session/schema" import { effectCmd, fail } from "../effect-cmd" import { UI } from "../ui" +import { CancelledError } from "../cancelled-error" import * as prompts from "@clack/prompts" import { EOL } from "os" import { Effect } from "effect" @@ -270,7 +271,7 @@ const run = Effect.fn("Cli.export.body")(function* (args: { sessionID?: string; ) if (prompts.isCancel(selectedSession)) { - return yield* Effect.die(new UI.CancelledError()) + return yield* Effect.die(new CancelledError()) } sessionID = selectedSession diff --git a/packages/opencode/src/cli/cmd/github.handler.ts b/packages/opencode/src/cli/cmd/github.handler.ts index fcf44279ce7f..391bddcf6e0e 100644 --- a/packages/opencode/src/cli/cmd/github.handler.ts +++ b/packages/opencode/src/cli/cmd/github.handler.ts @@ -17,6 +17,7 @@ import type { PullRequestEvent, } from "@octokit/webhooks-types" import { UI } from "../ui" +import { CancelledError } from "../cancelled-error" import { ModelsDev } from "@opencode-ai/core/models-dev" import { InstanceRef } from "@/effect/instance-ref" import { SessionShare } from "@/share/session" @@ -211,7 +212,7 @@ export const githubInstall = Effect.fn("Cli.github.install")(function* () { const project = ctx.project if (project.vcs !== "git") { prompts.log.error(`Could not find git repository. Please run this command from a git repository.`) - throw new UI.CancelledError() + throw new CancelledError() } // Get repo info @@ -221,7 +222,7 @@ export const githubInstall = Effect.fn("Cli.github.install")(function* () { const parsed = parseGitHubRemote(info) if (!parsed) { prompts.log.error(`Could not find git repository. Please run this command from a git repository.`) - throw new UI.CancelledError() + throw new CancelledError() } return { owner: parsed.owner, repo: parsed.repo, root: ctx.worktree } } @@ -251,7 +252,7 @@ export const githubInstall = Effect.fn("Cli.github.install")(function* () { ), }) - if (prompts.isCancel(provider)) throw new UI.CancelledError() + if (prompts.isCancel(provider)) throw new CancelledError() return provider } @@ -273,7 +274,7 @@ export const githubInstall = Effect.fn("Cli.github.install")(function* () { ), }) - if (prompts.isCancel(model)) throw new UI.CancelledError() + if (prompts.isCancel(model)) throw new CancelledError() return model } @@ -312,7 +313,7 @@ export const githubInstall = Effect.fn("Cli.github.install")(function* () { s.stop( `Failed to detect GitHub app installation. Make sure to install the app for the \`${app.owner}/${app.repo}\` repository.`, ) - throw new UI.CancelledError() + throw new CancelledError() } retries++ diff --git a/packages/opencode/src/cli/cmd/mcp.ts b/packages/opencode/src/cli/cmd/mcp.ts index c2d2ee2f3b73..fa4a36444e0d 100644 --- a/packages/opencode/src/cli/cmd/mcp.ts +++ b/packages/opencode/src/cli/cmd/mcp.ts @@ -8,6 +8,7 @@ import { UnauthorizedError } from "@modelcontextprotocol/sdk/client/auth.js" import { LATEST_PROTOCOL_VERSION } from "@modelcontextprotocol/sdk/types.js" import * as prompts from "@clack/prompts" import { UI } from "../ui" +import { CancelledError } from "../cancelled-error" import { MCP } from "../../mcp" import { McpAuth } from "../../mcp/auth" import { McpOAuthProvider } from "../../mcp/oauth-provider" @@ -220,7 +221,7 @@ export const McpAuthCommand = effectCmd({ options, }), ) - if (prompts.isCancel(selected)) throw new UI.CancelledError() + if (prompts.isCancel(selected)) throw new CancelledError() serverName = selected } @@ -375,7 +376,7 @@ export const McpLogoutCommand = effectCmd({ }), }), ) - if (prompts.isCancel(selected)) throw new UI.CancelledError() + if (prompts.isCancel(selected)) throw new CancelledError() serverName = selected } @@ -529,7 +530,7 @@ export const McpAddCommand = effectCmd({ }, ], }) - if (prompts.isCancel(scopeResult)) throw new UI.CancelledError() + if (prompts.isCancel(scopeResult)) throw new CancelledError() configPath = scopeResult } @@ -537,7 +538,7 @@ export const McpAddCommand = effectCmd({ message: "Enter MCP server name", validate: (x) => (x && x.length > 0 ? undefined : "Required"), }) - if (prompts.isCancel(name)) throw new UI.CancelledError() + if (prompts.isCancel(name)) throw new CancelledError() const type = await prompts.select({ message: "Select MCP server type", @@ -554,7 +555,7 @@ export const McpAddCommand = effectCmd({ }, ], }) - if (prompts.isCancel(type)) throw new UI.CancelledError() + if (prompts.isCancel(type)) throw new CancelledError() if (type === "local") { const command = await prompts.text({ @@ -562,7 +563,7 @@ export const McpAddCommand = effectCmd({ placeholder: "e.g., opencode x @modelcontextprotocol/server-filesystem", validate: (x) => (x && x.length > 0 ? undefined : "Required"), }) - if (prompts.isCancel(command)) throw new UI.CancelledError() + if (prompts.isCancel(command)) throw new CancelledError() const mcpConfig: ConfigMCPV1.Info = { type: "local", @@ -586,13 +587,13 @@ export const McpAddCommand = effectCmd({ return isValid ? undefined : "Invalid URL" }, }) - if (prompts.isCancel(url)) throw new UI.CancelledError() + if (prompts.isCancel(url)) throw new CancelledError() const useOAuth = await prompts.confirm({ message: "Does this server require OAuth authentication?", initialValue: false, }) - if (prompts.isCancel(useOAuth)) throw new UI.CancelledError() + if (prompts.isCancel(useOAuth)) throw new CancelledError() let mcpConfig: ConfigMCPV1.Info @@ -601,27 +602,27 @@ export const McpAddCommand = effectCmd({ message: "Do you have a pre-registered client ID?", initialValue: false, }) - if (prompts.isCancel(hasClientId)) throw new UI.CancelledError() + if (prompts.isCancel(hasClientId)) throw new CancelledError() if (hasClientId) { const clientId = await prompts.text({ message: "Enter client ID", validate: (x) => (x && x.length > 0 ? undefined : "Required"), }) - if (prompts.isCancel(clientId)) throw new UI.CancelledError() + if (prompts.isCancel(clientId)) throw new CancelledError() const hasSecret = await prompts.confirm({ message: "Do you have a client secret?", initialValue: false, }) - if (prompts.isCancel(hasSecret)) throw new UI.CancelledError() + if (prompts.isCancel(hasSecret)) throw new CancelledError() let clientSecret: string | undefined if (hasSecret) { const secret = await prompts.password({ message: "Enter client secret", }) - if (prompts.isCancel(secret)) throw new UI.CancelledError() + if (prompts.isCancel(secret)) throw new CancelledError() clientSecret = secret } diff --git a/packages/opencode/src/cli/cmd/providers.ts b/packages/opencode/src/cli/cmd/providers.ts index 3775123d83bd..aa813038219f 100644 --- a/packages/opencode/src/cli/cmd/providers.ts +++ b/packages/opencode/src/cli/cmd/providers.ts @@ -3,6 +3,7 @@ import { Auth } from "../../auth" import { cmd } from "./cmd" import { CliError, effectCmd, fail } from "../effect-cmd" import { UI } from "../ui" +import { CancelledError } from "../cancelled-error" import * as Prompt from "../effect/prompt" import { ModelsDev } from "@opencode-ai/core/models-dev" @@ -21,7 +22,7 @@ import { Effect, Option } from "effect" type PluginAuth = NonNullable const promptValue = (value: Option.Option) => { - if (Option.isNone(value)) return Effect.die(new UI.CancelledError()) + if (Option.isNone(value)) return Effect.die(new CancelledError()) return Effect.succeed(value.value) } diff --git a/packages/opencode/src/cli/lazy-command.ts b/packages/opencode/src/cli/lazy-command.ts new file mode 100644 index 000000000000..51f3212298de --- /dev/null +++ b/packages/opencode/src/cli/lazy-command.ts @@ -0,0 +1,43 @@ +import type { CommandModule } from "yargs" +import type { Argv } from "yargs" + +/** + * Lazy command registration for the CLI entrypoint. + * + * The full CLI statically imports every command module, which pulls in heavy + * dependencies (server, session, sdk, config schemas) and costs tens of + * seconds at startup — even for cheap invocations like `--version`. Register + * each command as a delegating module that only loads the real implementation + * when yargs parses an invocation that exercises it. `builder` and `handler` + * run lazily; `command`/`describe`/`aliases` stay eager so help text works. + */ +export const lazyCommand = ( + input: { + readonly command: string + readonly aliases?: readonly string[] + readonly describe?: string | false + readonly load: () => Promise> + readonly resolve: (mod: Record) => CommandModule + }, +): CommandModule => { + const command = input.command + const aliases = input.aliases + const describe = input.describe + const handle = async (args: unknown) => { + const mod = await input.load() + return input.resolve(mod).handler?.(args as U) + } + const build = async (args: Argv) => { + const mod = await input.load() + const builder = input.resolve(mod).builder as unknown + if (typeof builder === "function") return builder(args) + return args + } + return { + command, + aliases, + describe, + builder: build as never, + handler: handle as never, + } +} diff --git a/packages/opencode/src/cli/ui.ts b/packages/opencode/src/cli/ui.ts index 6ad6495cf10b..93d3b1986eae 100644 --- a/packages/opencode/src/cli/ui.ts +++ b/packages/opencode/src/cli/ui.ts @@ -1,5 +1,4 @@ import { EOL } from "os" -import { Schema } from "effect" import { logo as glyphs } from "./logo" const wordmark = [ @@ -9,8 +8,6 @@ const wordmark = [ `▀▀▀▀ █▀▀▀ ▀▀▀▀ ▀ ▀ ▀▀▀▀ ▀▀▀▀ ▀▀▀▀ ▀▀▀▀`, ] -export class CancelledError extends Schema.TaggedErrorClass()("UICancelledError", {}) {} - export const Style = { TEXT_HIGHLIGHT: "\x1b[96m", TEXT_HIGHLIGHT_BOLD: "\x1b[96m\x1b[1m", diff --git a/packages/opencode/src/index.ts b/packages/opencode/src/index.ts index 13540a73a36f..497c19e7b1ce 100644 --- a/packages/opencode/src/index.ts +++ b/packages/opencode/src/index.ts @@ -1,35 +1,20 @@ +import type { CommandModule } from "yargs" import yargs from "yargs" import { hideBin } from "yargs/helpers" -import { RunCommand } from "./cli/cmd/run" -import { GenerateCommand } from "./cli/cmd/generate" -import { ConsoleCommand } from "./cli/cmd/account" -import { ProvidersCommand } from "./cli/cmd/providers" -import { AgentCommand } from "./cli/cmd/agent" -import { UpgradeCommand } from "./cli/cmd/upgrade" -import { UninstallCommand } from "./cli/cmd/uninstall" -import { ModelsCommand } from "./cli/cmd/models" +import { lazyCommand } from "./cli/lazy-command" import { UI } from "./cli/ui" import { InstallationVersion } from "@opencode-ai/core/installation/version" -import { FormatError } from "./cli/error" -import { ServeCommand } from "./cli/cmd/serve" -import { DebugCommand } from "./cli/cmd/debug" -import { StatsCommand } from "./cli/cmd/stats" -import { McpCommand } from "./cli/cmd/mcp" -import { GithubCommand } from "./cli/cmd/github" -import { ExportCommand } from "./cli/cmd/export" -import { ImportCommand } from "./cli/cmd/import" -import { AttachCommand } from "./cli/cmd/attach" -import { TuiThreadCommand } from "./cli/cmd/tui" -import { AcpCommand } from "./cli/cmd/acp" import { EOL } from "os" -import { WebCommand } from "./cli/cmd/web" -import { PrCommand } from "./cli/cmd/pr" -import { SessionCommand } from "./cli/cmd/session" -import { DbCommand } from "./cli/cmd/db" -import { errorMessage } from "./util/error" -import { PluginCommand } from "./cli/cmd/plug" import { Heap } from "./cli/heap" +const lazy = (spec: { + readonly command: string + readonly aliases?: readonly string[] + readonly describe?: string | false + readonly load: () => Promise> + readonly resolve: (mod: Record) => CommandModule +}) => lazyCommand({ ...spec, load: spec.load as never, resolve: spec.resolve as never }) + const args = hideBin(process.argv) function show(out: string) { @@ -78,29 +63,29 @@ const cli = yargs(args) }) .usage("") .completion("completion", "generate shell completion script") - .command(AcpCommand) - .command(McpCommand) - .command(TuiThreadCommand) - .command(AttachCommand) - .command(RunCommand) - .command(GenerateCommand) - .command(DebugCommand) - .command(ConsoleCommand) - .command(ProvidersCommand) - .command(AgentCommand) - .command(UpgradeCommand) - .command(UninstallCommand) - .command(ServeCommand) - .command(WebCommand) - .command(ModelsCommand) - .command(StatsCommand) - .command(ExportCommand) - .command(ImportCommand) - .command(GithubCommand) - .command(PrCommand) - .command(SessionCommand) - .command(PluginCommand) - .command(DbCommand) + .command(lazy({ command: "acp", describe: "start ACP (Agent Client Protocol) server", load: () => import("./cli/cmd/acp"), resolve: (m) => m.AcpCommand })) + .command(lazy({ command: "mcp", describe: "manage MCP (Model Context Protocol) servers", load: () => import("./cli/cmd/mcp"), resolve: (m) => m.McpCommand })) + .command(lazy({ command: "$0 [project]", describe: "start opencode tui", load: () => import("./cli/cmd/tui"), resolve: (m) => m.TuiThreadCommand })) + .command(lazy({ command: "attach ", describe: "attach to a running opencode server", load: () => import("./cli/cmd/attach"), resolve: (m) => m.AttachCommand })) + .command(lazy({ command: "run [message..]", describe: "run opencode with a message", load: () => import("./cli/cmd/run"), resolve: (m) => m.RunCommand })) + .command(lazy({ command: "generate", load: () => import("./cli/cmd/generate"), resolve: (m) => m.GenerateCommand })) + .command(lazy({ command: "debug", describe: "debugging and troubleshooting tools", load: () => import("./cli/cmd/debug"), resolve: (m) => m.DebugCommand })) + .command(lazy({ command: "console", describe: false, load: () => import("./cli/cmd/account"), resolve: (m) => m.ConsoleCommand })) + .command(lazy({ command: "providers", aliases: ["auth"], describe: "manage AI providers and credentials", load: () => import("./cli/cmd/providers"), resolve: (m) => m.ProvidersCommand })) + .command(lazy({ command: "agent", describe: "manage agents", load: () => import("./cli/cmd/agent"), resolve: (m) => m.AgentCommand })) + .command(lazy({ command: "upgrade [target]", describe: "upgrade opencode to the latest or a specific version", load: () => import("./cli/cmd/upgrade"), resolve: (m) => m.UpgradeCommand })) + .command(lazy({ command: "uninstall", describe: "uninstall opencode and remove all related files", load: () => import("./cli/cmd/uninstall"), resolve: (m) => m.UninstallCommand })) + .command(lazy({ command: "serve", describe: "starts a headless opencode server", load: () => import("./cli/cmd/serve"), resolve: (m) => m.ServeCommand })) + .command(lazy({ command: "web", describe: "start opencode server and open web interface", load: () => import("./cli/cmd/web"), resolve: (m) => m.WebCommand })) + .command(lazy({ command: "models [provider]", describe: "list all available models", load: () => import("./cli/cmd/models"), resolve: (m) => m.ModelsCommand })) + .command(lazy({ command: "stats", describe: "show token usage and cost statistics", load: () => import("./cli/cmd/stats"), resolve: (m) => m.StatsCommand })) + .command(lazy({ command: "export [sessionID]", describe: "export session data as JSON", load: () => import("./cli/cmd/export"), resolve: (m) => m.ExportCommand })) + .command(lazy({ command: "import ", describe: "import session data from JSON file or URL", load: () => import("./cli/cmd/import"), resolve: (m) => m.ImportCommand })) + .command(lazy({ command: "github", describe: "manage GitHub agent", load: () => import("./cli/cmd/github"), resolve: (m) => m.GithubCommand })) + .command(lazy({ command: "pr ", describe: "fetch and checkout a GitHub PR branch, then run opencode", load: () => import("./cli/cmd/pr"), resolve: (m) => m.PrCommand })) + .command(lazy({ command: "session", describe: "manage sessions", load: () => import("./cli/cmd/session"), resolve: (m) => m.SessionCommand })) + .command(lazy({ command: "plugin ", aliases: ["plug"], describe: "install plugin and update config", load: () => import("./cli/cmd/plug"), resolve: (m) => m.PluginCommand })) + .command(lazy({ command: "db", describe: "database tools", load: () => import("./cli/cmd/db"), resolve: (m) => m.DbCommand })) .fail((msg, err) => { if ( msg?.startsWith("Unknown argument") || @@ -117,15 +102,14 @@ const cli = yargs(args) try { if (args.includes("-h") || args.includes("--help")) { - await cli.parse(args, (err: Error | undefined, _argv: unknown, out: string) => { - if (err) throw err - if (!out) return - show(out) - }) + const helpText = await cli.getHelp() + show(helpText) } else { await cli.parse() } } catch (e) { + const { FormatError } = await import("./cli/error") + const { errorMessage } = await import("./util/error") const formatted = FormatError(e) if (formatted) UI.error(formatted) if (formatted === undefined) { diff --git a/packages/opencode/src/session/session.ts b/packages/opencode/src/session/session.ts index a2a91cd47b5e..51d28353edd4 100644 --- a/packages/opencode/src/session/session.ts +++ b/packages/opencode/src/session/session.ts @@ -626,19 +626,32 @@ const layer: Layer.Layer< } }) + // Only persist message/part updates to the durable event log when + // workspaces (cross-instance sync) are enabled. Locally the projected + // tables are the sole reader (UI/SSE/LLM); the event rows are dead weight + // that grew the log superlinearly for long streaming turns. Workspaces ON + // keeps them, preserving byte-identical sync behavior. const updateMessage = (msg: T): Effect.Effect => Effect.gen(function* () { - yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID: msg.sessionID, info: msg }) + yield* events.publish( + SessionV1.Event.MessageUpdated, + { sessionID: msg.sessionID, info: msg }, + { persist: flags.experimentalWorkspaces }, + ) return msg }).pipe(Effect.withSpan("Session.updateMessage")) const updatePart = (part: T): Effect.Effect => Effect.gen(function* () { - yield* events.publish(SessionV1.Event.PartUpdated, { - sessionID: part.sessionID, - part: structuredClone(part), - time: Date.now(), - }) + yield* events.publish( + SessionV1.Event.PartUpdated, + { + sessionID: part.sessionID, + part: structuredClone(part), + time: Date.now(), + }, + { persist: flags.experimentalWorkspaces }, + ) return part }).pipe(Effect.withSpan("Session.updatePart")) diff --git a/packages/opencode/test/cli/error.test.ts b/packages/opencode/test/cli/error.test.ts index b29ca2b3bae1..353554f8553b 100644 --- a/packages/opencode/test/cli/error.test.ts +++ b/packages/opencode/test/cli/error.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test" import { AccountTransportError } from "../../src/account/schema" import { FormatError } from "../../src/cli/error" -import { UI } from "../../src/cli/ui" +import { CancelledError } from "../../src/cli/cancelled-error" describe("cli.error", () => { test("formats legacy and tagged config errors the same way", () => { @@ -90,6 +90,6 @@ describe("cli.error", () => { }) test("formats cancelled UI errors as empty output", () => { - expect(FormatError(new UI.CancelledError())).toBe("") + expect(FormatError(new CancelledError())).toBe("") }) }) diff --git a/packages/opencode/test/session/persist-gate.test.ts b/packages/opencode/test/session/persist-gate.test.ts new file mode 100644 index 000000000000..7347ec313ff1 --- /dev/null +++ b/packages/opencode/test/session/persist-gate.test.ts @@ -0,0 +1,170 @@ +import { describe, expect } from "bun:test" +import { Deferred, Effect, Layer } from "effect" +import { SessionV1 } from "@opencode-ai/core/v1/session" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { Session as SessionNs } from "@/session/session" +import { MessageID, PartID } from "../../src/session/schema" +import { CrossSpawnSpawner } from "@opencode-ai/core/cross-spawn-spawner" +import { testEffect } from "../lib/effect" +import { RuntimeFlags } from "@/effect/runtime-flags" +import { EventV2Bridge } from "@/event-v2-bridge" +import { AppNodeBuilder } from "@opencode-ai/core/effect/app-node-builder" +import { LayerNode } from "@opencode-ai/core/effect/layer-node" +import { InstanceStore } from "@/project/instance-store" +import { InstanceBootstrap } from "@/project/bootstrap" +import { Database } from "@opencode-ai/core/database/database" +import { EventSequenceTable, EventTable } from "@opencode-ai/core/event/sql" +import { MessageTable, PartTable } from "@opencode-ai/core/session/sql" +import { eq } from "drizzle-orm" + +const it = testEffect( + AppNodeBuilder.build( + LayerNode.group([ + SessionNs.node, + EventV2Bridge.node, + SessionProjector.node, + CrossSpawnSpawner.node, + InstanceStore.node, + Database.node, + ]), + [ + [RuntimeFlags.node, RuntimeFlags.layer({ experimentalWorkspaces: false })], + [ + InstanceBootstrap.node, + Layer.succeed(InstanceBootstrap.Service, InstanceBootstrap.Service.of({ run: Effect.void })), + ], + ], + ), +) + +const itWithWorkspaces = testEffect( + AppNodeBuilder.build( + LayerNode.group([ + SessionNs.node, + EventV2Bridge.node, + SessionProjector.node, + CrossSpawnSpawner.node, + InstanceStore.node, + Database.node, + ]), + [ + [RuntimeFlags.node, RuntimeFlags.layer({ experimentalWorkspaces: true })], + [ + InstanceBootstrap.node, + Layer.succeed(InstanceBootstrap.Service, InstanceBootstrap.Service.of({ run: Effect.void })), + ], + ], + ), +) + +describe("local persist gate (experimentalWorkspaces off)", () => { + it.instance("projects message/part but writes nothing to the event log", () => + Effect.gen(function* () { + const session = yield* SessionNs.Service + const { db } = yield* Database.Service + const events = yield* EventV2Bridge.Service + const created = yield* session.create({ title: "gate" }) + const received = yield* Deferred.make() + const unsub = yield* events.listen((event) => { + if (event.type.includes("message.updated") || event.type.includes("part")) { + Deferred.doneUnsafe(received, Effect.succeed(event.type)) + } + return Effect.void + }) + const info = yield* session.updateMessage({ + id: MessageID.ascending(), + sessionID: created.id, + role: "user", + agent: "build", + model: { providerID: "test", modelID: "test" }, + time: { created: Date.now() }, + tools: {}, + mode: "", + } as unknown as SessionV1.Info) + yield* session.updatePart({ + id: PartID.ascending(), + sessionID: created.id, + messageID: info.id, + type: "text", + text: "hello world", + }) + + // Projection tables contain the message and part. + const messages = yield* db + .select() + .from(MessageTable) + .where(eq(MessageTable.id, info.id)) + .all() + .pipe(Effect.orDie) + const parts = yield* db + .select() + .from(PartTable) + .where(eq(PartTable.message_id, info.id)) + .all() + .pipe(Effect.orDie) + expect(messages).toHaveLength(1) + expect(parts).toHaveLength(1) + + // The durable event log and sequence are untouched for this session. + const snapshots = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.type, "message.updated.1")) + .all() + .pipe(Effect.orDie) + const seq = yield* db + .select() + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, created.id)) + .all() + .pipe(Effect.orDie) + // session.created persists one sequence row; the gated message update adds none. + expect(snapshots).toHaveLength(0) + expect(seq).toHaveLength(1) + expect(seq[0]?.seq).toBe(0) + + // The event is still delivered to in-process subscribers (SSE/UI path). + const delivered = yield* Deferred.await(received).pipe( + Effect.timeoutOrElse({ duration: "2 seconds", orElse: () => Effect.succeed("none" as const) }), + ) + expect(delivered).toContain("message.updated") + yield* unsub + }), + ) +}) + +describe("experimentalWorkspaces ON full event sourcing", () => { + itWithWorkspaces.instance("persists gated events to the log", () => + Effect.gen(function* () { + const session = yield* SessionNs.Service + const { db } = yield* Database.Service + const created = yield* session.create({ title: "gate-on" }) + yield* session.updateMessage({ + id: MessageID.ascending(), + sessionID: created.id, + role: "user", + agent: "build", + model: { providerID: "test", modelID: "test" }, + time: { created: Date.now() }, + tools: {}, + mode: "", + } as unknown as SessionV1.Info) + + const snapshots = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.type, "message.updated.1")) + .all() + .pipe(Effect.orDie) + const seq = yield* db + .select() + .from(EventSequenceTable) + .where(eq(EventSequenceTable.aggregate_id, created.id)) + .all() + .pipe(Effect.orDie) + expect(snapshots).toHaveLength(1) + expect(seq).toHaveLength(1) + expect(seq[0]?.seq).toBe(1) + }), + ) +}) \ No newline at end of file diff --git a/packages/opencode/test/session/prompt.test.ts b/packages/opencode/test/session/prompt.test.ts index da6e0f8d036f..97a051340c84 100644 --- a/packages/opencode/test/session/prompt.test.ts +++ b/packages/opencode/test/session/prompt.test.ts @@ -1,6 +1,8 @@ import { ConfigV1 } from "@opencode-ai/core/v1/config/config" import { SessionV1 } from "@opencode-ai/core/v1/session" import { Database } from "@opencode-ai/core/database/database" +import { EventTable } from "@opencode-ai/core/event/sql" +import { MessageTable } from "@opencode-ai/core/session/sql" import { LayerNode } from "@opencode-ai/core/effect/layer-node" import { SessionProjector } from "@opencode-ai/core/session/projector" import { eq } from "drizzle-orm" @@ -2468,3 +2470,49 @@ noLLMServer.instance( }), 30_000, ) + +it.instance("full prompt loop writes projections but no durable snapshot events (gate OFF)", () => + Effect.gen(function* () { + const { llm } = yield* useServerConfig(providerCfg) + const prompt = yield* SessionPrompt.Service + const sessions = yield* Session.Service + const { db } = yield* Database.Service + const chat = yield* sessions.create({ + title: "Pinned", + permission: [{ permission: "*", pattern: "*", action: "allow" }], + }) + + yield* prompt.prompt({ + sessionID: chat.id, + agent: "build", + noReply: true, + parts: [{ type: "text", text: "hello" }], + }) + yield* llm.text("world") + yield* prompt.loop({ sessionID: chat.id }) + + const messageRows = yield* db + .select() + .from(MessageTable) + .where(eq(MessageTable.session_id, chat.id)) + .all() + .pipe(Effect.orDie) + const snapshots = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.type, "message.updated.1")) + .all() + .pipe(Effect.orDie) + const partSnapshots = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.type, "message.part.updated.1")) + .all() + .pipe(Effect.orDie) + + expect(messageRows.length).toBeGreaterThan(0) + expect(snapshots).toHaveLength(0) + expect(partSnapshots).toHaveLength(0) + }), + 60_000, +)