From f7d776df88db1d1e44439ed8c22ad3c2a4e90d09 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Mon, 13 Jul 2026 09:54:36 -0400 Subject: [PATCH 1/3] fix(core): compact superseded event snapshots --- .../core/src/session/event-log-compaction.ts | 130 ++++++++++++++++++ .../test/session-event-log-compaction.test.ts | 88 ++++++++++++ packages/opencode/src/cli/cmd/db.ts | 23 +++- packages/schema/src/durable-event-manifest.ts | 1 + packages/schema/src/event.ts | 15 ++ 5 files changed, 256 insertions(+), 1 deletion(-) create mode 100644 packages/core/src/session/event-log-compaction.ts create mode 100644 packages/core/test/session-event-log-compaction.test.ts diff --git a/packages/core/src/session/event-log-compaction.ts b/packages/core/src/session/event-log-compaction.ts new file mode 100644 index 000000000000..14d58275916c --- /dev/null +++ b/packages/core/src/session/event-log-compaction.ts @@ -0,0 +1,130 @@ +export * as SessionEventLogCompaction from "./event-log-compaction" + +import { sql } from "drizzle-orm" +import { Effect } from "effect" +import { Event } from "@opencode-ai/schema/event" +import type { Database } from "../database/database" + +const checkpointType = Event.versionedType(Event.Compacted.type, Event.Compacted.durable!.version) + +type Candidate = { + id: string + aggregateID: string + type: string + bytes: number + supersededBy: string +} + +const checkpoint = (candidate: Candidate) => ({ + aggregateID: candidate.aggregateID, + supersededType: candidate.type, + supersededBy: candidate.supersededBy, +}) + +const reclaimedBytes = (candidate: Candidate) => + Math.max(0, candidate.bytes - Buffer.byteLength(JSON.stringify(checkpoint(candidate)))) + +export type Report = { + readonly dryRun: boolean + readonly candidates: number + readonly rewritten: number + readonly payloadBytesReclaimed: number + readonly aggregates: number + readonly byType: Readonly> +} + +export interface Options { + readonly aggregateID?: string + readonly dryRun?: boolean + readonly limit?: number +} + +const policies = [ + { type: "message.updated.1", path: "$.info.id", table: "message" }, + { type: "message.part.updated.1", path: "$.part.id", table: "part" }, +] as const + +/** + * Reclaims full snapshots only after a later snapshot for the same projected + * entity exists. The replacement marker is durable and sequence-preserving, + * so peers can replay the compacted aggregate with the normal strict protocol. + */ +export const compact = Effect.fn("SessionEventLogCompaction.compact")(function* ( + db: Database.Interface["db"], + options: Options = {}, +) { + const limit = options.limit ?? Number.MAX_SAFE_INTEGER + const candidates = new Array() + + for (const policy of policies) { + if (candidates.length >= limit) break + const rows = yield* db + .all( + sql` + WITH latest AS ( + SELECT aggregate_id, json_extract(data, ${policy.path}) AS entity_id, MAX(seq) AS seq + FROM event + WHERE type = ${policy.type} + AND json_valid(data) + ${options.aggregateID === undefined ? sql`` : sql`AND aggregate_id = ${options.aggregateID}`} + GROUP BY aggregate_id, json_extract(data, ${policy.path}) + ) + SELECT event.id, + event.aggregate_id AS aggregateID, + event.type, + length(event.data) AS bytes, + replacement.id AS supersededBy + FROM event + JOIN latest + ON latest.aggregate_id = event.aggregate_id + AND latest.entity_id = json_extract(event.data, ${policy.path}) + JOIN event AS replacement + ON replacement.aggregate_id = latest.aggregate_id + AND replacement.type = ${policy.type} + AND replacement.seq = latest.seq + JOIN ${sql.identifier(policy.table)} AS projection + ON projection.id = latest.entity_id + AND projection.session_id = latest.aggregate_id + WHERE event.type = ${policy.type} + AND event.seq < latest.seq + ORDER BY event.aggregate_id, event.seq + LIMIT ${limit - candidates.length} + `, + ) + .pipe(Effect.orDie) + candidates.push(...rows) + } + + const byType: Record = {} + for (const candidate of candidates) { + const summary = byType[candidate.type] ?? { events: 0, payloadBytesReclaimed: 0 } + summary.events++ + summary.payloadBytesReclaimed += reclaimedBytes(candidate) + byType[candidate.type] = summary + } + const report = (): Report => ({ + dryRun: options.dryRun ?? false, + candidates: candidates.length, + rewritten: options.dryRun ? 0 : candidates.length, + payloadBytesReclaimed: candidates.reduce((total, candidate) => total + reclaimedBytes(candidate), 0), + aggregates: new Set(candidates.map((candidate) => candidate.aggregateID)).size, + byType, + }) + if (options.dryRun || candidates.length === 0) return report() + + yield* db + .transaction( + () => + Effect.forEach(candidates, (candidate) => + db.run(sql` + UPDATE event + SET type = ${checkpointType}, + data = ${JSON.stringify(checkpoint(candidate))} + WHERE id = ${candidate.id} AND type = ${candidate.type} + `), + ), + { behavior: "immediate" }, + ) + .pipe(Effect.orDie) + return report() +}) diff --git a/packages/core/test/session-event-log-compaction.test.ts b/packages/core/test/session-event-log-compaction.test.ts new file mode 100644 index 000000000000..9aecfddf6a79 --- /dev/null +++ b/packages/core/test/session-event-log-compaction.test.ts @@ -0,0 +1,88 @@ +import { describe, expect } from "bun:test" +import { Effect } from "effect" +import { asc, eq } from "drizzle-orm" +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 { EventV2 } from "@opencode-ai/core/event" +import { EventTable } from "@opencode-ai/core/event/sql" +import { Project } from "@opencode-ai/core/project" +import { ProjectTable } from "@opencode-ai/core/project/sql" +import { ModelV2 } from "@opencode-ai/core/model" +import { ProviderV2 } from "@opencode-ai/core/provider" +import { SessionEventLogCompaction } from "@opencode-ai/core/session/event-log-compaction" +import { SessionProjector } from "@opencode-ai/core/session/projector" +import { MessageTable, SessionTable } from "@opencode-ai/core/session/sql" +import { SessionV1 } from "@opencode-ai/schema/session-v1" +import { SessionID } from "@opencode-ai/schema/session-id" +import { AbsolutePath } from "@opencode-ai/core/schema" +import { testEffect } from "./lib/effect" + +const it = testEffect(AppNodeBuilder.build(LayerNode.group([Database.node, EventV2.node, SessionProjector.node]))) + +describe("SessionEventLogCompaction", () => { + it.effect("keeps replay and the message projection identical while reclaiming superseded snapshots", () => + Effect.gen(function* () { + const { db } = yield* Database.Service + const events = yield* EventV2.Service + const sessionID = SessionID.descending("ses_event_log_compaction") + const messageID = SessionV1.MessageID.ascending("msg_event_log_compaction") + yield* db + .insert(ProjectTable) + .values({ id: Project.ID.global, worktree: AbsolutePath.make("/project"), sandboxes: [] }) + .run() + yield* db + .insert(SessionTable) + .values({ + id: sessionID, + project_id: Project.ID.global, + slug: "compaction", + directory: "/project", + title: "compaction", + version: "test", + }) + .run() + const message = (agent: string) => ({ + id: messageID, + sessionID, + role: "user" as const, + time: { created: 1 }, + agent, + model: { providerID: ProviderV2.ID.make("provider"), modelID: ModelV2.ID.make("model") }, + }) + + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: message("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: message("after") }) + const projection = yield* db.select().from(MessageTable).where(eq(MessageTable.id, messageID)).get() + const dryRun = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, dryRun: true }) + + expect(dryRun).toMatchObject({ candidates: 1, rewritten: 0, payloadBytesReclaimed: expect.any(Number) }) + expect(projection?.data).toMatchObject({ agent: "after" }) + + const applied = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID }) + const compacted = yield* db + .select() + .from(EventTable) + .where(eq(EventTable.aggregate_id, sessionID)) + .orderBy(asc(EventTable.seq)) + .all() + + expect(applied).toMatchObject({ candidates: 1, rewritten: 1 }) + expect(compacted.map((event) => event.seq)).toEqual([0, 1]) + expect(compacted.map((event) => event.type)).toEqual(["event.compacted.1", "message.updated.1"]) + expect(yield* db.select().from(MessageTable).where(eq(MessageTable.id, messageID)).get()).toEqual(projection) + + expect( + yield* events.replayAll( + compacted.map((event) => ({ + id: event.id, + aggregateID: event.aggregate_id, + seq: event.seq, + type: event.type, + data: event.data, + })), + ), + ).toBe(sessionID) + }), + ) +}) diff --git a/packages/opencode/src/cli/cmd/db.ts b/packages/opencode/src/cli/cmd/db.ts index 9e7e37e18e91..9056e3e798ca 100644 --- a/packages/opencode/src/cli/cmd/db.ts +++ b/packages/opencode/src/cli/cmd/db.ts @@ -1,6 +1,7 @@ import type { Argv } from "yargs" import { spawn } from "child_process" import { Database } from "@opencode-ai/core/database/database" +import { SessionEventLogCompaction } from "@opencode-ai/core/session/event-log-compaction" import { Effect } from "effect" import { sql } from "drizzle-orm" import { effectCmd } from "../effect-cmd" @@ -51,12 +52,32 @@ const PathCommand = effectCmd({ }), }) +const CompactEventsCommand = effectCmd({ + command: "compact-events", + describe: "replace superseded message and part snapshots with replay-safe checkpoints", + instance: false, + builder: (yargs: Argv) => + yargs + .option("apply", { type: "boolean", default: false, describe: "write checkpoints; default is dry-run" }) + .option("session", { type: "string", describe: "compact one session aggregate" }) + .option("limit", { type: "number", describe: "maximum snapshots to compact" }), + handler: Effect.fn("Cli.db.compactEvents")(function* (args: { apply: boolean; session?: string; limit?: number }) { + const { db } = yield* Database.Service + const report = yield* SessionEventLogCompaction.compact(db, { + aggregateID: args.session, + dryRun: !args.apply, + limit: args.limit, + }) + console.log(JSON.stringify(report, null, 2)) + }), +}) + 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(CompactEventsCommand).demandCommand() }, handler: Effect.fn("Cli.db")(function* () {}), }) diff --git a/packages/schema/src/durable-event-manifest.ts b/packages/schema/src/durable-event-manifest.ts index acdcb3e9d568..a6f0822a3db3 100644 --- a/packages/schema/src/durable-event-manifest.ts +++ b/packages/schema/src/durable-event-manifest.ts @@ -10,6 +10,7 @@ export const SessionDurable = { } as const export const Durable = Event.durable([ + Event.Compacted, ...SessionV1.Event.Definitions.filter((definition) => definition.durable !== undefined), ...SessionEvent.DurableDefinitions, ]) diff --git a/packages/schema/src/event.ts b/packages/schema/src/event.ts index 0d6ec9775aa4..94078269c64c 100644 --- a/packages/schema/src/event.ts +++ b/packages/schema/src/event.ts @@ -12,6 +12,21 @@ export const ID = Schema.String.check(Schema.isStartsWith("evt_")).pipe( ) export type ID = typeof ID.Type +/** + * Replaces a durable snapshot that is represented by a later full snapshot. + * Keeping this small marker preserves the aggregate's contiguous sequence for + * strict sync replay while allowing the superseded payload to be reclaimed. + */ +export const Compacted = define({ + type: "event.compacted", + durable: { version: 1, aggregate: "aggregateID" }, + schema: { + aggregateID: Schema.String, + supersededType: Schema.String, + supersededBy: Schema.String, + }, +}) + export type Definition< Type extends string = string, DataSchema extends Schema.Codec = Schema.Codec, From 86d8e4992eea43eed98bf959535181a9f0f3d692 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Mon, 13 Jul 2026 13:51:25 -0400 Subject: [PATCH 2/3] fix(core): bound event log compaction --- .../core/src/session/event-log-compaction.ts | 258 +++++++++++++----- .../test/session-event-log-compaction.test.ts | 62 ++++- packages/opencode/src/cli/cmd/db.ts | 34 ++- 3 files changed, 270 insertions(+), 84 deletions(-) diff --git a/packages/core/src/session/event-log-compaction.ts b/packages/core/src/session/event-log-compaction.ts index 14d58275916c..370b6f03dc34 100644 --- a/packages/core/src/session/event-log-compaction.ts +++ b/packages/core/src/session/event-log-compaction.ts @@ -1,130 +1,238 @@ export * as SessionEventLogCompaction from "./event-log-compaction" +import { isDeepStrictEqual } from "node:util" import { sql } from "drizzle-orm" import { Effect } from "effect" import { Event } from "@opencode-ai/schema/event" import type { Database } from "../database/database" +const DEFAULT_LIMIT = 1000 +const MAX_LIMIT = 10_000 const checkpointType = Event.versionedType(Event.Compacted.type, Event.Compacted.durable!.version) +type Policy = (typeof policies)[number] type Candidate = { id: string aggregateID: string type: string bytes: number supersededBy: string + latestData: string + projectionData: string + workspaceID: string | null + ownerID: string | null } -const checkpoint = (candidate: Candidate) => ({ - aggregateID: candidate.aggregateID, - supersededType: candidate.type, - supersededBy: candidate.supersededBy, -}) - -const reclaimedBytes = (candidate: Candidate) => - Math.max(0, candidate.bytes - Buffer.byteLength(JSON.stringify(checkpoint(candidate)))) +export type Options = { + readonly aggregateID?: string + readonly all?: boolean + readonly apply?: boolean + readonly limit?: number +} export type Report = { readonly dryRun: boolean readonly candidates: number readonly rewritten: number + readonly projectionMismatches: number + readonly compatibilityRejected: number + readonly malformed: number readonly payloadBytesReclaimed: number - readonly aggregates: number + readonly hasMore: boolean + readonly continuation: string readonly byType: Readonly> } -export interface Options { - readonly aggregateID?: string - readonly dryRun?: boolean - readonly limit?: number +export type Status = { + readonly events: number + readonly payloadBytes: number + readonly compactableEvents: number + readonly recommended: boolean } const policies = [ - { type: "message.updated.1", path: "$.info.id", table: "message" }, - { type: "message.part.updated.1", path: "$.part.id", table: "part" }, + { type: "message.updated.1", path: "$.info.id", table: "message", fields: ["id", "sessionID"] }, + { type: "message.part.updated.1", path: "$.part.id", table: "part", fields: ["id", "messageID", "sessionID"] }, ] as const -/** - * Reclaims full snapshots only after a later snapshot for the same projected - * entity exists. The replacement marker is durable and sequence-preserving, - * so peers can replay the compacted aggregate with the normal strict protocol. - */ -export const compact = Effect.fn("SessionEventLogCompaction.compact")(function* ( +function validate(options: Options) { + if (Boolean(options.aggregateID) === Boolean(options.all)) { + throw new Error("Specify exactly one compaction scope: aggregateID or all") + } + const limit = options.limit ?? DEFAULT_LIMIT + if (!Number.isSafeInteger(limit) || limit <= 0 || limit > MAX_LIMIT) { + throw new Error(`limit must be a positive integer no greater than ${MAX_LIMIT}`) + } + return limit +} + +function continuation(options: Options, limit: number) { + const scope = options.all ? "--all" : `--session ${options.aggregateID}` + return `opencode db compact-events ${scope} --apply --limit ${limit}` +} + +function checkpoint(candidate: Candidate) { + return { aggregateID: candidate.aggregateID, supersededType: candidate.type, supersededBy: candidate.supersededBy } +} + +function reclaimedBytes(candidate: Candidate) { + return Math.max(0, candidate.bytes - Buffer.byteLength(JSON.stringify(checkpoint(candidate)))) +} + +function projectedData(candidate: Candidate, policy: Policy) { + const latest = JSON.parse(candidate.latestData) as { info?: Record; part?: Record } + const value = policy.type === "message.updated.1" ? latest.info : latest.part + if (!value) return undefined + const result = { ...value } + for (const field of policy.fields) delete result[field] + return result +} + +function safe(candidate: Candidate, policy: Policy) { + try { + return isDeepStrictEqual(projectedData(candidate, policy), JSON.parse(candidate.projectionData)) + } catch { + return false + } +} + +const candidates = Effect.fn("SessionEventLogCompaction.candidates")(function* ( db: Database.Interface["db"], - options: Options = {}, + options: Options, + limit: number, ) { - const limit = options.limit ?? Number.MAX_SAFE_INTEGER - const candidates = new Array() - + const rows = new Array() for (const policy of policies) { - if (candidates.length >= limit) break - const rows = yield* db + if (rows.length >= limit + 1) break + const selected = yield* db .all( sql` - WITH latest AS ( - SELECT aggregate_id, json_extract(data, ${policy.path}) AS entity_id, MAX(seq) AS seq + WITH latest AS ( + SELECT aggregate_id, json_extract(data, ${policy.path}) AS entity_id, MAX(seq) AS seq + FROM event + WHERE type = ${policy.type} + AND json_valid(data) + ${options.all ? sql`` : sql`AND aggregate_id = ${options.aggregateID!}`} + GROUP BY aggregate_id, json_extract(data, ${policy.path}) + ) + SELECT event.id, event.aggregate_id AS aggregateID, event.type, length(event.data) AS bytes, + replacement.id AS supersededBy, replacement.data AS latestData, + projection.data AS projectionData, session.workspace_id AS workspaceID, + event_sequence.owner_id AS ownerID FROM event - WHERE type = ${policy.type} - AND json_valid(data) - ${options.aggregateID === undefined ? sql`` : sql`AND aggregate_id = ${options.aggregateID}`} - GROUP BY aggregate_id, json_extract(data, ${policy.path}) - ) - SELECT event.id, - event.aggregate_id AS aggregateID, - event.type, - length(event.data) AS bytes, - replacement.id AS supersededBy - FROM event - JOIN latest - ON latest.aggregate_id = event.aggregate_id - AND latest.entity_id = json_extract(event.data, ${policy.path}) - JOIN event AS replacement - ON replacement.aggregate_id = latest.aggregate_id - AND replacement.type = ${policy.type} - AND replacement.seq = latest.seq - JOIN ${sql.identifier(policy.table)} AS projection - ON projection.id = latest.entity_id - AND projection.session_id = latest.aggregate_id - WHERE event.type = ${policy.type} - AND event.seq < latest.seq - ORDER BY event.aggregate_id, event.seq - LIMIT ${limit - candidates.length} - `, + JOIN latest ON latest.aggregate_id = event.aggregate_id + AND latest.entity_id = json_extract(CASE WHEN json_valid(event.data) THEN event.data END, ${policy.path}) + JOIN event AS replacement ON replacement.aggregate_id = latest.aggregate_id + AND replacement.type = ${policy.type} AND replacement.seq = latest.seq + JOIN ${sql.identifier(policy.table)} AS projection ON projection.id = latest.entity_id + AND projection.session_id = latest.aggregate_id + JOIN session ON session.id = event.aggregate_id + JOIN event_sequence ON event_sequence.aggregate_id = event.aggregate_id + WHERE event.type = ${policy.type} AND json_valid(event.data) AND event.seq < latest.seq + ORDER BY event.aggregate_id, event.seq + LIMIT ${limit + 1 - rows.length} + `, ) .pipe(Effect.orDie) - candidates.push(...rows) + rows.push(...selected.map((candidate) => ({ ...candidate, policy }))) } + return rows +}) +const malformed = Effect.fn("SessionEventLogCompaction.malformed")(function* ( + db: Database.Interface["db"], + options: Options, +) { + const row = yield* db + .get<{ count: number }>( + sql` + SELECT count(*) AS count FROM event + WHERE type IN (${policies[0].type}, ${policies[1].type}) AND NOT json_valid(data) + ${options.all ? sql`` : sql`AND aggregate_id = ${options.aggregateID!}`} + `, + ) + .pipe(Effect.orDie) + return row?.count ?? 0 +}) + +function report( + rows: ReadonlyArray, + malformedCount: number, + options: Options, + limit: number, +) { + const inspected = rows.slice(0, limit) + const safeRows = inspected.filter((candidate) => safe(candidate, candidate.policy)) + const eligible = safeRows.filter((candidate) => candidate.workspaceID === null && candidate.ownerID === null) const byType: Record = {} - for (const candidate of candidates) { + for (const candidate of eligible) { const summary = byType[candidate.type] ?? { events: 0, payloadBytesReclaimed: 0 } summary.events++ summary.payloadBytesReclaimed += reclaimedBytes(candidate) byType[candidate.type] = summary } - const report = (): Report => ({ - dryRun: options.dryRun ?? false, - candidates: candidates.length, - rewritten: options.dryRun ? 0 : candidates.length, - payloadBytesReclaimed: candidates.reduce((total, candidate) => total + reclaimedBytes(candidate), 0), - aggregates: new Set(candidates.map((candidate) => candidate.aggregateID)).size, + return { + dryRun: !options.apply, + candidates: eligible.length, + rewritten: options.apply ? eligible.length : 0, + projectionMismatches: inspected.filter((candidate) => !safe(candidate, candidate.policy)).length, + compatibilityRejected: safeRows.length - eligible.length, + malformed: malformedCount, + payloadBytesReclaimed: eligible.reduce((total, candidate) => total + reclaimedBytes(candidate), 0), + hasMore: rows.length > limit, + continuation: continuation(options, limit), byType, - }) - if (options.dryRun || candidates.length === 0) return report() + } satisfies Report +} - yield* db +/** + * Replaces only locally-owned, projection-verified snapshots. Sync has no + * version negotiation for checkpoint markers, so aggregates with a workspace + * or sync owner are always dry-run-only until that protocol exists. + */ +export const compact = Effect.fn("SessionEventLogCompaction.compact")(function* ( + db: Database.Interface["db"], + options: Options, +) { + const limit = validate(options) + if (!options.apply) + return report(yield* candidates(db, options, limit), yield* malformed(db, options), options, limit) + return yield* db .transaction( () => - Effect.forEach(candidates, (candidate) => - db.run(sql` - UPDATE event - SET type = ${checkpointType}, - data = ${JSON.stringify(checkpoint(candidate))} - WHERE id = ${candidate.id} AND type = ${candidate.type} - `), - ), + Effect.gen(function* () { + const selected = yield* candidates(db, options, limit) + const result = report(selected, yield* malformed(db, options), options, limit) + const eligible = selected + .slice(0, limit) + .filter( + (candidate) => + safe(candidate, candidate.policy) && candidate.workspaceID === null && candidate.ownerID === null, + ) + yield* Effect.forEach(eligible, (candidate) => + db.run(sql` + UPDATE event SET type = ${checkpointType}, data = ${JSON.stringify(checkpoint(candidate))} + WHERE id = ${candidate.id} AND type = ${candidate.type} + `), + ) + return result + }), { behavior: "immediate" }, ) .pipe(Effect.orDie) - return report() +}) + +export const status = Effect.fn("SessionEventLogCompaction.status")(function* (db: Database.Interface["db"]) { + const row = yield* db + .get<{ events: number; payloadBytes: number; compactableEvents: number }>( + sql` + SELECT count(*) AS events, + coalesce(sum(length(data)), 0) AS payloadBytes, + coalesce(sum(type IN (${policies[0].type}, ${policies[1].type})), 0) AS compactableEvents + FROM event + `, + ) + .pipe(Effect.orDie) + const result = row ?? { events: 0, payloadBytes: 0, compactableEvents: 0 } + return { ...result, recommended: result.compactableEvents > DEFAULT_LIMIT } satisfies Status }) diff --git a/packages/core/test/session-event-log-compaction.test.ts b/packages/core/test/session-event-log-compaction.test.ts index 9aecfddf6a79..c3446ea4a9ee 100644 --- a/packages/core/test/session-event-log-compaction.test.ts +++ b/packages/core/test/session-event-log-compaction.test.ts @@ -1,6 +1,6 @@ import { describe, expect } from "bun:test" import { Effect } from "effect" -import { asc, eq } from "drizzle-orm" +import { asc, eq, sql } from "drizzle-orm" 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" @@ -54,12 +54,12 @@ describe("SessionEventLogCompaction", () => { yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: message("before") }) yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: message("after") }) const projection = yield* db.select().from(MessageTable).where(eq(MessageTable.id, messageID)).get() - const dryRun = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, dryRun: true }) + const dryRun = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID }) expect(dryRun).toMatchObject({ candidates: 1, rewritten: 0, payloadBytesReclaimed: expect.any(Number) }) expect(projection?.data).toMatchObject({ agent: "after" }) - const applied = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID }) + const applied = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, apply: true }) const compacted = yield* db .select() .from(EventTable) @@ -83,6 +83,62 @@ describe("SessionEventLogCompaction", () => { })), ), ).toBe(sessionID) + + const partID = SessionV1.PartID.ascending("prt_event_log_compaction") + const part = (text: string) => ({ id: partID, sessionID, messageID, type: "text" as const, text }) + yield* events.publish(SessionV1.Event.PartUpdated, { sessionID, part: part("one"), time: 1 }) + yield* events.publish(SessionV1.Event.PartUpdated, { sessionID, part: part("two"), time: 2 }) + yield* events.publish(SessionV1.Event.PartUpdated, { sessionID, part: part("three"), time: 3 }) + + const firstPartBatch = yield* SessionEventLogCompaction.compact(db, { + aggregateID: sessionID, + apply: true, + limit: 1, + }) + const secondPartBatch = yield* SessionEventLogCompaction.compact(db, { + aggregateID: sessionID, + apply: true, + limit: 1, + }) + const idempotent = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, apply: true }) + expect(firstPartBatch).toMatchObject({ candidates: 1, rewritten: 1, hasMore: true }) + expect(firstPartBatch.continuation).toBe(`opencode db compact-events --session ${sessionID} --apply --limit 1`) + expect(secondPartBatch).toMatchObject({ candidates: 1, rewritten: 1 }) + expect(idempotent).toMatchObject({ candidates: 0, rewritten: 0 }) + + const mismatchID = SessionV1.MessageID.ascending("msg_event_log_mismatch") + const mismatch = (agent: string) => ({ ...message(agent), id: mismatchID }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: mismatch("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: mismatch("after") }) + yield* db.run(sql`UPDATE message SET data = ${JSON.stringify({ agent: "stale" })} WHERE id = ${mismatchID}`) + expect(yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID })).toMatchObject({ + projectionMismatches: 1, + }) + + const malformedID = SessionV1.MessageID.ascending("msg_event_log_malformed") + const malformed = (agent: string) => ({ ...message(agent), id: malformedID }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: malformed("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: malformed("after") }) + yield* db.run(sql`UPDATE event SET data = 'not json' WHERE type = 'message.updated.1' AND seq = 6`) + expect(yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID })).toMatchObject({ malformed: 1 }) + + const ownedID = SessionV1.MessageID.ascending("msg_event_log_owned") + const owned = (agent: string) => ({ ...message(agent), id: ownedID }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: owned("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: owned("after") }) + yield* events.claim(sessionID, "sync-owner") + const rejected = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, apply: true }) + expect(rejected).toMatchObject({ candidates: 0, rewritten: 0, compatibilityRejected: expect.any(Number) }) + + for (const options of [ + {}, + { aggregateID: sessionID, all: true }, + { aggregateID: sessionID, limit: 0 }, + { aggregateID: sessionID, limit: 10_001 }, + ]) { + const exit = yield* SessionEventLogCompaction.compact(db, options).pipe(Effect.exit) + expect(String(exit)).toContain("Error") + } }), ) }) diff --git a/packages/opencode/src/cli/cmd/db.ts b/packages/opencode/src/cli/cmd/db.ts index 9056e3e798ca..ec5acd801a43 100644 --- a/packages/opencode/src/cli/cmd/db.ts +++ b/packages/opencode/src/cli/cmd/db.ts @@ -4,7 +4,7 @@ import { Database } from "@opencode-ai/core/database/database" import { SessionEventLogCompaction } from "@opencode-ai/core/session/event-log-compaction" import { Effect } from "effect" import { sql } from "drizzle-orm" -import { effectCmd } from "../effect-cmd" +import { effectCmd, fail } from "../effect-cmd" const QueryCommand = effectCmd({ command: "$0 [query]", @@ -60,24 +60,46 @@ const CompactEventsCommand = effectCmd({ yargs .option("apply", { type: "boolean", default: false, describe: "write checkpoints; default is dry-run" }) .option("session", { type: "string", describe: "compact one session aggregate" }) - .option("limit", { type: "number", describe: "maximum snapshots to compact" }), - handler: Effect.fn("Cli.db.compactEvents")(function* (args: { apply: boolean; session?: string; limit?: number }) { + .option("all", { type: "boolean", default: false, describe: "inspect all session aggregates" }) + .option("limit", { type: "number", describe: "maximum snapshots per bounded batch" }), + handler: Effect.fn("Cli.db.compactEvents")(function* (args: { + apply: boolean + session?: string + all: boolean + limit?: number + }) { const { db } = yield* Database.Service const report = yield* SessionEventLogCompaction.compact(db, { aggregateID: args.session, - dryRun: !args.apply, + all: args.all, + apply: args.apply, limit: args.limit, - }) + }).pipe(Effect.catchDefect((error) => fail(error instanceof Error ? error.message : String(error)))) console.log(JSON.stringify(report, null, 2)) }), }) +const EventLogStatusCommand = effectCmd({ + command: "event-log-status", + describe: "report event-log growth and compaction recommendation", + instance: false, + handler: Effect.fn("Cli.db.eventLogStatus")(function* () { + const { db } = yield* Database.Service + console.log(JSON.stringify(yield* SessionEventLogCompaction.status(db), null, 2)) + }), +}) + export const DbCommand = effectCmd({ command: "db", describe: "database tools", instance: false, builder: (yargs: Argv) => { - return yargs.command(QueryCommand).command(PathCommand).command(CompactEventsCommand).demandCommand() + return yargs + .command(QueryCommand) + .command(PathCommand) + .command(CompactEventsCommand) + .command(EventLogStatusCommand) + .demandCommand() }, handler: Effect.fn("Cli.db")(function* () {}), }) From 04954f8bde41730251c7516f45d34b975b1d45f5 Mon Sep 17 00:00:00 2001 From: Chris Huber Date: Mon, 13 Jul 2026 13:55:08 -0400 Subject: [PATCH 3/3] test(core): cover compaction append safety --- .../core/test/session-event-log-compaction.test.ts | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/packages/core/test/session-event-log-compaction.test.ts b/packages/core/test/session-event-log-compaction.test.ts index c3446ea4a9ee..24a99c2c904a 100644 --- a/packages/core/test/session-event-log-compaction.test.ts +++ b/packages/core/test/session-event-log-compaction.test.ts @@ -54,6 +54,7 @@ describe("SessionEventLogCompaction", () => { yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: message("before") }) yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: message("after") }) const projection = yield* db.select().from(MessageTable).where(eq(MessageTable.id, messageID)).get() + expect(yield* SessionEventLogCompaction.status(db)).toMatchObject({ events: 2, compactableEvents: 2 }) const dryRun = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID }) expect(dryRun).toMatchObject({ candidates: 1, rewritten: 0, payloadBytesReclaimed: expect.any(Number) }) @@ -122,6 +123,17 @@ describe("SessionEventLogCompaction", () => { yield* db.run(sql`UPDATE event SET data = 'not json' WHERE type = 'message.updated.1' AND seq = 6`) expect(yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID })).toMatchObject({ malformed: 1 }) + const appendedID = SessionV1.MessageID.ascending("msg_event_log_appended") + const appended = (agent: string) => ({ ...message(agent), id: appendedID }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: appended("before") }) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: appended("middle") }) + expect((yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID })).candidates).toBeGreaterThan(0) + yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: appended("latest") }) + yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, apply: true }) + expect((yield* db.select().from(MessageTable).where(eq(MessageTable.id, appendedID)).get())?.data).toMatchObject({ + agent: "latest", + }) + const ownedID = SessionV1.MessageID.ascending("msg_event_log_owned") const owned = (agent: string) => ({ ...message(agent), id: ownedID }) yield* events.publish(SessionV1.Event.MessageUpdated, { sessionID, info: owned("before") })