fix(core): bound event log compaction

packages/core/src/session/event-log-compaction.ts

@@ -0,0 +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
+}
+
+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 hasMore: boolean
+  readonly continuation: string
+  readonly byType: Readonly<Record<string, { readonly events: number; readonly payloadBytesReclaimed: 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", fields: ["id", "sessionID"] },
+  { type: "message.part.updated.1", path: "$.part.id", table: "part", fields: ["id", "messageID", "sessionID"] },
+] as const
+
+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<string, unknown>; part?: Record<string, unknown> }
+  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,
+  limit: number,
+) {
+  const rows = new Array<Candidate & { policy: Policy }>()
+  for (const policy of policies) {
+    if (rows.length >= limit + 1) break
+    const selected = yield* db
+      .all<Candidate>(
+        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.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
+        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)
+    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<Candidate & { policy: Policy }>,
+  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<string, { events: number; payloadBytesReclaimed: number }> = {}
+  for (const candidate of eligible) {
+    const summary = byType[candidate.type] ?? { events: 0, payloadBytesReclaimed: 0 }
+    summary.events++
+    summary.payloadBytesReclaimed += reclaimedBytes(candidate)
+    byType[candidate.type] = summary
+  }
+  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,
+  } satisfies Report
+}
+
+/**
+ * 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.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)
+})
+
+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
+})

packages/core/test/session-event-log-compaction.test.ts

@@ -0,0 +1,156 @@
+import { describe, expect } from "bun:test"
+import { Effect } from "effect"
+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"
+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()
+      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) })
+      expect(projection?.data).toMatchObject({ agent: "after" })
+
+      const applied = yield* SessionEventLogCompaction.compact(db, { aggregateID: sessionID, apply: true })
+      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)
+
+      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 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") })
+      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")
+      }
+    }),
+  )
+})