From 05efb69ac8bf88d2dcb673848d44257332d9d388 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:10:24 -0800 Subject: [PATCH 1/5] perf(server): prune events the projections already cover Projections are the source of truth once they apply an event, so the event log only keeps what live readers need. Once a day the background pass deletes events through the head it recorded a day earlier (and no further than the projection watermark), keeping each stream's newest event, thread.created, the first running update of each run attempt, and prompt anchors that retained rows reference. Thread and shell resumes from a cursor below the recorded pruned-through sequence load a snapshot instead of a partial replay. verify/rebuild move to a test-only replay oracle; t3 doctor no longer refuses to compact a database whose log cannot rebuild its projections. Co-Authored-By: Claude Opus 5.5 --- .../FoundationPersistence.test.ts | 144 +---- .../ProjectionMaintenance.test.ts | 355 +++++++++++ .../orchestration-v2/ProjectionMaintenance.ts | 558 +++++++----------- .../TurnItemCompaction.test.ts | 488 --------------- .../LegacyV1Cutover.integration.test.ts | 8 +- .../legacy/LegacyV1ThreadImporter.test.ts | 10 +- .../src/orchestration-v2/runtimeLayer.test.ts | 116 +--- .../testkit/ProjectionReplay.ts | 182 ++++++ .../ProviderSwitch.integration.test.ts | 6 +- .../persistence/DatabaseMaintenance.test.ts | 16 +- .../src/persistence/DatabaseMaintenance.ts | 29 +- .../src/persistence/DatabaseStorage.test.ts | 9 +- .../server/src/persistence/DatabaseStorage.ts | 3 +- ...CanonicalizeThreadPullRequestLinks.test.ts | 6 +- ...6_RepairClientSettlementTimestamps.test.ts | 6 +- .../persistence/OrchestrationEventStore.ts | 19 +- .../project/ProjectService.deletion.test.ts | 6 +- apps/server/src/server.ts | 8 +- apps/server/src/ws.test.ts | 137 ++--- apps/server/src/ws.ts | 13 +- docs/internals/overview.md | 16 +- docs/user/background-service.md | 4 +- 22 files changed, 943 insertions(+), 1196 deletions(-) create mode 100644 apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts delete mode 100644 apps/server/src/orchestration-v2/TurnItemCompaction.test.ts create mode 100644 apps/server/src/orchestration-v2/testkit/ProjectionReplay.ts diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index c5c51646584aa..475e0a0e4ae7b 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -62,6 +62,7 @@ import * as EventSink from "./EventSink.ts"; import * as EventStore from "./EventStore.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ProjectionMaintenance from "./ProjectionMaintenance.ts"; +import * as ProjectionReplay from "./testkit/ProjectionReplay.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProjectStore from "./ProjectStore.ts"; import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts"; @@ -85,6 +86,9 @@ const layerCommandReceiptStoreProvided = CommandReceiptStore.layer.pipe( ); const layerIsolatedOutbox = Layer.fresh(EffectOutbox.layer.pipe(Layer.provideMerge(layerDatabase))); const layerProjectionMaintenanceProvided = ProjectionMaintenance.layer.pipe( + Layer.provide(layerDatabase), +); +const layerProjectionReplayProvided = ProjectionReplay.layer.pipe( Layer.provide(layerStoresProvided), ); const layerTest = Layer.mergeAll( @@ -94,6 +98,7 @@ const layerTest = Layer.mergeAll( layerCommandReceiptStoreProvided, IdAllocator.layer, layerProjectionMaintenanceProvided, + layerProjectionReplayProvided, ); const providerInstanceId = ProviderInstanceId.make("codex"); @@ -148,85 +153,6 @@ function threadCreatedEvent(input: { }; } -it.effect("rebuilds event history one bounded page at a time", () => - Effect.gen(function* () { - const eventStore = yield* EventStore.EventStoreV2; - const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:foundation-paged-rebuild"); - const thread = makeThread(threadId, now); - const eventCount = 1_005; - yield* eventStore.append({ - events: [ - threadCreatedEvent({ id: "event:paged-rebuild:0", thread, now }), - ...Array.from({ length: eventCount - 1 }, (_, index) => ({ - id: EventId.make(`event:paged-rebuild:${index + 1}`), - type: "thread.metadata-updated" as const, - threadId, - occurredAt: now, - payload: { ...thread, title: `Rebuilt update ${index + 1}` }, - })), - ], - }); - let applied = 0; - let failAfterFirstPage = false; - const appliedAtRead: Array = []; - const layerObservedStores = Layer.mergeAll( - Layer.succeed(EventStore.EventStoreV2, { - ...eventStore, - read: (input) => - Stream.suspend(() => { - appliedAtRead.push(applied); - if (failAfterFirstPage && applied >= 500) { - return Stream.fail( - new EventStore.EventStoreReadEventsError({ afterSequence: input?.afterSequence }), - ); - } - return eventStore.read(input); - }), - }), - Layer.succeed(ProjectionStore.ProjectionStoreV2, { - ...projectionStore, - apply: (event) => - projectionStore.apply(event).pipe( - Effect.tap(() => - Effect.sync(() => { - applied += 1; - }), - ), - ), - }), - ); - const rebuild = Effect.gen(function* () { - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; - return yield* maintenance.rebuild; - }).pipe( - Effect.provide( - Layer.fresh(ProjectionMaintenance.layer.pipe(Layer.provide(layerObservedStores))), - ), - ); - const rebuilt = yield* rebuild; - assert.isTrue(rebuilt.valid); - assert.equal(applied, eventCount); - assert.deepEqual(appliedAtRead, [0, 500, 1_000]); - assert.equal( - (yield* projectionStore.getThreadProjection(threadId)).thread.title, - "Rebuilt update 1004", - ); - applied = 0; - failAfterFirstPage = true; - const failed = yield* Effect.exit(rebuild); - assert.equal(failed._tag, "Failure"); - assert.equal(applied, 500); - assert.equal( - (yield* projectionStore.getThreadProjection(threadId)).thread.title, - "Rebuilt update 1004", - ); - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; - assert.isTrue((yield* maintenance.verify).valid); - }).pipe(Effect.provide(layerTest)), -); - const seedCompaction = Effect.fnUntraced(function* () { const sql = yield* SqlClient.SqlClient; yield* sql`INSERT INTO orchestration_v2_legacy_imports ( @@ -240,6 +166,13 @@ const seedCompaction = Effect.fnUntraced(function* () { ) SELECT 'event-' || n, 'thread', 'thread', n, CASE WHEN n IN (1, 51, 101) THEN 'message.updated' ELSE 'thread.visited' END, '2026-10-17', 'server', '{"id":"message"}', '{}', 2 FROM history`; + // A pass a day ago marked the whole log; projections cover it. + yield* sql`INSERT INTO orchestration_v2_projection_metadata ( + projection_name, schema_version, last_sequence, updated_at + ) VALUES + ('event-log-prune-mark', 2, 151, '1969-12-30T00:00:00.000Z'), + ('thread-projections', 2, 151, '1969-12-30T00:00:00.000Z') + ON CONFLICT(projection_name) DO UPDATE SET last_sequence = excluded.last_sequence`; yield* sql`WITH RECURSIVE history(n) AS ( SELECT 1 UNION ALL SELECT n + 1 FROM history WHERE n < 100 ) INSERT INTO orchestration_command_receipts ( @@ -284,11 +217,12 @@ it.effect("discovers on the reader worker and deletes on the writer", () => }), ); const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; - const result = yield* maintenance.compactEventStore(); - assert.equal(result.deletedEventCount, 148); + const result = yield* maintenance.pruneEventStore(); + assert.equal(result.deletedEventCount, 150); assert.equal(result.deletedReceiptCount, 100); + // Only the in-transaction check for rows appended during the pass reads on the writer. assert.deepEqual( - reads.filter((query) => /\bSELECT\b/.test(query)), + reads.filter((query) => /\bSELECT\b/.test(query) && !query.includes("AS prompt_sequence")), [], ); assert(writes.some((query) => query.includes("DELETE FROM orchestration_events"))); @@ -296,7 +230,6 @@ it.effect("discovers on the reader worker and deletes on the writer", () => }).pipe( Effect.provide( ProjectionMaintenance.layer.pipe( - Layer.provide(Layer.merge(EventStore.layer, ProjectionStore.layer)), Layer.provideMerge(SqlitePersistence.layerFromPath(filename)), ), ), @@ -318,7 +251,7 @@ it.effect("paces scheduled compaction while preserving retention state and fixed const tracer = Tracer.make({ span(options) { const span = new Tracer.NativeSpan(options); - if (span.name === "ProjectionMaintenance.compactEventStore") compactionSpans.push(span); + if (span.name === "ProjectionMaintenance.pruneEventStore") compactionSpans.push(span); const end = span.end.bind(span); span.end = (time, exit) => { end(time, exit); @@ -377,15 +310,13 @@ it.effect("paces scheduled compaction while preserving retention state and fixed }, }), ); - assert.deepEqual(pauses, Array(5).fill(270)); + assert.deepEqual(pauses, Array(8).fill(270)); assert.deepEqual( compactionSpans.map((span) => span.status._tag), ["Ended"], ); + // 151 stopped being the stream head when 152 committed mid-pass; 152 is past the bound. assert.deepEqual(yield* sql`SELECT sequence FROM orchestration_events ORDER BY sequence`, [ - { sequence: 1 }, - { sequence: 101 }, - { sequence: 151 }, { sequence: 152 }, ]); assert.deepEqual(yield* sql`SELECT command_id FROM orchestration_command_receipts`, [ @@ -768,7 +699,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const now = yield* DateTime.now; const firstThreadId = ThreadId.make("thread:foundation-shared-session:first"); const secondThreadId = ThreadId.make("thread:foundation-shared-session:second"); @@ -829,7 +760,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { }), ); - it.effect("compacts superseded state events, imported v1 events, and legacy receipts", () => + it.effect("prunes covered history, imported v1 events, and legacy receipts", () => Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; @@ -983,9 +914,12 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { ON CONFLICT(command_id) DO NOTHING `; - const summary = yield* maintenance.compactEventStore(); + // The first pass only marks the head and drops imported legacy rows. + const marked = yield* maintenance.pruneEventStore(); + yield* TestClock.adjust("1 day"); + const summary = yield* maintenance.pruneEventStore(); assert.isAtLeast(summary.deletedEventCount, 504); - assert.isAtLeast(summary.deletedReceiptCount, 1); + assert.equal(marked.deletedReceiptCount + summary.deletedReceiptCount, 1); const remaining = yield* sql<{ readonly event_id: string }>` SELECT event_id @@ -996,18 +930,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { `; assert.deepEqual( remaining.map((row) => row.event_id), - [ - "event:foundation-compact:create", - "event:foundation-compact:meta", - "event:foundation-compact:message-1", - "event:foundation-compact:node-1", - "event:foundation-compact:item-1", - "event:foundation-compact:visit-2", - "event:foundation-compact:message-2", - "event:foundation-compact:node-2", - "event:foundation-compact:item-2", - "event:foundation-compact:tags-2", - ], + ["event:foundation-compact:create", "event:foundation-compact:tags-2"], ); const remainingV1 = yield* sql<{ readonly stream_id: string }>` @@ -1031,8 +954,6 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { ["command:foundation-compact:pending"], ); - assert.isTrue((yield* maintenance.verify).valid); - assert.isTrue((yield* maintenance.rebuild).valid); assert.deepEqual(yield* projections.getThreadProjection(threadId), beforeCompaction); }), ); @@ -1041,7 +962,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const sql = yield* SqlClient.SqlClient; const now = yield* DateTime.now; const parentThreadId = ThreadId.make("thread:foundation-cross-thread:parent"); @@ -3152,7 +3073,7 @@ it.layer(layerTest)("orchestration V2 foundation persistence", (it) => { Effect.gen(function* () { const eventSink = yield* EventSink.EventSinkV2; const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const sql = yield* SqlClient.SqlClient; const now = yield* DateTime.now; const threadId = ThreadId.make("thread:foundation-many-items"); @@ -3384,10 +3305,7 @@ it.effect.each(["events", "receipts"] as const)( const path = yield* Path.Path; const directory = yield* fs.makeTempDirectoryScoped({ prefix: "compaction-reader-" }); const database = SqlitePersistence.layerFromPath(path.join(directory, "statev2.sqlite")); - const maintenanceLayer = ProjectionMaintenance.layer.pipe( - Layer.provide(Layer.merge(EventStore.layer, ProjectionStore.layer)), - Layer.provideMerge(database), - ); + const maintenanceLayer = ProjectionMaintenance.layer.pipe(Layer.provideMerge(database)); yield* Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; @@ -3434,7 +3352,7 @@ it.effect.each(["events", "receipts"] as const)( : Effect.succeed(statement); }; const compacted = yield* Deferred.make(); - const compaction = yield* maintenance.compactEventStore().pipe( + const compaction = yield* maintenance.pruneEventStore().pipe( Effect.provideService(Statement.CurrentTransformer, observeDiscovery), Effect.provideService(Statement.SpanPropagationEnabled, true), Effect.tap(() => Deferred.succeed(compacted, undefined)), diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts new file mode 100644 index 0000000000000..50ed615f5b2f5 --- /dev/null +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts @@ -0,0 +1,355 @@ +import { assert, it } from "@effect/vitest"; +import { + EventId, + NodeId, + PlanId, + type OrchestrationV2AppThread, + type OrchestrationV2DomainEvent, + type OrchestrationV2TurnItem, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Fiber from "effect/Fiber"; +import * as Layer from "effect/Layer"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; + +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as EventSink from "./EventSink.ts"; +import * as EventStore from "./EventStore.ts"; +import * as ProjectionMaintenance from "./ProjectionMaintenance.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; +import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; + +const database = SqlitePersistence.layerMemory; +const eventStoreProvided = EventStore.layer.pipe(Layer.provideMerge(database)); +const projectionStoreProvided = ProjectionStore.layer.pipe(Layer.provideMerge(database)); +const stores = Layer.mergeAll(database, eventStoreProvided, projectionStoreProvided); +const layerTest = Layer.mergeAll( + stores, + OrchestrationEventStore.layer.pipe(Layer.provide(database)), + EventSink.layer.pipe(Layer.provide(stores)), + ProjectionMaintenance.layer.pipe(Layer.provide(stores)), +); + +const providerInstanceId = ProviderInstanceId.make("codex"); +const providerDriver = ProviderDriverKind.make("codex"); +const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" }; + +function makeThread(threadId: ThreadId, now: DateTime.Utc): OrchestrationV2AppThread { + return { + id: threadId, + projectId: ProjectId.make(`project:${threadId}`), + title: "Turn item compaction", + providerInstanceId, + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null }, + forkedFrom: null, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + }; +} + +function threadCreated(thread: OrchestrationV2AppThread, now: DateTime.Utc) { + return { + id: EventId.make("event:turn-item-compaction:thread"), + type: "thread.created", + threadId: thread.id, + providerInstanceId, + occurredAt: now, + payload: thread, + } satisfies OrchestrationV2DomainEvent; +} + +const itemSpecs = [ + { key: "command", type: "command_execution", ordinal: 10 }, + { key: "reasoning", type: "reasoning", ordinal: 20 }, + { key: "todo", type: "todo_list", ordinal: 30 }, + { key: "commanded", type: "command_execution", ordinal: 40 }, +] as const; + +function itemEvent( + threadId: ThreadId, + now: DateTime.Utc, + spec: (typeof itemSpecs)[number], + revision: number, + eventId = `event:turn-item-compaction:${spec.key}:${revision}`, +): OrchestrationV2DomainEvent { + const id = TurnItemId.make(`turn-item:compaction:${spec.key}`); + const growing = `${spec.key}:${revision}:` + "x".repeat((revision + 1) * 192); + const completed = revision === 69; + const base = { + id, + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: spec.ordinal, + status: completed ? ("completed" as const) : ("running" as const), + title: spec.key, + startedAt: now, + completedAt: completed ? now : null, + updatedAt: now, + }; + let payload: OrchestrationV2TurnItem; + switch (spec.type) { + case "command_execution": + payload = { + ...base, + type: "command_execution", + input: `run ${spec.key}`, + output: growing, + }; + break; + case "reasoning": + payload = { ...base, type: "reasoning", text: growing, streaming: !completed }; + break; + case "todo_list": + payload = { + ...base, + type: "todo_list", + planId: PlanId.make("plan:turn-item-compaction"), + steps: [{ id: "step", text: growing, status: completed ? "completed" : "running" }], + explanation: growing, + }; + break; + } + return { + id: EventId.make(eventId), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload, + }; +} + +function subagentEvents( + threadId: ThreadId, + now: DateTime.Utc, + revision: number, + eventSuffix = String(revision), +): ReadonlyArray { + const id = NodeId.make("subagent:turn-item-compaction"); + const prompt = revision === 0 ? "Review prompt A. ".repeat(40) : "Review prompt B. ".repeat(40); + const progress = `Review progress ${revision}`; + return [ + { + id: EventId.make(`event:turn-item-compaction:subagent:${eventSuffix}`), + type: "subagent.updated", + threadId, + occurredAt: now, + payload: { + id, + threadId, + runId: null, + parentNodeId: NodeId.make("node:turn-item-compaction-parent"), + origin: "provider_native", + createdBy: "agent", + driver: providerDriver, + providerInstanceId, + providerThreadId: null, + childThreadId: null, + nativeTaskRef: { + driver: providerDriver, + nativeId: "compaction-review", + strength: "strong", + }, + prompt, + progress, + title: "Review", + model: null, + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + }, + { + id: EventId.make(`event:turn-item-compaction:subagent-item:${eventSuffix}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make("turn-item:compaction:subagent"), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 50, + status: "running", + title: "Review", + startedAt: now, + completedAt: null, + updatedAt: now, + type: "subagent", + subagentId: id, + origin: "provider_native", + driver: providerDriver, + providerInstanceId, + childThreadId: null, + prompt, + progress, + result: null, + }, + }, + ] satisfies ReadonlyArray; +} + +const insertRunningRunUpdate = (threadId: ThreadId) => + Effect.flatMap( + SqlClient.SqlClient, + (sql) => sql<{ readonly sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + 'event:prune:run-running', 'thread', ${threadId}, + (SELECT MAX(stream_version) + 1 FROM orchestration_events WHERE stream_id = ${threadId}), + 'run.updated', '2026-11-01T00:00:00.000Z', 'server', '{"status":"running"}', '{}', 2 + ) RETURNING sequence + `, + ); + +const remainingEventIds = (threadId: ThreadId) => + Effect.flatMap( + SqlClient.SqlClient, + (sql) => sql<{ readonly event_id: string }>` + SELECT event_id FROM orchestration_events WHERE stream_id = ${threadId} ORDER BY sequence + `, + ).pipe(Effect.map((rows) => rows.map((row) => row.event_id))); + +it.effect("prunes events the projections cover once the retained day has passed", () => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const eventStore = yield* EventStore.EventStoreV2; + const events = yield* OrchestrationEventStore.OrchestrationEventStore; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:prune"); + yield* sink.write({ events: [threadCreated(makeThread(threadId, now), now)] }); + for (const revision of [0, 1, 2]) { + yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[0], revision)] }); + } + yield* sink.write({ events: subagentEvents(threadId, now, 0) }); + yield* insertRunningRunUpdate(threadId); + yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[1], 0)] }); + + // The first pass only marks the head; nothing is a day old yet. + assert.equal((yield* maintenance.pruneEventStore()).deletedEventCount, 0); + const markedThrough = yield* eventStore.latestSequence(); + + // Repeating prompt A borrows it from the anchors written before the mark. + yield* sink.write({ events: subagentEvents(threadId, now, 0, "repeat") }); + yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[0], 3)] }); + const highWater = yield* eventStore.latestSequence(); + const projection = yield* projections.getThreadProjection(threadId); + + yield* TestClock.adjust("23 hours"); + assert.equal((yield* maintenance.pruneEventStore()).deletedEventCount, 0); + yield* TestClock.adjust("1 hour"); + assert.equal((yield* maintenance.pruneEventStore()).deletedEventCount, 4); + + assert.deepEqual(yield* remainingEventIds(threadId), [ + "event:turn-item-compaction:thread", + "event:turn-item-compaction:subagent:0", + "event:turn-item-compaction:subagent-item:0", + "event:prune:run-running", + "event:turn-item-compaction:subagent:repeat", + "event:turn-item-compaction:subagent-item:repeat", + "event:turn-item-compaction:command:3", + ]); + assert.equal(yield* events.prunedThroughSequence, markedThrough); + assert.equal(yield* eventStore.latestSequence(), highWater); + assert.equal(yield* eventStore.latestSequence({ threadId }), highWater); + assert.deepEqual(yield* projections.getThreadProjection(threadId), projection); + const repeated = yield* eventStore + .read({ threadId, eventType: "subagent.updated", afterSequence: markedThrough }) + .pipe(Stream.runCollect); + assert.deepEqual( + Array.from(repeated, (stored) => + stored.event.type === "subagent.updated" ? stored.event.payload.prompt : null, + ), + ["Review prompt A. ".repeat(40)], + ); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("keeps an anchor that a row appended during the pass borrows from", () => + Effect.scoped( + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const eventStore = yield* EventStore.EventStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:prune-race"); + yield* sink.write({ events: [threadCreated(makeThread(threadId, now), now)] }); + yield* sink.write({ events: subagentEvents(threadId, now, 0) }); + yield* maintenance.pruneEventStore(); + yield* sink.write({ events: [itemEvent(threadId, now, itemSpecs[0], 0)] }); + yield* TestClock.adjust("1 day"); + + const scanStarted = yield* Deferred.make(); + const releaseScan = yield* Deferred.make(); + let blocked = false; + const blockFirstScanPage: Statement.Transformer = (statement) => { + const [query] = statement.compile(); + if (!blocked && query.includes("END AS prunable")) { + blocked = true; + return Deferred.succeed(scanStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseScan)), + Effect.as(statement), + ); + } + return Effect.succeed(statement); + }; + const pass = yield* maintenance + .pruneEventStore() + .pipe( + Effect.provideService(Statement.CurrentTransformer, blockFirstScanPage), + Effect.forkScoped, + ); + yield* Deferred.await(scanStarted); + yield* sink.write({ events: subagentEvents(threadId, now, 0, "late") }); + yield* Deferred.succeed(releaseScan, undefined); + assert.equal((yield* Fiber.join(pass)).deletedEventCount, 0); + + const late = yield* eventStore + .read({ threadId, eventType: "subagent.updated" }) + .pipe(Stream.runCollect); + assert.deepEqual( + Array.from(late, (stored) => + stored.event.type === "subagent.updated" ? stored.event.payload.prompt : null, + ), + ["Review prompt A. ".repeat(40), "Review prompt A. ".repeat(40)], + ); + }), + ).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 211fb2ad8e7a3..9f6f2596d443e 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -1,26 +1,18 @@ -import { ThreadId } from "@t3tools/contracts"; import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient"; import * as Context from "effect/Context"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Schema from "effect/Schema"; -import * as Stream from "effect/Stream"; import * as SqlClient from "effect/sql/SqlClient"; -import * as EventStore from "./EventStore.ts"; +import { + EVENT_LOG_PRUNE_MARK, + EVENT_LOG_PRUNED_THROUGH, +} from "../persistence/OrchestrationEventStore.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; -export interface ProjectionVerificationV2 { - readonly valid: boolean; - readonly schemaVersion: number; - readonly expectedSequence: number; - readonly projectionSequence: number; - readonly unreadableThreadIds: ReadonlyArray; - readonly missingThreadIds: ReadonlyArray; - readonly unexpectedThreadIds: ReadonlyArray; -} - export class ProjectionMaintenanceError extends Schema.TaggedError()( "ProjectionMaintenanceError", { @@ -30,9 +22,8 @@ export class ProjectionMaintenanceError extends Schema.TaggedError; - readonly rebuild: Effect.Effect; - readonly compactEventStore: (betweenPages?: Effect.Effect) => Effect.Effect< + /** Deletes events the projections already cover once they are older than the retained history. */ + readonly pruneEventStore: (betweenPages?: Effect.Effect) => Effect.Effect< { readonly deletedEventCount: number; readonly deletedReceiptCount: number; @@ -47,369 +38,268 @@ export class ProjectionMaintenanceV2 extends Context.Service< ProjectionMaintenanceV2Shape >()("t3/orchestration-v2/ProjectionMaintenance/ProjectionMaintenanceV2") {} -type ProjectionMetadataRow = { - readonly schema_version: number; +// Reconnect replay and command-id retries read this tail; older cursors resync from a snapshot. +const RETAINED_EVENT_HISTORY = Duration.days(1); +// Small pages bound work on the reader worker and synchronous fallback. +const PAGE_SIZE = 50; +const REFERENCE_SCAN_WINDOW = 1000; + +type MetadataRow = { + readonly projection_name: string; readonly last_sequence: number; + readonly updated_at: string; }; -const encodeEntityKey = Schema.encodeSync( - Schema.fromJsonString(Schema.Tuple([Schema.String, Schema.String, Schema.NullOr(Schema.String)])), -); - -export const layer: Layer.Layer< - ProjectionMaintenanceV2, - never, - EventStore.EventStoreV2 | ProjectionStore.ProjectionStoreV2 | SqlClient.SqlClient -> = Layer.effect( +export const layer: Layer.Layer = Layer.effect( ProjectionMaintenanceV2, Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; - const eventStore = yield* EventStore.EventStoreV2; - const projectionStore = yield* ProjectionStore.ProjectionStoreV2; - const verify = Effect.gen(function* () { - const expectedThreadRows = yield* sql<{ readonly thread_id: string }>` - SELECT DISTINCT stream_id AS thread_id - FROM orchestration_events INDEXED BY orchestration_events_v2_created_threads_idx - WHERE application_event_version = 2 - AND aggregate_kind = 'thread' - AND event_type = 'thread.created' - ORDER BY stream_id ASC - `; - const projectionRows = yield* sql<{ readonly thread_id: string }>` - SELECT thread_id - FROM orchestration_v2_projection_threads - ORDER BY thread_id ASC - `; - const actualIds = projectionRows.map((row) => ThreadId.make(row.thread_id)); - const expectedIds = expectedThreadRows.map((row) => ThreadId.make(row.thread_id)); - const actualSet = new Set(actualIds); - const expectedSet = new Set(expectedIds); - const missingThreadIds = expectedIds.filter((threadId) => !actualSet.has(threadId)); - const unexpectedThreadIds = actualIds.filter((threadId) => !expectedSet.has(threadId)); - const unreadableThreadIds = yield* projectionStore.getUnreadableThreadIds(); - const metadata = yield* sql` - SELECT schema_version, last_sequence - FROM orchestration_v2_projection_metadata - WHERE projection_name = 'thread-projections' - LIMIT 1 - `; - const expectedSequence = yield* eventStore.latestSequence(); - const schemaVersion = metadata[0]?.schema_version ?? 0; - const projectionSequence = metadata[0]?.last_sequence ?? 0; - return { - valid: - schemaVersion === ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION && - projectionSequence === expectedSequence && - missingThreadIds.length === 0 && - unexpectedThreadIds.length === 0 && - unreadableThreadIds.length === 0, - schemaVersion, - expectedSequence, - projectionSequence, - unreadableThreadIds, - missingThreadIds, - unexpectedThreadIds, - } satisfies ProjectionVerificationV2; - }).pipe(sql.withTransaction); + const writeMetadata = (name: string, sequence: number, at: DateTime.Utc) => sql` + INSERT INTO orchestration_v2_projection_metadata ( + projection_name, schema_version, last_sequence, updated_at + ) + VALUES ( + ${name}, + ${ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION}, + ${sequence}, + ${DateTime.formatIso(at)} + ) + ON CONFLICT(projection_name) DO UPDATE SET + last_sequence = excluded.last_sequence, + updated_at = excluded.updated_at + `; - const rebuild = Effect.gen(function* () { - yield* sql.withTransaction( - Effect.gen(function* () { - const throughSequence = yield* eventStore.latestSequence(); - yield* sql`DELETE FROM orchestration_v2_projection_context_transfers`; - yield* sql`DELETE FROM orchestration_v2_projection_context_handoffs`; - yield* sql`DELETE FROM orchestration_v2_projection_checkpoints`; - yield* sql`DELETE FROM orchestration_v2_projection_checkpoint_scopes`; - yield* sql`DELETE FROM orchestration_v2_projection_turn_items`; - yield* sql`DELETE FROM orchestration_v2_projection_plans`; - yield* sql`DELETE FROM orchestration_v2_projection_messages`; - yield* sql`DELETE FROM orchestration_v2_projection_runtime_requests`; - yield* sql`DELETE FROM orchestration_v2_projection_provider_turns`; - yield* sql`DELETE FROM orchestration_v2_projection_provider_threads`; - yield* sql`DELETE FROM orchestration_v2_projection_provider_session_bindings`; - yield* sql`DELETE FROM orchestration_v2_projection_provider_sessions`; - yield* sql`DELETE FROM orchestration_v2_projection_subagents`; - yield* sql`DELETE FROM orchestration_v2_projection_nodes`; - yield* sql`DELETE FROM orchestration_v2_projection_run_attempts`; - yield* sql`DELETE FROM orchestration_v2_projection_runs`; - yield* sql`DELETE FROM orchestration_v2_projection_threads`; - yield* sql`DELETE FROM orchestration_v2_turn_item_positions`; + // Collects anchors whose prompt later rows borrow through promptSequence. + const readPromptReferences = Effect.fnUntraced(function* ( + afterSequence: number, + throughSequence: number, + betweenPages: Effect.Effect, + ) { + const referenced = new Set(); + for (let cursor = afterSequence; cursor < throughSequence;) { + const windowEnd = Math.min(cursor + REFERENCE_SCAN_WINDOW, throughSequence); + const rows = yield* sql<{ readonly prompt_sequence: number }>` + SELECT json_extract(payload_json, '$.promptSequence') AS prompt_sequence + FROM orchestration_events + WHERE sequence > ${cursor} + AND sequence <= ${windowEnd} + AND event_type IN ('subagent.updated', 'turn-item.updated') + AND json_type(payload_json, '$.promptSequence') = 'integer' + `.pipe(NodeSqliteClient.readOnly); + for (const row of rows) referenced.add(row.prompt_sequence); + cursor = windowEnd; + yield* betweenPages; + } + return referenced; + }); - const pageSize = 500; - let lastSequence = 0; - while (true) { - const page = yield* eventStore - .read({ afterSequence: lastSequence, throughSequence, limit: pageSize }) - .pipe(Stream.runCollect); - for (const stored of page) { - yield* projectionStore.apply(stored.event); - if (stored.event.type === "turn-item.updated") { - yield* sql` - INSERT INTO orchestration_v2_turn_item_positions ( - thread_id, - turn_item_id, - ordinal - ) - VALUES ( - ${stored.event.threadId}, - ${stored.event.payload.id}, - ${stored.event.payload.ordinal} - ) - ON CONFLICT(thread_id, turn_item_id) DO UPDATE SET - ordinal = excluded.ordinal - `; + const pruneEvents = Effect.fnUntraced(function* ( + throughSequence: number, + head: number, + betweenPages: Effect.Effect, + ) { + const referenced = yield* readPromptReferences(throughSequence, head, betweenPages); + let referencesThrough = head; + let deleted = 0; + let cursor = 0; + while (cursor < throughSequence) { + const rows = yield* sql<{ readonly sequence: number; readonly prunable: number }>` + SELECT + event.sequence, + CASE + WHEN event.application_event_version = 1 THEN 0 + -- The legacy importer treats a thread without thread.created as not yet imported. + WHEN event.event_type = 'thread.created' THEN 0 + -- Codex usage history reads the selection a run attempt started with. + WHEN event.event_type = 'run.updated' + AND json_extract(event.payload_json, '$.status') = 'running' THEN 0 + -- The newest row keeps each stream's high water, stream_version, and snapshot sequence. + WHEN event.sequence = ( + SELECT MAX(newer.sequence) FROM orchestration_events AS newer + WHERE newer.aggregate_kind = event.aggregate_kind + AND newer.stream_id = event.stream_id + ) THEN 0 + ELSE 1 + END AS prunable + FROM orchestration_events AS event + WHERE event.sequence > ${cursor} + AND event.sequence <= ${throughSequence} + ORDER BY event.sequence ASC + LIMIT ${PAGE_SIZE} + `.pipe(NodeSqliteClient.readOnly); + const candidates = rows.filter((row) => row.prunable === 1).map((row) => row.sequence); + // Never hold a transaction or the reader lock across a yield or a rest. + yield* Effect.yieldNow; + if (candidates.length > 0) { + deleted += yield* sql.withTransaction( + Effect.gen(function* () { + // Rows appended since the reference scan may borrow a prompt from a candidate. + const appended = yield* sql<{ + readonly sequence: number; + readonly prompt_sequence: number | null; + }>` + SELECT + sequence, + CASE + WHEN event_type IN ('subagent.updated', 'turn-item.updated') + AND json_type(payload_json, '$.promptSequence') = 'integer' + THEN json_extract(payload_json, '$.promptSequence') + END AS prompt_sequence + FROM orchestration_events + WHERE sequence > ${referencesThrough} + `; + for (const row of appended) { + if (row.prompt_sequence !== null) referenced.add(row.prompt_sequence); + referencesThrough = Math.max(referencesThrough, row.sequence); } - lastSequence = stored.sequence; - } - if (page.length < pageSize) break; - yield* Effect.yieldNow; - } - const now = DateTime.formatIso(yield* DateTime.now); - yield* sql` - INSERT INTO orchestration_v2_projection_metadata ( - projection_name, - schema_version, - last_sequence, - updated_at - ) - VALUES ( - 'thread-projections', - ${ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION}, - ${lastSequence}, - ${now} - ) - ON CONFLICT(projection_name) DO UPDATE SET - schema_version = excluded.schema_version, - last_sequence = excluded.last_sequence, - updated_at = excluded.updated_at - `; - }), - ); - return yield* verify; + const obsolete = candidates.filter((sequence) => !referenced.has(sequence)); + if (obsolete.length === 0) return 0; + yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`; + return obsolete.length; + }), + ); + yield* Effect.yieldNow; + } + // Rest after every page, including the last, so the next loop starts a fresh slice. + yield* betweenPages; + if (rows.length < PAGE_SIZE) break; + cursor = rows.at(-1)?.sequence ?? throughSequence; + } + return deleted; }); - const mapError = - (operation: string) => - (effect: Effect.Effect) => - effect.pipe( - Effect.mapError((cause) => new ProjectionMaintenanceError({ operation, cause })), - ); - - // Keep thread.created as the verification and replay anchor; clients recognize deletion by thread.deleted event type. - const SUPERSEDABLE_THREAD_EVENT_TYPES = [ - "thread.archived", - "thread.unarchived", - "thread.settled", - "thread.unsettled", - "thread.snoozed", - "thread.unsnoozed", - "thread.pinned", - "thread.auto-settle-set", - "thread.unpinned", - "thread.pin-reordered", - "thread.active-reordered", - "thread.metadata-updated", - "thread.pull-request-synced", - "thread.runtime-mode-updated", - "thread.interaction-mode-updated", - "thread.model-selection-updated", - "thread.provider-switched", - "thread.guest-share-updated", - ]; - // Read-state events preserve updatedAt and supersede only each other. Activity events supersede both groups. - const READ_STATE_THREAD_EVENT_TYPES = ["thread.visited", "thread.marked-unread"]; - // Older clients skip tag events, so these supersede only each other. - const TAG_THREAD_EVENT_TYPE = "thread.tags-updated"; - - const supersedableThreadEventTypes = new Set(SUPERSEDABLE_THREAD_EVENT_TYPES); - const readStateThreadEventTypes = new Set(READ_STATE_THREAD_EVENT_TYPES); - // Subagent snapshots stay intact because later rows can reference an earlier prompt. - // Commanded turn items stay for command-result replay. - // Small pages bound work on the reader worker and synchronous fallback. - const COMPACTION_PAGE_SIZE = 50; - - // Keep first/latest keyed events for replay order; page every event to bound sparse scans. - const compactEventStore = Effect.fn("ProjectionMaintenance.compactEventStore")(function* ( - betweenPages: Effect.Effect = Effect.void, + // Imported legacy threads are served from v2 rows; their v1 events and receipts have no reader. + const pruneImportedLegacyHistory = Effect.fnUntraced(function* ( + throughSequence: number, + throughReceiptRowId: number, + betweenPages: Effect.Effect, ) { - const bounds = yield* sql<{ - readonly event_sequence: number; - readonly receipt_row_id: number; - }>` - SELECT - COALESCE((SELECT MAX(sequence) FROM orchestration_events), 0) AS event_sequence, - COALESCE((SELECT MAX(rowid) FROM orchestration_command_receipts), 0) AS receipt_row_id - `.pipe(NodeSqliteClient.readOnly); - let throughSequence = bounds[0]?.event_sequence ?? 0; - let throughReceiptRowId = bounds[0]?.receipt_row_id ?? 0; - const retainedThreadIds = new Set(); - const retainedReadStateThreadIds = new Set(); - const retainedTagThreadIds = new Set(); - // Retain each key's oldest candidate, deleting it only when an older event appears. - const oldestCandidateByEntityKey = new Map(); let deletedEventCount = 0; let deletedReceiptCount = 0; - - while (throughSequence > 0) { - const rows = yield* sql<{ - readonly sequence: number; - readonly application_event_version: number; - readonly aggregate_kind: string; - readonly stream_id: string; - readonly event_type: string; - readonly entity_id: string | null; - readonly imported_legacy_thread: number; - }>` + for (let cursor = 0; ;) { + const rows = yield* sql<{ readonly sequence: number; readonly imported: number }>` SELECT event.sequence, - event.application_event_version, - event.aggregate_kind, - event.stream_id, - event.event_type, - CASE - WHEN event.application_event_version = 2 - AND event.event_type IN ('message.updated', 'node.updated') - THEN json_extract(event.payload_json, '$.id') - WHEN event.application_event_version = 2 - AND event.event_type = 'turn-item.updated' - AND event.command_id IS NULL - AND json_extract(event.payload_json, '$.type') <> 'subagent' - THEN json_extract(event.payload_json, '$.id') - WHEN event.application_event_version = 2 - AND event.event_type = 'thread.pull-request-link-updated' - THEN json_array(json_extract(event.payload_json, '$.host'), - json_extract(event.payload_json, '$.repository'), - json_extract(event.payload_json, '$.number')) - ELSE NULL - END AS entity_id, - CASE - WHEN event.application_event_version = 1 AND event.aggregate_kind = 'thread' - THEN EXISTS ( - SELECT 1 FROM orchestration_v2_legacy_imports AS legacy_import - WHERE legacy_import.thread_id = event.stream_id - AND legacy_import.transcript_imported_at IS NOT NULL - ) - ELSE 0 - END AS imported_legacy_thread + event.aggregate_kind = 'thread' AND EXISTS ( + SELECT 1 FROM orchestration_v2_legacy_imports AS legacy_import + WHERE legacy_import.thread_id = event.stream_id + AND legacy_import.transcript_imported_at IS NOT NULL + ) AS imported FROM orchestration_events AS event - WHERE event.sequence <= ${throughSequence} - ORDER BY event.sequence DESC - LIMIT ${COMPACTION_PAGE_SIZE + 1} + INDEXED BY idx_orchestration_events_application_sequence + WHERE event.application_event_version = 1 + AND event.sequence > ${cursor} + AND event.sequence <= ${throughSequence} + ORDER BY event.sequence ASC + LIMIT ${PAGE_SIZE} `.pipe(NodeSqliteClient.readOnly); - const page = rows.slice(0, COMPACTION_PAGE_SIZE); - const obsolete: number[] = []; - for (const row of page) { - if (row.imported_legacy_thread === 1) { - obsolete.push(row.sequence); - } else if (row.application_event_version === 2) { - if ( - row.aggregate_kind === "thread" && - supersedableThreadEventTypes.has(row.event_type) - ) { - if (retainedThreadIds.has(row.stream_id)) obsolete.push(row.sequence); - else retainedThreadIds.add(row.stream_id); - } else if ( - row.aggregate_kind === "thread" && - readStateThreadEventTypes.has(row.event_type) - ) { - if ( - retainedThreadIds.has(row.stream_id) || - retainedReadStateThreadIds.has(row.stream_id) - ) { - obsolete.push(row.sequence); - } else retainedReadStateThreadIds.add(row.stream_id); - } else if ( - row.aggregate_kind === "thread" && - row.event_type === TAG_THREAD_EVENT_TYPE - ) { - if (retainedThreadIds.has(row.stream_id) || retainedTagThreadIds.has(row.stream_id)) { - obsolete.push(row.sequence); - } else retainedTagThreadIds.add(row.stream_id); - } else if ( - row.event_type === "thread.pull-request-link-updated" && - retainedThreadIds.has(row.stream_id) - ) { - // A newer whole-thread payload already carries this link's state. - obsolete.push(row.sequence); - } else if ( - row.event_type === "thread.pull-request-link-updated" || - row.event_type === "message.updated" || - row.event_type === "node.updated" || - (row.event_type === "turn-item.updated" && row.entity_id !== null) - ) { - const key = encodeEntityKey([row.event_type, row.stream_id, row.entity_id]); - const candidate = oldestCandidateByEntityKey.get(key); - if (candidate === undefined) oldestCandidateByEntityKey.set(key, null); - else { - if (candidate !== null) obsolete.push(candidate); - oldestCandidateByEntityKey.set(key, row.sequence); - } - } - } - } - // Never hold a transaction or the reader lock across a yield or a rest. + const obsolete = rows.filter((row) => row.imported === 1).map((row) => row.sequence); yield* Effect.yieldNow; if (obsolete.length > 0) { yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`; deletedEventCount += obsolete.length; yield* Effect.yieldNow; } - throughSequence = - rows.length <= COMPACTION_PAGE_SIZE ? 0 : (page.at(-1)?.sequence ?? 1) - 1; - if (throughSequence > 0 || throughReceiptRowId > 0) yield* betweenPages; + yield* betweenPages; + if (rows.length < PAGE_SIZE) break; + cursor = rows.at(-1)?.sequence ?? cursor; } - - while (throughReceiptRowId > 0) { + for (let cursor = 0; ;) { const rows = yield* sql<{ readonly row_id: number; readonly command_id: string; - readonly imported_legacy_thread: number; + readonly imported: number; }>` SELECT receipt.rowid AS row_id, receipt.command_id, - CASE - WHEN receipt.command_type = 'legacy' AND receipt.aggregate_kind = 'thread' - THEN EXISTS ( - SELECT 1 FROM orchestration_v2_legacy_imports AS legacy_import - WHERE legacy_import.thread_id = receipt.aggregate_id - AND legacy_import.transcript_imported_at IS NOT NULL - ) - ELSE 0 - END AS imported_legacy_thread + receipt.command_type = 'legacy' AND receipt.aggregate_kind = 'thread' AND EXISTS ( + SELECT 1 FROM orchestration_v2_legacy_imports AS legacy_import + WHERE legacy_import.thread_id = receipt.aggregate_id + AND legacy_import.transcript_imported_at IS NOT NULL + ) AS imported FROM orchestration_command_receipts AS receipt - WHERE receipt.rowid <= ${throughReceiptRowId} - ORDER BY receipt.rowid DESC - LIMIT ${COMPACTION_PAGE_SIZE + 1} + WHERE receipt.rowid > ${cursor} AND receipt.rowid <= ${throughReceiptRowId} + ORDER BY receipt.rowid ASC + LIMIT ${PAGE_SIZE} `.pipe(NodeSqliteClient.readOnly); - const page = rows.slice(0, COMPACTION_PAGE_SIZE); - const obsolete = page - .filter((row) => row.imported_legacy_thread === 1) - .map((row) => row.command_id); + const obsolete = rows.filter((row) => row.imported === 1).map((row) => row.command_id); yield* Effect.yieldNow; if (obsolete.length > 0) { yield* sql`DELETE FROM orchestration_command_receipts WHERE command_id IN ${sql.in(obsolete)}`; deletedReceiptCount += obsolete.length; yield* Effect.yieldNow; } - throughReceiptRowId = (page.at(-1)?.row_id ?? 1) - 1; - if (rows.length <= COMPACTION_PAGE_SIZE) break; yield* betweenPages; + if (rows.length < PAGE_SIZE) break; + cursor = rows.at(-1)?.row_id ?? cursor; } + return { deletedEventCount, deletedReceiptCount }; + }); + + // A pass prunes through the mark the previous pass left once that mark is a day old, so + // at least RETAINED_EVENT_HISTORY of events always remains. Receipts stay for idempotency. + const pruneEventStore = Effect.fn("ProjectionMaintenance.pruneEventStore")( + function* (betweenPages: Effect.Effect = Effect.void) { + const now = yield* DateTime.now; + const metadata = yield* sql` + SELECT projection_name, last_sequence, updated_at + FROM orchestration_v2_projection_metadata + WHERE projection_name IN ( + 'thread-projections', ${EVENT_LOG_PRUNE_MARK}, ${EVENT_LOG_PRUNED_THROUGH} + ) + `.pipe(NodeSqliteClient.readOnly); + const read = (name: string) => metadata.find((row) => row.projection_name === name); + const projected = read("thread-projections")?.last_sequence ?? 0; + const mark = read(EVENT_LOG_PRUNE_MARK); + const prunedThrough = read(EVENT_LOG_PRUNED_THROUGH)?.last_sequence ?? 0; + // Fixed bounds: foreground commits during the pass wait for the next one. + const bounds = yield* sql<{ readonly sequence: number; readonly receipt_row_id: number }>` + SELECT + COALESCE((SELECT MAX(sequence) FROM orchestration_events), 0) AS sequence, + COALESCE((SELECT MAX(rowid) FROM orchestration_command_receipts), 0) AS receipt_row_id + `.pipe(NodeSqliteClient.readOnly); + const head = bounds[0]?.sequence ?? 0; - const freelistRows = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`; - const pageSizeRows = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`; - const reclaimableBytes = - (freelistRows[0]?.freelist_count ?? 0) * (pageSizeRows[0]?.page_size ?? 0); + let deletedEventCount = 0; + const markIsDue = + mark !== undefined && + DateTime.toEpochMillis(now) - Date.parse(mark.updated_at) >= + Duration.toMillis(RETAINED_EVENT_HISTORY); + if (mark !== undefined && markIsDue) { + const throughSequence = Math.max(Math.min(mark.last_sequence, projected), prunedThrough); + if (throughSequence > prunedThrough) { + // Publish the boundary first so resuming clients take a snapshot, never a partial replay. + yield* writeMetadata(EVENT_LOG_PRUNED_THROUGH, throughSequence, now); + } + deletedEventCount += yield* pruneEvents(throughSequence, head, betweenPages); + } + if (mark === undefined || markIsDue) yield* writeMetadata(EVENT_LOG_PRUNE_MARK, head, now); + + const legacy = yield* pruneImportedLegacyHistory( + head, + bounds[0]?.receipt_row_id ?? 0, + betweenPages, + ); + deletedEventCount += legacy.deletedEventCount; - return { deletedEventCount, deletedReceiptCount, reclaimableBytes }; - }, mapError("compact event store")); + const freelistRows = yield* sql<{ readonly freelist_count: number }>`PRAGMA freelist_count`; + const pageSizeRows = yield* sql<{ readonly page_size: number }>`PRAGMA page_size`; + const reclaimableBytes = + (freelistRows[0]?.freelist_count ?? 0) * (pageSizeRows[0]?.page_size ?? 0); - return ProjectionMaintenanceV2.of({ - verify: mapError("verify")(verify), - rebuild: mapError("rebuild")(rebuild), - compactEventStore, - }); + return { + deletedEventCount, + deletedReceiptCount: legacy.deletedReceiptCount, + reclaimableBytes, + }; + }, + Effect.mapError( + (cause) => new ProjectionMaintenanceError({ operation: "prune event store", cause }), + ), + ); + + return ProjectionMaintenanceV2.of({ pruneEventStore }); }), ); diff --git a/apps/server/src/orchestration-v2/TurnItemCompaction.test.ts b/apps/server/src/orchestration-v2/TurnItemCompaction.test.ts deleted file mode 100644 index 5799c5fbdcd31..0000000000000 --- a/apps/server/src/orchestration-v2/TurnItemCompaction.test.ts +++ /dev/null @@ -1,488 +0,0 @@ -import { applyOrchestrationV2ProjectionEvent } from "../../../../packages/client-runtime/src/state/orchestrationV2Projection.ts"; -import { assert, it } from "@effect/vitest"; -import { - CommandId, - EventId, - NodeId, - PlanId, - type OrchestrationV2AppThread, - type OrchestrationV2DomainEvent, - type OrchestrationV2TurnItem, - ProjectId, - ProviderDriverKind, - ProviderInstanceId, - ThreadId, - TurnItemId, -} from "@t3tools/contracts"; -import * as DateTime from "effect/DateTime"; -import * as Deferred from "effect/Deferred"; -import * as Effect from "effect/Effect"; -import * as Fiber from "effect/Fiber"; -import * as Layer from "effect/Layer"; -import * as Stream from "effect/Stream"; -import * as Tracer from "effect/Tracer"; -import * as SqlClient from "effect/sql/SqlClient"; -import * as Statement from "effect/sql/Statement"; - -import * as SqlitePersistence from "../persistence/Sqlite.ts"; -import * as EventSink from "./EventSink.ts"; -import * as EventStore from "./EventStore.ts"; -import * as ProjectionMaintenance from "./ProjectionMaintenance.ts"; -import * as ProjectionStore from "./ProjectionStore.ts"; - -const database = SqlitePersistence.layerMemory; -const eventStoreProvided = EventStore.layer.pipe(Layer.provideMerge(database)); -const projectionStoreProvided = ProjectionStore.layer.pipe(Layer.provideMerge(database)); -const stores = Layer.mergeAll(database, eventStoreProvided, projectionStoreProvided); -const layerTest = Layer.mergeAll( - stores, - EventSink.layer.pipe(Layer.provide(stores)), - ProjectionMaintenance.layer.pipe(Layer.provide(stores)), -); - -const providerInstanceId = ProviderInstanceId.make("codex"); -const providerDriver = ProviderDriverKind.make("codex"); -const modelSelection = { instanceId: providerInstanceId, model: "gpt-5.4" }; - -function makeThread(threadId: ThreadId, now: DateTime.Utc): OrchestrationV2AppThread { - return { - id: threadId, - projectId: ProjectId.make(`project:${threadId}`), - title: "Turn item compaction", - providerInstanceId, - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - activeProviderThreadId: null, - lineage: { rootThreadId: threadId, parentThreadId: null, relationshipToParent: null }, - forkedFrom: null, - createdBy: "user", - creationSource: "web", - createdAt: now, - updatedAt: now, - archivedAt: null, - settledOverride: null, - settledAt: null, - lastVisitedAt: null, - deletedAt: null, - }; -} - -function threadCreated(thread: OrchestrationV2AppThread, now: DateTime.Utc) { - return { - id: EventId.make("event:turn-item-compaction:thread"), - type: "thread.created", - threadId: thread.id, - providerInstanceId, - occurredAt: now, - payload: thread, - } satisfies OrchestrationV2DomainEvent; -} - -const itemSpecs = [ - { key: "command", type: "command_execution", ordinal: 10 }, - { key: "reasoning", type: "reasoning", ordinal: 20 }, - { key: "todo", type: "todo_list", ordinal: 30 }, - { key: "commanded", type: "command_execution", ordinal: 40 }, -] as const; - -function itemEvent( - threadId: ThreadId, - now: DateTime.Utc, - spec: (typeof itemSpecs)[number], - revision: number, - eventId = `event:turn-item-compaction:${spec.key}:${revision}`, -): OrchestrationV2DomainEvent { - const id = TurnItemId.make(`turn-item:compaction:${spec.key}`); - const growing = `${spec.key}:${revision}:` + "x".repeat((revision + 1) * 192); - const completed = revision === 69; - const base = { - id, - threadId, - runId: null, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: spec.ordinal, - status: completed ? ("completed" as const) : ("running" as const), - title: spec.key, - startedAt: now, - completedAt: completed ? now : null, - updatedAt: now, - }; - let payload: OrchestrationV2TurnItem; - switch (spec.type) { - case "command_execution": - payload = { - ...base, - type: "command_execution", - input: `run ${spec.key}`, - output: growing, - }; - break; - case "reasoning": - payload = { ...base, type: "reasoning", text: growing, streaming: !completed }; - break; - case "todo_list": - payload = { - ...base, - type: "todo_list", - planId: PlanId.make("plan:turn-item-compaction"), - steps: [{ id: "step", text: growing, status: completed ? "completed" : "running" }], - explanation: growing, - }; - break; - } - return { - id: EventId.make(eventId), - type: "turn-item.updated", - threadId, - occurredAt: now, - payload, - }; -} - -function subagentEvents( - threadId: ThreadId, - now: DateTime.Utc, - revision: number, -): ReadonlyArray { - const id = NodeId.make("subagent:turn-item-compaction"); - const prompt = revision === 0 ? "Review prompt A. ".repeat(40) : "Review prompt B. ".repeat(40); - const progress = `Review progress ${revision}`; - return [ - { - id: EventId.make(`event:turn-item-compaction:subagent:${revision}`), - type: "subagent.updated", - threadId, - occurredAt: now, - payload: { - id, - threadId, - runId: null, - parentNodeId: NodeId.make("node:turn-item-compaction-parent"), - origin: "provider_native", - createdBy: "agent", - driver: providerDriver, - providerInstanceId, - providerThreadId: null, - childThreadId: null, - nativeTaskRef: { - driver: providerDriver, - nativeId: "compaction-review", - strength: "strong", - }, - prompt, - progress, - title: "Review", - model: null, - status: "running", - result: null, - startedAt: now, - completedAt: null, - updatedAt: now, - }, - }, - { - id: EventId.make(`event:turn-item-compaction:subagent-item:${revision}`), - type: "turn-item.updated", - threadId, - occurredAt: now, - payload: { - id: TurnItemId.make("turn-item:compaction:subagent"), - threadId, - runId: null, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 50, - status: "running", - title: "Review", - startedAt: now, - completedAt: null, - updatedAt: now, - type: "subagent", - subagentId: id, - origin: "provider_native", - driver: providerDriver, - providerInstanceId, - childThreadId: null, - prompt, - progress, - result: null, - }, - }, - ] satisfies ReadonlyArray; -} - -it.effect( - "compacts ordinary snapshots while preserving replay, prompts, command lookup, and positions", - () => - Effect.gen(function* () { - const sink = yield* EventSink.EventSinkV2; - const eventStore = yield* EventStore.EventStoreV2; - const projections = yield* ProjectionStore.ProjectionStoreV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; - const sql = yield* SqlClient.SqlClient; - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:turn-item-compaction"); - const thread = makeThread(threadId, now); - const writes = [...(yield* sink.write({ events: [threadCreated(thread, now)] }))]; - const commandId = CommandId.make("command:turn-item-compaction:middle"); - const snapshotsPerItem = 70; - for (let revision = 0; revision < snapshotsPerItem; revision += 1) { - const order = - revision === 0 - ? [itemSpecs[0], itemSpecs[1], itemSpecs[2], itemSpecs[3]] - : [itemSpecs[1], itemSpecs[0], itemSpecs[2], itemSpecs[3]]; - const events = order - .filter((spec) => spec.key !== "commanded" || revision !== 35) - .map((spec) => itemEvent(threadId, now, spec, revision)); - if (revision < 3) events.push(...subagentEvents(threadId, now, revision)); - const stored = yield* sink.write({ events }); - writes.push(...stored); - if (revision === 35) { - const middle = itemEvent( - threadId, - now, - itemSpecs[3], - revision, - "event:turn-item-compaction:commanded:middle", - ); - writes.push(...(yield* sink.write({ commandId, events: [middle] }))); - } - } - const beforeProjection = yield* projections.getThreadProjection(threadId); - const highWater = yield* eventStore.latestSequence(); - const bytes = yield* sql<{ readonly rows: number; readonly payload_bytes: number }>` - SELECT COUNT(*) AS rows, COALESCE(SUM(octet_length(payload_json)), 0) AS payload_bytes - FROM orchestration_events WHERE stream_id = ${threadId} - `; - const summary = yield* maintenance.compactEventStore(); - const replay = yield* eventStore.read({ threadId }).pipe(Stream.runCollect); - const expectedIds = new Set([ - "event:turn-item-compaction:thread", - ...itemSpecs.flatMap((spec) => [ - `event:turn-item-compaction:${spec.key}:0`, - `event:turn-item-compaction:${spec.key}:${snapshotsPerItem - 1}`, - ]), - "event:turn-item-compaction:commanded:middle", - ...Array.from({ length: 3 }, (_, revision) => [ - `event:turn-item-compaction:subagent:${revision}`, - `event:turn-item-compaction:subagent-item:${revision}`, - ]).flat(), - ]); - assert.deepEqual( - replay, - writes.filter((stored) => expectedIds.has(stored.event.id)), - ); - assert.lengthOf(replay, 16); - assert.equal(summary.deletedEventCount, bytes[0]!.rows - replay.length); - assert.isBelow( - (yield* sql<{ readonly payload_bytes: number }>` - SELECT COALESCE(SUM(octet_length(payload_json)), 0) AS payload_bytes - FROM orchestration_events WHERE stream_id = ${threadId} - `)[0]!.payload_bytes, - bytes[0]!.payload_bytes, - ); - assert.equal(yield* eventStore.latestSequence(), highWater); - - const commanded = yield* sink.readByCommandId({ commandId }).pipe(Stream.runCollect); - assert.deepEqual( - commanded.map((stored) => stored.event.id), - [EventId.make("event:turn-item-compaction:commanded:middle")], - ); - const storedPrompts = yield* sql<{ - readonly event_id: string; - readonly sequence: number; - readonly prompt: string; - readonly prompt_sequence: number | null; - }>` - SELECT event_id, sequence, - json_extract(payload_json, '$.prompt') AS prompt, - json_extract(payload_json, '$.promptSequence') AS prompt_sequence - FROM orchestration_events - WHERE stream_id = ${threadId} AND event_type = 'subagent.updated' - ORDER BY sequence ASC - `; - assert.lengthOf(storedPrompts, 3); - assert.equal(storedPrompts[0]!.prompt_sequence, null); - assert.equal(storedPrompts[1]!.prompt_sequence, null); - assert.equal(storedPrompts[2]!.prompt, ""); - assert.equal(storedPrompts[2]!.prompt_sequence, storedPrompts[1]!.sequence); - - const deletedCursor = writes.find( - (stored) => stored.event.id === "event:turn-item-compaction:command:35", - )!.sequence; - assert.isFalse(replay.some((stored) => stored.sequence === deletedCursor)); - const cursors = [0, deletedCursor, highWater]; - for (const cursor of cursors) { - const cursorReplay = yield* eventStore - .read({ threadId, afterSequence: cursor, throughSequence: highWater }) - .pipe(Stream.runCollect); - const expected = replay.filter((stored) => stored.sequence > cursor); - assert.deepEqual( - cursorReplay.map((stored) => stored.sequence), - expected.map((stored) => stored.sequence), - ); - - let clientProjection: typeof beforeProjection | null = { - ...beforeProjection, - subagents: [], - turnItems: [], - visibleTurnItems: [], - }; - let latestLocalTurnOrdinal: number | null = null; - for (const stored of [ - ...writes.filter((entry) => entry.sequence <= cursor), - ...cursorReplay, - ]) { - clientProjection = applyOrchestrationV2ProjectionEvent(clientProjection, stored.event, { - partialTimeline: true, - latestLocalTurnOrdinal, - }); - if (stored.event.type === "turn-item.updated") { - latestLocalTurnOrdinal = Math.max( - latestLocalTurnOrdinal ?? 0, - stored.event.payload.ordinal, - ); - } - } - assert.deepEqual(clientProjection?.turnItems, beforeProjection.turnItems); - assert.deepEqual(clientProjection?.visibleTurnItems, beforeProjection.visibleTurnItems); - } - assert.isTrue((yield* maintenance.verify).valid); - assert.isTrue((yield* maintenance.rebuild).valid); - assert.deepEqual(yield* projections.getThreadProjection(threadId), beforeProjection); - const positions = yield* sql<{ - readonly turn_item_id: string; - readonly ordinal: number; - }>` - SELECT turn_item_id, ordinal - FROM orchestration_v2_turn_item_positions - WHERE thread_id = ${threadId} - ORDER BY ordinal ASC - `; - assert.deepEqual( - positions.map((row) => [row.turn_item_id, row.ordinal]), - [ - ...itemSpecs.map((spec, index) => [`turn-item:compaction:${spec.key}`, index + 1]), - ["turn-item:compaction:subagent", 5], - ], - ); - }).pipe(Effect.provide(layerTest)), -); - -it.effect("keeps post-bound appends visible during compaction", () => - Effect.scoped( - Effect.gen(function* () { - const sink = yield* EventSink.EventSinkV2; - const eventStore = yield* EventStore.EventStoreV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; - const now = yield* DateTime.now; - const threadId = ThreadId.make("thread:turn-item-compaction-race"); - const thread = makeThread(threadId, now); - const initial = yield* sink.write({ - events: [ - threadCreated(thread, now), - ...[0, 1, 2].map((revision) => itemEvent(threadId, now, itemSpecs[0], revision)), - ], - }); - const beforeSequence = initial.at(-1)!.sequence; - - const streamReady = yield* Deferred.make(); - const streamTracer = Tracer.make({ - span(options) { - const span = new Tracer.NativeSpan(options); - const end = span.end.bind(span); - span.end = (endTime, exit) => { - end(endTime, exit); - const query = span.attributes.get("db.query.text"); - if (typeof query === "string" && query.includes("MAX(sequence)")) { - Deferred.doneUnsafe(streamReady, Effect.void); - } - }; - return span; - }, - }); - const liveEvents = yield* sink - .stream({ threadId, afterSequence: beforeSequence }) - .pipe( - Stream.take(3), - Stream.runCollect, - Effect.withTracer(streamTracer), - Effect.forkScoped, - ); - yield* Deferred.await(streamReady); - - const scanStarted = yield* Deferred.make(); - const releaseScan = yield* Deferred.make(); - let blocked = false; - const blockFirstScanPage: Statement.Transformer = (statement) => { - const [query] = statement.compile(); - if ( - !blocked && - query.includes("FROM orchestration_events AS event") && - query.includes("ORDER BY event.sequence DESC") - ) { - blocked = true; - return Deferred.succeed(scanStarted, undefined).pipe( - Effect.andThen(Deferred.await(releaseScan)), - Effect.as(statement), - ); - } - return Effect.succeed(statement); - }; - const compaction = yield* maintenance - .compactEventStore() - .pipe( - Effect.provideService(Statement.CurrentTransformer, blockFirstScanPage), - Effect.forkScoped, - ); - yield* Deferred.await(scanStarted); - const snapshots = [3, 4, 5].map((revision) => - itemEvent( - threadId, - now, - itemSpecs[0], - revision, - `event:turn-item-compaction-race:${revision}`, - ), - ); - const appended = yield* sink.write({ events: snapshots }); - yield* Deferred.succeed(releaseScan, undefined); - assert.equal((yield* Fiber.join(compaction)).deletedEventCount, 1); - const streamed = yield* Fiber.join(liveEvents); - assert.deepEqual( - streamed.map((stored) => stored.sequence), - appended.map((stored) => stored.sequence), - ); - assert.deepEqual( - streamed.map((stored) => stored.event.id), - snapshots.map((event) => event.id), - ); - assert.equal(yield* eventStore.latestSequence(), appended.at(-1)!.sequence); - const firstPassRows = yield* eventStore - .read({ threadId, eventType: "turn-item.updated" }) - .pipe(Stream.runCollect); - assert.deepEqual( - firstPassRows.map((stored) => stored.sequence), - [initial[1]!.sequence, beforeSequence, ...appended.map((stored) => stored.sequence)], - ); - yield* maintenance.compactEventStore(); - const secondPassRows = yield* eventStore - .read({ threadId, eventType: "turn-item.updated" }) - .pipe(Stream.runCollect); - assert.deepEqual( - secondPassRows.map((stored) => stored.sequence), - [initial[1]!.sequence, appended.at(-1)!.sequence], - ); - }), - ).pipe(Effect.provide(layerTest)), -); diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1Cutover.integration.test.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1Cutover.integration.test.ts index 5d2343f6a1ea6..83c34c7edab17 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1Cutover.integration.test.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1Cutover.integration.test.ts @@ -42,7 +42,7 @@ import * as EventSink from "../EventSink.ts"; import * as EventStore from "../EventStore.ts"; import * as LegacyV1ThreadImporter from "./LegacyV1ThreadImporter.ts"; import * as Orchestrator from "../Orchestrator.ts"; -import * as ProjectionMaintenance from "../ProjectionMaintenance.ts"; +import * as ProjectionReplay from "../testkit/ProjectionReplay.ts"; import * as ProjectionStore from "../ProjectionStore.ts"; import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts"; import * as ProviderReplayHarness from "../testkit/ProviderReplayHarness.ts"; @@ -600,9 +600,7 @@ const layerBoot = (input: { const layerImporterProvided = LegacyV1ThreadImporter.layer.pipe( Layer.provide(Layer.mergeAll(layerStoresProvided, layerEventSinkProvided)), ); - const layerMaintenanceProvided = ProjectionMaintenance.layer.pipe( - Layer.provide(layerStoresProvided), - ); + const layerMaintenanceProvided = ProjectionReplay.layer.pipe(Layer.provide(layerStoresProvided)); const layerOrchestratorProvided = ProviderReplayHarness.layerWithRegistry( { name: input.name, @@ -676,7 +674,7 @@ describe("orchestration v2 legacy v1 cutover", () => { Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const projections = yield* ProjectionStore.ProjectionStoreV2; const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts index 87059b7c1d0ae..50671194d46d2 100644 --- a/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts +++ b/apps/server/src/orchestration-v2/legacy/LegacyV1ThreadImporter.test.ts @@ -19,7 +19,7 @@ import { listLinkedPullRequestThreads } from "../../pullRequest/linkedThreads.ts import * as EventSink from "../EventSink.ts"; import * as EventStore from "../EventStore.ts"; import * as LegacyV1ThreadImporter from "./LegacyV1ThreadImporter.ts"; -import * as ProjectionMaintenance from "../ProjectionMaintenance.ts"; +import * as ProjectionReplay from "../testkit/ProjectionReplay.ts"; import * as ProjectionStore from "../ProjectionStore.ts"; const layerDatabase = SqlitePersistence.layerMemory; @@ -34,14 +34,14 @@ const layerEventSinkProvided = EventSink.layer.pipe(Layer.provide(layerStoresPro const layerImporterProvided = LegacyV1ThreadImporter.layer.pipe( Layer.provide(Layer.mergeAll(layerStoresProvided, layerEventSinkProvided)), ); -const layerProjectionMaintenanceProvided = ProjectionMaintenance.layer.pipe( +const layerProjectionReplayProvided = ProjectionReplay.layer.pipe( Layer.provide(layerStoresProvided), ); const layerTest = Layer.mergeAll( layerStoresProvided, layerEventSinkProvided, layerImporterProvided, - layerProjectionMaintenanceProvided, + layerProjectionReplayProvided, ); const seedPendingTranscripts = Effect.gen(function* () { @@ -220,7 +220,7 @@ it.layer(layerTest)("LegacyV1ThreadImporter", (it) => { Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const projections = yield* ProjectionStore.ProjectionStoreV2; const eventSink = yield* EventSink.EventSinkV2; const threadId = ThreadId.make("thread:legacy-import"); @@ -605,7 +605,7 @@ it.layer(layerTest)("LegacyV1ThreadImporter", (it) => { Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const projections = yield* ProjectionStore.ProjectionStoreV2; const eventSink = yield* EventSink.EventSinkV2; const threadId = ThreadId.make("thread:legacy-metadata-upgrade"); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index b38fb737c91db..34d252457e63a 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -68,7 +68,8 @@ import * as EffectWorker from "./EffectWorker.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; import * as EventSink from "./EventSink.ts"; import * as ProviderRuntimeRecoveryService from "./ProviderRuntimeRecoveryService.ts"; -import * as ProjectionMaintenance from "./ProjectionMaintenance.ts"; +import * as EventStore from "./EventStore.ts"; +import * as ProjectionReplay from "./testkit/ProjectionReplay.ts"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as PullRequestWatchReactor from "./PullRequestWatchReactor.ts"; import { PULL_REQUEST_WATCH_WAKE_LIMIT } from "./pullRequestWatch.ts"; @@ -277,8 +278,21 @@ const moveProject = (projectId: ProjectId, workspaceRoot: string, updatedAt: str }), ); +// Read-only oracle: projections still equal a replay of these unpruned test logs. +const layerProjectionReplayTest = ProjectionReplay.layer.pipe( + Layer.provide( + Layer.merge( + EventStore.layerFromOrchestrationEventStore.pipe( + Layer.provide(RuntimeLayer.layerEventInfrastructure), + ), + ProjectionStore.layer, + ), + ), +); + const layerTest = Layer.mergeAll( RuntimeLayer.layer, + layerProjectionReplayTest, RuntimeLayer.layerEventSink, ProjectStore.layer, ProjectionStore.layer, @@ -297,7 +311,7 @@ const layerTest = Layer.mergeAll( Layer.provide(layerPlatformTest), ); -const layerLegacyImportTest = RuntimeLayer.layer.pipe( +const layerLegacyImportTest = Layer.merge(RuntimeLayer.layer, layerProjectionReplayTest).pipe( Layer.provide(McpSessionRegistryTestkit.layer), Layer.provide(McpProviderSessions.layer), Layer.provideMerge(SqlitePersistence.layerMemory), @@ -817,6 +831,7 @@ it.layer(layerProjectDeletionTest)("project deletion during thread commands", (i }); const layerSharedApplicationDataPlaneTest = Layer.mergeAll( + layerProjectionReplayTest, RuntimeLayer.layer.pipe(Layer.provide(RuntimeLayer.layerProjectService)), RuntimeLayer.layerProjectService, RuntimeLayer.layerEventSink, @@ -2210,7 +2225,7 @@ it.layer(layerLegacyImportTest)("OrchestrationV2 legacy import", (it) => { Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const orchestrator = yield* Orchestrator.OrchestratorV2; const threadManagement = yield* ThreadManagementService.ThreadManagementService; const metadataThreadId = ThreadId.make("runtime-layer-legacy-metadata-thread"); @@ -2725,7 +2740,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect("keeps the branch pull request when linking another pull request", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const threadId = ThreadId.make("branch-pr-link"); const projectId = ProjectId.make("branch-pr-project"); yield* seedProject({ @@ -2834,7 +2849,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect("retains multiple pull requests and dismissed stack members through rebuilds", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const threadId = ThreadId.make("runtime-multiple-pull-requests"); yield* orchestrator.dispatch({ type: "thread.create", @@ -2925,7 +2940,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect("starts, records, and stops a pull request watch", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const threadId = ThreadId.make("runtime-pull-request-watch"); yield* orchestrator.dispatch({ type: "thread.create", @@ -3017,7 +3032,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect.each(["manual", "automatic"])("settling ends every pull request watch: %s", (mode) => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const threadId = ThreadId.make(`pr-watch-settle-${mode}`); yield* orchestrator.dispatch({ type: "thread.create", @@ -4753,7 +4768,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect("cancels preparing work on archive and keeps it cancelled through replay", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const threadId = ThreadId.make("runtime-layer-archive-preparing-thread"); yield* orchestrator.dispatch({ type: "thread.create", @@ -6321,7 +6336,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { assert.equal(queued.length, 2); yield* recovery.reconcile(trigger); yield* recovery.reconcile("startup"); - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; assert.isTrue((yield* maintenance.rebuild).valid); assert.equal(yield* orchestrator.resumeQueuedRuns, 0); const held = yield* orchestrator.getThreadProjection(threadId); @@ -7547,11 +7562,11 @@ it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("thread opens", (it) ); }); -it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", (it) => { +it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("PR link events", (it) => { it.effect("persists only the changed PR link and replays it without losing other links", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const sql = yield* SqlClient.SqlClient; const threadId = ThreadId.make("compact-pr-sync"); yield* orchestrator.dispatch({ @@ -7650,85 +7665,6 @@ it.layer(Layer.fresh(layerSharedApplicationDataPlaneTest))("compact PR events", (yield* orchestrator.getThreadShell(threadId))?.pullRequests, after?.pullRequests, ); - for (const [commandId, title] of [ - ["compact-pr-sync-again", "Still open PR"], - ["compact-pr-sync-merged", "Merged PR"], - ] as const) { - yield* orchestrator.dispatch({ - type: "thread.pull-request-link.sync", - commandId: CommandId.make(commandId), - threadId, - ...key, - number: 2, - snapshot: { - state: title === "Merged PR" ? "merged" : "open", - title, - headBranch: "feature/pr", - baseBranch: "main", - isDraft: false, - updatedAt: null, - syncedAt: "2026-10-13T13:00:00.000Z", - }, - stack: null, - }); - } - const latest = yield* orchestrator.getThreadShell(threadId); - yield* maintenance.compactEventStore(); - const linkEvents = yield* sql<{ command_id: string }>` - SELECT command_id FROM orchestration_events - WHERE stream_id = ${threadId} AND event_type = 'thread.pull-request-link-updated' - ORDER BY sequence - `; - assert.deepEqual( - linkEvents.map((row) => row.command_id), - ["compact-pr-sync", "compact-pr-sync-merged"], - ); - assert.isTrue((yield* maintenance.rebuild).valid); - assert.deepEqual( - (yield* orchestrator.getThreadShell(threadId))?.pullRequests, - latest?.pullRequests, - ); - - // Visits preserve updatedAt, so compaction must retain earlier renames; later renames supersede visits. - const threadEvents = () => sql<{ event_type: string }>` - SELECT event_type FROM orchestration_events - WHERE stream_id = ${threadId} - AND event_type IN ('thread.metadata-updated', 'thread.visited') - ORDER BY sequence - `; - yield* orchestrator.dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make("compact-pr-rename"), - threadId, - title: "Renamed before visit", - }); - for (const [commandId, visitedAt] of [ - ["compact-pr-visit", "2026-10-13T14:00:00.000Z"], - ["compact-pr-visit-again", "2026-10-13T15:00:00.000Z"], - ] as const) { - yield* orchestrator.dispatch({ - type: "thread.visit", - commandId: CommandId.make(commandId), - threadId, - visitedAt, - }); - } - yield* maintenance.compactEventStore(); - assert.deepEqual( - (yield* threadEvents()).map((row) => row.event_type), - ["thread.metadata-updated", "thread.visited"], - ); - yield* orchestrator.dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make("compact-pr-rename-again"), - threadId, - title: "Renamed after visit", - }); - yield* maintenance.compactEventStore(); - assert.deepEqual( - (yield* threadEvents()).map((row) => row.event_type), - ["thread.metadata-updated"], - ); }), ); }); diff --git a/apps/server/src/orchestration-v2/testkit/ProjectionReplay.ts b/apps/server/src/orchestration-v2/testkit/ProjectionReplay.ts new file mode 100644 index 0000000000000..437db15c375f8 --- /dev/null +++ b/apps/server/src/orchestration-v2/testkit/ProjectionReplay.ts @@ -0,0 +1,182 @@ +import { ThreadId } from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as EventStore from "../EventStore.ts"; +import * as ProjectionStore from "../ProjectionStore.ts"; + +export interface ProjectionReplayVerification { + readonly valid: boolean; + readonly schemaVersion: number; + readonly expectedSequence: number; + readonly projectionSequence: number; + readonly unreadableThreadIds: ReadonlyArray; + readonly missingThreadIds: ReadonlyArray; + readonly unexpectedThreadIds: ReadonlyArray; +} + +export class ProjectionReplayError extends Schema.TaggedError()( + "ProjectionReplayError", + { cause: Schema.Defect() }, +) {} + +/** + * Test oracle: projections must equal a replay of an unpruned event log. Production prunes + * events the projections cover, so it never replays (see ProjectionMaintenance). + */ +export class ProjectionReplay extends Context.Service< + ProjectionReplay, + { + readonly verify: Effect.Effect; + readonly rebuild: Effect.Effect; + } +>()("t3/orchestration-v2/testkit/ProjectionReplay") {} + +type ProjectionMetadataRow = { + readonly schema_version: number; + readonly last_sequence: number; +}; + +export const layer: Layer.Layer< + ProjectionReplay, + never, + EventStore.EventStoreV2 | ProjectionStore.ProjectionStoreV2 | SqlClient.SqlClient +> = Layer.effect( + ProjectionReplay, + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const eventStore = yield* EventStore.EventStoreV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + + const verify = Effect.gen(function* () { + const expectedThreadRows = yield* sql<{ readonly thread_id: string }>` + SELECT DISTINCT stream_id AS thread_id + FROM orchestration_events INDEXED BY orchestration_events_v2_created_threads_idx + WHERE application_event_version = 2 + AND aggregate_kind = 'thread' + AND event_type = 'thread.created' + ORDER BY stream_id ASC + `; + const projectionRows = yield* sql<{ readonly thread_id: string }>` + SELECT thread_id + FROM orchestration_v2_projection_threads + ORDER BY thread_id ASC + `; + const actualIds = projectionRows.map((row) => ThreadId.make(row.thread_id)); + const expectedIds = expectedThreadRows.map((row) => ThreadId.make(row.thread_id)); + const actualSet = new Set(actualIds); + const expectedSet = new Set(expectedIds); + const missingThreadIds = expectedIds.filter((threadId) => !actualSet.has(threadId)); + const unexpectedThreadIds = actualIds.filter((threadId) => !expectedSet.has(threadId)); + const unreadableThreadIds = yield* projectionStore.getUnreadableThreadIds(); + const metadata = yield* sql` + SELECT schema_version, last_sequence + FROM orchestration_v2_projection_metadata + WHERE projection_name = 'thread-projections' + LIMIT 1 + `; + const expectedSequence = yield* eventStore.latestSequence(); + const schemaVersion = metadata[0]?.schema_version ?? 0; + const projectionSequence = metadata[0]?.last_sequence ?? 0; + return { + valid: + schemaVersion === ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION && + projectionSequence === expectedSequence && + missingThreadIds.length === 0 && + unexpectedThreadIds.length === 0 && + unreadableThreadIds.length === 0, + schemaVersion, + expectedSequence, + projectionSequence, + unreadableThreadIds, + missingThreadIds, + unexpectedThreadIds, + } satisfies ProjectionReplayVerification; + }).pipe(sql.withTransaction); + + const rebuild = Effect.gen(function* () { + yield* sql.withTransaction( + Effect.gen(function* () { + const throughSequence = yield* eventStore.latestSequence(); + yield* sql`DELETE FROM orchestration_v2_projection_context_transfers`; + yield* sql`DELETE FROM orchestration_v2_projection_context_handoffs`; + yield* sql`DELETE FROM orchestration_v2_projection_checkpoints`; + yield* sql`DELETE FROM orchestration_v2_projection_checkpoint_scopes`; + yield* sql`DELETE FROM orchestration_v2_projection_turn_items`; + yield* sql`DELETE FROM orchestration_v2_projection_plans`; + yield* sql`DELETE FROM orchestration_v2_projection_messages`; + yield* sql`DELETE FROM orchestration_v2_projection_runtime_requests`; + yield* sql`DELETE FROM orchestration_v2_projection_provider_turns`; + yield* sql`DELETE FROM orchestration_v2_projection_provider_threads`; + yield* sql`DELETE FROM orchestration_v2_projection_provider_session_bindings`; + yield* sql`DELETE FROM orchestration_v2_projection_provider_sessions`; + yield* sql`DELETE FROM orchestration_v2_projection_subagents`; + yield* sql`DELETE FROM orchestration_v2_projection_nodes`; + yield* sql`DELETE FROM orchestration_v2_projection_run_attempts`; + yield* sql`DELETE FROM orchestration_v2_projection_runs`; + yield* sql`DELETE FROM orchestration_v2_projection_threads`; + yield* sql`DELETE FROM orchestration_v2_turn_item_positions`; + + const pageSize = 500; + let lastSequence = 0; + while (true) { + const page = yield* eventStore + .read({ afterSequence: lastSequence, throughSequence, limit: pageSize }) + .pipe(Stream.runCollect); + for (const stored of page) { + yield* projectionStore.apply(stored.event); + if (stored.event.type === "turn-item.updated") { + yield* sql` + INSERT INTO orchestration_v2_turn_item_positions ( + thread_id, + turn_item_id, + ordinal + ) + VALUES ( + ${stored.event.threadId}, + ${stored.event.payload.id}, + ${stored.event.payload.ordinal} + ) + ON CONFLICT(thread_id, turn_item_id) DO UPDATE SET + ordinal = excluded.ordinal + `; + } + lastSequence = stored.sequence; + } + if (page.length < pageSize) break; + yield* Effect.yieldNow; + } + const now = DateTime.formatIso(yield* DateTime.now); + yield* sql` + INSERT INTO orchestration_v2_projection_metadata ( + projection_name, + schema_version, + last_sequence, + updated_at + ) + VALUES ( + 'thread-projections', + ${ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION}, + ${lastSequence}, + ${now} + ) + ON CONFLICT(projection_name) DO UPDATE SET + schema_version = excluded.schema_version, + last_sequence = excluded.last_sequence, + updated_at = excluded.updated_at + `; + }), + ); + return yield* verify; + }); + + const mapError = (effect: Effect.Effect) => + effect.pipe(Effect.mapError((cause) => new ProjectionReplayError({ cause }))); + return ProjectionReplay.of({ verify: mapError(verify), rebuild: mapError(rebuild) }); + }), +); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts index bb21d8ca9929a..cfedb73eeff74 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderSwitch.integration.test.ts @@ -52,7 +52,7 @@ import * as LegacyV1ThreadImporter from "../legacy/LegacyV1ThreadImporter.ts"; import * as Orchestrator from "../Orchestrator.ts"; import * as EffectWorker from "../EffectWorker.ts"; import * as EffectOutbox from "../EffectOutbox.ts"; -import * as ProjectionMaintenance from "../ProjectionMaintenance.ts"; +import * as ProjectionReplay from "./ProjectionReplay.ts"; import * as ProjectionStore from "../ProjectionStore.ts"; import * as ProviderAdapterRegistry from "../ProviderAdapterRegistry.ts"; import { makeProviderFailure } from "@t3tools/provider-core/server/failure"; @@ -3006,7 +3006,7 @@ describe("orchestration v2 provider switching", () => { const layerImporterProvided = LegacyV1ThreadImporter.layer.pipe( Layer.provide(Layer.mergeAll(layerStoresProvided, layerEventSinkProvided)), ); - const layerMaintenanceProvided = ProjectionMaintenance.layer.pipe( + const layerMaintenanceProvided = ProjectionReplay.layer.pipe( Layer.provide(layerStoresProvided), ); const layerOrchestratorProvided = ProviderReplayHarness.layerWithRegistry( @@ -3035,7 +3035,7 @@ describe("orchestration v2 provider switching", () => { const projection = yield* Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const orchestrator = yield* Orchestrator.OrchestratorV2; const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; diff --git a/apps/server/src/persistence/DatabaseMaintenance.test.ts b/apps/server/src/persistence/DatabaseMaintenance.test.ts index 381b143923adf..c767cebd0f886 100644 --- a/apps/server/src/persistence/DatabaseMaintenance.test.ts +++ b/apps/server/src/persistence/DatabaseMaintenance.test.ts @@ -12,9 +12,6 @@ import * as SqlClient from "effect/sql/SqlClient"; import * as DatabaseMaintenance from "./DatabaseMaintenance.ts"; import * as Sqlite from "./Sqlite.ts"; -import * as EventStore from "../orchestration-v2/EventStore.ts"; -import * as ProjectionMaintenance from "../orchestration-v2/ProjectionMaintenance.ts"; -import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; it.layer(DatabaseMaintenance.layer.pipe(Layer.provideMerge(NodeServices.layer)))( "DatabaseMaintenance", @@ -111,17 +108,8 @@ it.layer(DatabaseMaintenance.layer.pipe(Layer.provideMerge(NodeServices.layer))) const maintenance = yield* DatabaseMaintenance.DatabaseMaintenance; const baseDir = yield* fs.makeTempDirectoryScoped({ prefix: "t3-compact-" }); const dbPath = path.join(baseDir, "userdata", "statev2.sqlite"); - yield* Effect.flatMap( - ProjectionMaintenance.ProjectionMaintenanceV2, - (projections) => projections.rebuild, - ).pipe( - Effect.provide( - ProjectionMaintenance.layer.pipe( - Layer.provide(Layer.merge(EventStore.layer, ProjectionStore.layer)), - Layer.provideMerge(Sqlite.layerFromPath(dbPath)), - ), - ), - ); + // Opening the database migrates it. + yield* Effect.void.pipe(Effect.provide(Sqlite.layerFromPath(dbPath))); yield* Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; yield* sql`PRAGMA auto_vacuum = NONE`; diff --git a/apps/server/src/persistence/DatabaseMaintenance.ts b/apps/server/src/persistence/DatabaseMaintenance.ts index 4a97e6dcace98..d7876785c898d 100644 --- a/apps/server/src/persistence/DatabaseMaintenance.ts +++ b/apps/server/src/persistence/DatabaseMaintenance.ts @@ -9,9 +9,7 @@ import * as Schema from "effect/Schema"; import * as SqlClient from "effect/sql/SqlClient"; import { isSqlError } from "effect/sql/SqlError"; -import * as EventStore from "../orchestration-v2/EventStore.ts"; import * as ProjectionMaintenance from "../orchestration-v2/ProjectionMaintenance.ts"; -import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; import { rebuildMessageSearch } from "../orchestration-v2/messageSearchIndex.ts"; export class DatabaseMissingError extends Schema.TaggedError()( @@ -32,15 +30,6 @@ export class DatabaseInUseError extends Schema.TaggedError() } } -export class DatabaseProjectionsInvalidError extends Schema.TaggedError()( - "DatabaseProjectionsInvalidError", - { path: Schema.String }, -) { - override get message() { - return "Database projections need repair before compaction. Start T3 Code once so it rebuilds them, then run doctor again. If this CLI is older than the one that wrote the database, update T3 Code first."; - } -} - export class DatabaseInspectError extends Schema.TaggedError()( "DatabaseInspectError", { path: Schema.String, cause: Schema.Defect() }, @@ -89,10 +78,7 @@ export class DatabaseMaintenance extends Context.Service< readonly deletedEventCount: number; readonly deletedReceiptCount: number; }, - | DatabaseMissingError - | DatabaseInUseError - | DatabaseProjectionsInvalidError - | DatabaseCompactError + DatabaseMissingError | DatabaseInUseError | DatabaseCompactError >; /** Deletes only the pre-0.1.0 database; earlier versions still use it. */ readonly removeLegacyDatabase: ( @@ -101,9 +87,7 @@ export class DatabaseMaintenance extends Context.Service< } >()("t3/persistence/DatabaseMaintenance") {} -const isDomainError = Schema.is( - Schema.Union([DatabaseMissingError, DatabaseInUseError, DatabaseProjectionsInvalidError]), -); +const isDomainError = Schema.is(Schema.Union([DatabaseMissingError, DatabaseInUseError])); const readStorage = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; @@ -180,13 +164,7 @@ const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; yield* lockExclusively(databasePath); const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; - if (!(yield* maintenance.verify).valid) { - return yield* new DatabaseProjectionsInvalidError({ path: databasePath }); - } - const summary = yield* maintenance.compactEventStore(); - if (!(yield* maintenance.verify).valid) { - return yield* new DatabaseProjectionsInvalidError({ path: databasePath }); - } + const summary = yield* maintenance.pruneEventStore(); // VACUUM is required to enable incremental auto_vacuum on existing databases. yield* sql`PRAGMA auto_vacuum = INCREMENTAL`; yield* sql`VACUUM`; @@ -196,7 +174,6 @@ const make = Effect.gen(function* () { }).pipe( Effect.provide( ProjectionMaintenance.layer.pipe( - Layer.provide(Layer.merge(EventStore.layer, ProjectionStore.layer)), Layer.provideMerge(NodeSqliteClient.layer({ filename: databasePath })), ), ), diff --git a/apps/server/src/persistence/DatabaseStorage.test.ts b/apps/server/src/persistence/DatabaseStorage.test.ts index 878f596600888..73a0d4cac7cd8 100644 --- a/apps/server/src/persistence/DatabaseStorage.test.ts +++ b/apps/server/src/persistence/DatabaseStorage.test.ts @@ -9,9 +9,7 @@ import * as TestClock from "effect/testing/TestClock"; import * as SqlClient from "effect/sql/SqlClient"; import * as ServerConfig from "../config.ts"; -import * as EventStore from "../orchestration-v2/EventStore.ts"; import * as ProjectionMaintenance from "../orchestration-v2/ProjectionMaintenance.ts"; -import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; import * as DatabaseStorage from "./DatabaseStorage.ts"; import * as Sqlite from "./Sqlite.ts"; @@ -28,7 +26,6 @@ const storageLayer = ( ) => DatabaseStorage.layer.pipe( Layer.provide(ProjectionMaintenance.layer), - Layer.provide(Layer.merge(EventStore.layer, ProjectionStore.layer)), Layer.provideMerge(Sqlite.layerMemory), Layer.provideMerge(config), Layer.provide(NodeServices.layer), @@ -54,14 +51,12 @@ it.effect("a new database returns freed pages online without losing data", () => ); const compactionLayer = ( - compactEventStore: ProjectionMaintenance.ProjectionMaintenanceV2["Service"]["compactEventStore"], + pruneEventStore: ProjectionMaintenance.ProjectionMaintenanceV2["Service"]["pruneEventStore"], ) => DatabaseStorage.layer.pipe( Layer.provide( Layer.succeed(ProjectionMaintenance.ProjectionMaintenanceV2, { - compactEventStore, - verify: Effect.die("unexpected verification"), - rebuild: Effect.die("unexpected rebuild"), + pruneEventStore, }), ), Layer.provide(Sqlite.layerMemory), diff --git a/apps/server/src/persistence/DatabaseStorage.ts b/apps/server/src/persistence/DatabaseStorage.ts index bbb4dd1399efd..a0dd307237d46 100644 --- a/apps/server/src/persistence/DatabaseStorage.ts +++ b/apps/server/src/persistence/DatabaseStorage.ts @@ -31,7 +31,6 @@ export class DatabaseStorage extends Context.Service< } >()("t3/persistence/DatabaseStorage") {} -// Deleted events have newer retained replacements; old cursors replay to the same state. const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; @@ -108,7 +107,7 @@ const make = Effect.gen(function* () { yield* Effect.sleep((Number(elapsed) / 1_000_000) * 9); sliceStarted = yield* Clock.monotonicTimeNanos; }); - const result = yield* maintenance.compactEventStore(betweenPages); + const result = yield* maintenance.pruneEventStore(betweenPages); const completedAt = yield* Clock.currentTimeMillis; yield* fs.writeFileString(compactedAtPath, String(completedAt)).pipe(Effect.ignore); if (result.deletedEventCount > 0 || result.deletedReceiptCount > 0) { diff --git a/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts b/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts index 69eb8eff57027..b53309c187ed3 100644 --- a/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts +++ b/apps/server/src/persistence/Migrations/064_CanonicalizeThreadPullRequestLinks.test.ts @@ -9,7 +9,7 @@ import * as SqlClient from "effect/sql/SqlClient"; import * as EventStore from "../../orchestration-v2/EventStore.ts"; import * as ProjectionStore from "../../orchestration-v2/ProjectionStore.ts"; -import * as ProjectionMaintenance from "../../orchestration-v2/ProjectionMaintenance.ts"; +import * as ProjectionReplay from "../../orchestration-v2/testkit/ProjectionReplay.ts"; import { listLinkedPullRequestThreads } from "../../pullRequest/linkedThreads.ts"; import { runMigrations } from "../Migrations.ts"; @@ -53,7 +53,7 @@ const thread = (id: string) => ({ }); const layer = it.layer( - ProjectionMaintenance.layer.pipe( + ProjectionReplay.layer.pipe( Layer.provideMerge( Layer.mergeAll(EventStore.layer, ProjectionStore.layer).pipe( Layer.provideMerge(NodeSqliteClient.layer({ filename: ":memory:" })), @@ -156,7 +156,7 @@ layer("064_CanonicalizeThreadPullRequestLinks", (it) => { `; assert.equal(v1?.payload_json, v1Payload); - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; assert.isTrue((yield* maintenance.rebuild).valid); for (const [host, repository, id] of [ ["github.com", "acme/web", "github"], diff --git a/apps/server/src/persistence/Migrations/066_RepairClientSettlementTimestamps.test.ts b/apps/server/src/persistence/Migrations/066_RepairClientSettlementTimestamps.test.ts index f30289cb92de9..d62654b37ece6 100644 --- a/apps/server/src/persistence/Migrations/066_RepairClientSettlementTimestamps.test.ts +++ b/apps/server/src/persistence/Migrations/066_RepairClientSettlementTimestamps.test.ts @@ -6,13 +6,13 @@ import * as Layer from "effect/Layer"; import * as SqlClient from "effect/sql/SqlClient"; import * as EventStore from "../../orchestration-v2/EventStore.ts"; -import * as ProjectionMaintenance from "../../orchestration-v2/ProjectionMaintenance.ts"; +import * as ProjectionReplay from "../../orchestration-v2/testkit/ProjectionReplay.ts"; import * as ProjectionStore from "../../orchestration-v2/ProjectionStore.ts"; import { runMigrations } from "../Migrations.ts"; import Migration0066 from "./066_RepairClientSettlementTimestamps.ts"; const layer = it.layer( - ProjectionMaintenance.layer.pipe( + ProjectionReplay.layer.pipe( Layer.provideMerge( Layer.mergeAll(EventStore.layer, ProjectionStore.layer).pipe( Layer.provideMerge(NodeSqliteClient.layer({ filename: ":memory:" })), @@ -363,7 +363,7 @@ layer("066_RepairClientSettlementTimestamps", (it) => { yield* Migration0066; assert.deepEqual(yield* snapshot, repaired); - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; assert.isTrue((yield* maintenance.rebuild).valid); assert.deepEqual(yield* imported, expectedImported); }), diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index dbea09d666988..e55d82713871b 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -33,6 +33,10 @@ import { type OrchestrationEventStoreError, } from "./Errors.ts"; +/** Projection metadata rows: events at or below the pruned-through sequence may be gone. */ +export const EVENT_LOG_PRUNED_THROUGH = "event-log-pruned-through"; +export const EVENT_LOG_PRUNE_MARK = "event-log-prune-mark"; + export type UnsequencedProjectEvent = ApplicationProjectEvent extends infer Event ? Event extends ApplicationProjectEvent ? Omit @@ -93,6 +97,9 @@ export class OrchestrationEventStore extends Context.Service< readonly latestApplicationSequence: Effect.Effect; + /** Replay from a cursor below this misses pruned events; resync from a snapshot instead. */ + readonly prunedThroughSequence: Effect.Effect; + /** Unknown event types fail unless skipUnknownEventTypes logs and drops them; only client replay may opt in. */ readonly readApplicationEvents: (input: { readonly afterSequence: number; @@ -450,7 +457,7 @@ const makeEventStore = Effect.gen(function* () { const promptKey = subagent === undefined ? undefined : `${event.type}:${subagent.id}`; const hint = promptKey === undefined ? undefined : promptHints.get(promptKey); const payloadJson = yield* encodeJson(encoded.payload); - // Deduplicate repeated progress prompts via anchor rows that ProjectionMaintenance never deletes. + // Deduplicate repeated progress prompts via anchor rows; pruning keeps anchors that retained rows reference. const storedPayload = hint !== undefined && subagent !== undefined ? sql`CASE WHEN EXISTS ( @@ -672,6 +679,15 @@ const makeEventStore = Effect.gen(function* () { ), ); + const prunedThroughSequence = sql<{ readonly last_sequence: number }>` + SELECT last_sequence FROM orchestration_v2_projection_metadata + WHERE projection_name = ${EVENT_LOG_PRUNED_THROUGH} + `.pipe( + NodeSqliteClient.readOnly, + Effect.map((rows) => rows[0]?.last_sequence ?? 0), + Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.prunedThroughSequence:query")), + ); + // Advance over raw rows so even a full page of unknown types cannot end replay early. const catchUpApplicationEvents = (input: { readonly afterSequence: number; @@ -748,6 +764,7 @@ const makeEventStore = Effect.gen(function* () { getReplayStats, latestAgentSequence, latestApplicationSequence, + prunedThroughSequence, readApplicationEvents: catchUpApplicationEvents, publishCommitted: (events) => PubSub.publishAll(committedEvents, events).pipe(Effect.asVoid), streamApplicationEvents, diff --git a/apps/server/src/project/ProjectService.deletion.test.ts b/apps/server/src/project/ProjectService.deletion.test.ts index 41804842d5205..28b1c40541ddf 100644 --- a/apps/server/src/project/ProjectService.deletion.test.ts +++ b/apps/server/src/project/ProjectService.deletion.test.ts @@ -22,7 +22,7 @@ import * as EventSink from "../orchestration-v2/EventSink.ts"; import * as EventStore from "../orchestration-v2/EventStore.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as LegacyV1ThreadImporter from "../orchestration-v2/legacy/LegacyV1ThreadImporter.ts"; -import * as ProjectionMaintenance from "../orchestration-v2/ProjectionMaintenance.ts"; +import * as ProjectionReplay from "../orchestration-v2/testkit/ProjectionReplay.ts"; import * as ProjectionStore from "../orchestration-v2/ProjectionStore.ts"; import * as ProjectStore from "../orchestration-v2/ProjectStore.ts"; import * as ThreadCommandExecutor from "../orchestration-v2/ThreadCommandExecutor.ts"; @@ -40,7 +40,7 @@ const layerEventPersistence = EventSink.layer.pipe( ); const layerServices = Layer.mergeAll( LegacyV1ThreadImporter.layer.pipe(Layer.provideMerge(layerEventPersistence)), - ProjectionMaintenance.layer.pipe(Layer.provide(layerEventPersistence)), + ProjectionReplay.layer.pipe(Layer.provide(layerEventPersistence)), ProjectStore.layer, IdAllocator.layer, ThreadCommandExecutor.layer, @@ -310,7 +310,7 @@ it.effect( yield* Effect.gen(function* () { const importer = yield* LegacyV1ThreadImporter.LegacyV1ThreadImporter; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const maintenance = yield* ProjectionReplay.ProjectionReplay; const projections = yield* ProjectionStore.ProjectionStoreV2; const service = yield* ProjectService.make; assert.deepEqual(yield* importer.reconcileShells, { diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index a4433b59e8b77..395afcf56631a 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -198,7 +198,6 @@ import * as ThreadSettlementService from "./orchestration-v2/ThreadSettlementSer import * as IdleCompactionService from "./orchestration-v2/IdleCompactionService.ts"; import * as ThreadPullRequestService from "./orchestration-v2/ThreadPullRequestService.ts"; import * as RunFinalizationService from "./orchestration-v2/RunFinalizationService.ts"; -import * as EventStoreV2 from "./orchestration-v2/EventStore.ts"; import * as ProjectionMaintenance from "./orchestration-v2/ProjectionMaintenance.ts"; import * as ProjectionStoreV2 from "./orchestration-v2/ProjectionStore.ts"; import { makePersistedServerRuntimeState } from "./serverRuntimeState.ts"; @@ -629,12 +628,7 @@ const layerRuntimeCoreDependenciesBase = Layer.mergeAll( layerThreadSettlementWorker, layerIdleCompactionWorker, StorageCleanup.layer.pipe(Layer.provide(ProjectionStoreV2.layer)), - DatabaseStorage.layer.pipe( - Layer.provide(ProjectionMaintenance.layer), - Layer.provide( - Layer.merge(EventStoreV2.layerFromOrchestrationEventStore, ProjectionStoreV2.layer), - ), - ), + DatabaseStorage.layer.pipe(Layer.provide(ProjectionMaintenance.layer)), layerThreadPullRequestWorker, DelegatedTaskAnalytics.layer, Layer.effectDiscard( diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index c52f8fd917497..8ed7cb55580ee 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -22,6 +22,8 @@ import { import * as Deferred from "effect/Deferred"; import * as DateTime from "effect/DateTime"; import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Cause from "effect/Cause"; import * as Fiber from "effect/Fiber"; import * as Layer from "effect/Layer"; import * as PubSub from "effect/PubSub"; @@ -115,9 +117,6 @@ import * as Keybindings from "./keybindings.ts"; import * as ServerSettings from "./serverSettings.ts"; import * as ManagedPolicy from "./policy/ManagedPolicy.ts"; -import * as EventStore from "./orchestration-v2/EventStore.ts"; -import * as ProjectionMaintenance from "./orchestration-v2/ProjectionMaintenance.ts"; -import * as ProjectionStore from "./orchestration-v2/ProjectionStore.ts"; import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; @@ -1036,93 +1035,59 @@ it.effect("on-demand clients get provider config without workspace snapshots", ( ), ); -// Compaction keeps the oldest and newest event per message, so a cursor older -// than the deleted rows still replays each message where the client would -// first have seen it (reducers append on first sight) and reaches the same -// final state. No snapshot fallback is needed. -it.effect("thread resume across compacted history keeps message order and final state", () => - Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; - const now = "2026-10-13T00:00:00.000Z"; - const threadId = ThreadId.make("thread:compacted"); - const insertMessage = (version: number, id: string, text: string) => sql<{ - sequence: number; - }>` - INSERT INTO orchestration_events ( - event_id, aggregate_kind, stream_id, stream_version, event_type, - occurred_at, actor_kind, payload_json, metadata_json, application_event_version - ) VALUES ( - ${`compacted:${version}`}, 'thread', ${threadId}, ${version}, 'message.updated', - ${now}, 'server', ${JSON.stringify({ - createdBy: "user", - creationSource: "web", - id, - threadId, - runId: null, - nodeId: null, - role: "user", - text, - attachments: [], - streaming: false, - createdAt: now, - updatedAt: now, - })}, '{}', 2 - ) RETURNING sequence - `; - const [first] = yield* insertMessage(0, "message:a", "a created"); - const [second] = yield* insertMessage(1, "message:b", "b created"); - yield* insertMessage(2, "message:a", "a superseded"); - const [head] = yield* insertMessage(3, "message:a", "a final"); - // thread.deleted is never superseded: clients switch on the event type. - for (const version of [0, 1]) { - yield* sql` +// Pruned history cannot be replayed: a cursor below the boundary must resync from a snapshot. +it.effect.each(["thread", "shell"] as const)( + "%s resume from a cursor below the pruned boundary loads a snapshot", + (scope) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-10-13T00:00:00.000Z"; + const threadId = ThreadId.make("thread:pruned"); + const insert = (version: number) => sql<{ sequence: number }>` INSERT INTO orchestration_events ( event_id, aggregate_kind, stream_id, stream_version, event_type, occurred_at, actor_kind, payload_json, metadata_json, application_event_version ) VALUES ( - ${`deleted:${version}`}, 'thread', 'thread:deleted-twice', ${version}, 'thread.deleted', - ${now}, 'server', '{}', '{}', 2 - ) + ${`pruned:${version}`}, 'thread', ${threadId}, ${version}, 'provider-session.detached', + ${now}, 'server', ${JSON.stringify({ providerSessionId: "session:pruned", detachedAt: now })}, + '{}', 2 + ) RETURNING sequence `; - } - assert.equal((yield* maintenance.compactEventStore()).deletedEventCount, 1); - - const items = yield* subscribeOrchestrationV2Thread({ - threadId, - afterSequence: first!.sequence - 1, - requestCompletionMarker: true, - }).pipe( - Effect.flatMap((stream) => - stream.pipe( - Stream.takeUntil((item) => item.kind === "synchronized"), - Stream.runCollect, + const [first] = yield* insert(0); + const [second] = yield* insert(1); + yield* insert(2); + yield* sql` + INSERT INTO orchestration_v2_projection_metadata ( + projection_name, schema_version, last_sequence, updated_at + ) VALUES (${OrchestrationEventStore.EVENT_LOG_PRUNED_THROUGH}, 2, ${second!.sequence}, ${now}) + `; + const subscription = Effect.gen(function* () { + const stream = + scope === "thread" + ? yield* subscribeOrchestrationV2Thread({ threadId, afterSequence: first!.sequence }) + : yield* subscribeOrchestrationV2Shell({ afterSequence: first!.sequence }); + return stream; + }); + const exit = yield* subscription.pipe( + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + getThreadSnapshot: () => Effect.die("snapshot"), + listActiveShellIndex: () => Effect.die("snapshot"), + readShellSnapshot: () => Effect.die("snapshot"), + listActiveSubagentThreadIds: () => Effect.die("replay"), + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provideService(ProjectStore.ProjectStoreV2, { + listShells: () => Effect.succeed([]), + } as unknown as ProjectStore.ProjectStoreV2["Service"]), + Effect.provideService( + ProjectService.ProjectService, + {} as ProjectService.ProjectService["Service"], ), - ), - Effect.provideService(ThreadManagement.ThreadManagementService, { - ensureLegacyTranscript: () => Effect.void, - streamStoredEventsFrom: () => Stream.never, - } as unknown as ThreadManagement.ThreadManagementService["Service"]), - ); - assert.deepEqual( - Array.from(items, (item) => - item.kind === "event" - ? [item.sequence, item.event.type === "message.updated" ? item.event.payload.text : null] - : [item.kind], - ), - [ - [first!.sequence, "a created"], - [second!.sequence, "b created"], - [head!.sequence, "a final"], - ["synchronized"], - ], - ); - }).pipe( - Effect.provide( - ProjectionMaintenance.layer.pipe( - Layer.provide(Layer.merge(EventStore.layer, ProjectionStore.layer)), - Layer.provideMerge(ReplayTestLayer), - ), - ), - ), + Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, { + subscribeChanges: PubSub.subscribe(yield* PubSub.unbounded()), + } as ProjectEnrichment.ProjectEnrichmentService["Service"]), + Effect.exit, + ); + assert.deepEqual(Exit.isFailure(exit) ? Cause.squash(exit.cause) : exit, "snapshot"); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0f96fdc040a11..0e1bd74339d1c 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -783,6 +783,7 @@ const canReplayPersistedRange = Effect.fnUntraced(function* ( if (replayGap < 0 || replayGap > maxGap) { return false; } + if (afterSequence < (yield* applicationEvents.prunedThroughSequence)) return false; const stats = yield* applicationEvents.getReplayStats({ afterSequence, throughSequence: headSequence, @@ -971,7 +972,17 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs }), ), ); - if (input.afterSequence > highWater) { + const prunedThrough = yield* applicationEvents.prunedThroughSequence.pipe( + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: `Failed to prepare orchestration V2 thread ${input.threadId} replay`, + cause, + }), + ), + ); + if (input.afterSequence > highWater || input.afterSequence < prunedThrough) { return yield* snapshotThenLive(); } const stats = yield* applicationEvents diff --git a/docs/internals/overview.md b/docs/internals/overview.md index 9e5cf6c9d8a6c..2ae85b443b1c0 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -44,8 +44,8 @@ select known environment-local checkouts; the group itself does not store inheri ## Durable intent and side effects -The event log is the source of truth for orchestration state. The -[v2 orchestrator](../../apps/server/src/orchestration-v2/Orchestrator.ts) serializes commands and +Orchestration state is decided as events, and projections are the source of truth once they apply +them. The [v2 orchestrator](../../apps/server/src/orchestration-v2/Orchestrator.ts) serializes commands and decides events without performing provider or filesystem work. [EventSink](../../apps/server/src/orchestration-v2/EventSink.ts) commits events, persisted projections, the accepted command receipt, and outbox effects in one database transaction. Subscribers receive @@ -58,7 +58,17 @@ therefore means the intent committed, not that the provider, checkpoint, or othe finished. Keep external I/O out of command decisions and the database transaction. Effects tied to a lost provider process cannot simply replay; recovery retires them before admitting new work. -Persisted events must remain decodable on replay. Changing a schema affects old environments at +The event log only keeps what live readers need: about a day of history for reconnect replay and +command-id retries. +[ProjectionMaintenance](../../apps/server/src/orchestration-v2/ProjectionMaintenance.ts) deletes +older events the projections cover, so projections cannot be rebuilt from the log. It keeps each +stream's newest event (high-water marks, stream versions, snapshot sequences), `thread.created` +(the legacy importer's marker), the first `running` update of each run attempt (Codex usage +history), and prompt anchors that retained subagent rows reference. A client cursor below the +recorded pruned-through sequence resumes from a snapshot. Command receipts stay for idempotency; a +retry older than the retained history can find its events gone and must answer from projections. + +Retained events must remain decodable on replay. Changing a schema affects old environments at startup as well as live RPC traffic. Compatibility work must account for stored history, not just what the newest client sends. diff --git a/docs/user/background-service.md b/docs/user/background-service.md index c36527b2c425c..83974146c2cdf 100644 --- a/docs/user/background-service.md +++ b/docs/user/background-service.md @@ -161,8 +161,8 @@ For failures after signing in to T3 Connect, see ## Reclaim database space -T3 Code removes superseded internal history on its own while it runs; your -conversations are kept. Databases created before this shrank automatically +T3 Code removes internal history older than a day on its own while it runs; +your conversations are kept. Databases created before this shrank automatically still hold the freed space on disk until a one-time offline repair returns it and turns on automatic shrinking. Settings > Maintenance (web and desktop) or Settings > Maintenance (mobile) suggests the repair when at least 1 GiB can be From 0955847e6f138a012574634c0c9f818cc689f3ea Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 17:36:41 -0800 Subject: [PATCH 2/5] fix(server): idle threads resume without a snapshot; keep anchors of retained stream heads A thread idle since before the pruned boundary resumes at its own high-water, so it missed nothing; it no longer resends its full body on every reconnect. A stream's newest row is kept even below the boundary, so the prompt anchor it borrows is now kept too; deleting it left the row undecodable. Co-Authored-By: Claude Opus 5.5 --- .../ProjectionMaintenance.test.ts | 30 ++++++++++++++++ .../orchestration-v2/ProjectionMaintenance.ts | 14 ++++++++ apps/server/src/ws.test.ts | 35 +++++++++++++++++++ apps/server/src/ws.ts | 6 +++- docs/internals/overview.md | 7 ++-- 5 files changed, 88 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts index 50ed615f5b2f5..1f42953cc4ec3 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.test.ts @@ -302,6 +302,36 @@ it.effect("prunes events the projections cover once the retained day has passed" }).pipe(Effect.provide(layerTest)), ); +it.effect("keeps the anchor a retained stream head borrows from", () => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const eventStore = yield* EventStore.EventStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:prune-head"); + yield* sink.write({ events: [threadCreated(makeThread(threadId, now), now)] }); + yield* sink.write({ events: subagentEvents(threadId, now, 0) }); + // The stream's newest row borrows prompt A, so it is kept below the boundary. + yield* sink.write({ events: subagentEvents(threadId, now, 0, "repeat") }); + yield* maintenance.pruneEventStore(); + yield* TestClock.adjust("1 day"); + assert.isAbove((yield* maintenance.pruneEventStore()).deletedEventCount, 0); + + assert.include( + yield* remainingEventIds(threadId), + "event:turn-item-compaction:subagent-item:0", + ); + const stored = yield* eventStore.read({ threadId }).pipe(Stream.runCollect); + const head = Array.from(stored).at(-1)?.event; + assert.deepEqual( + head?.type === "turn-item.updated" && head.payload.type === "subagent" + ? head.payload.prompt + : null, + "Review prompt A. ".repeat(40), + ); + }).pipe(Effect.provide(layerTest)), +); + it.effect("keeps an anchor that a row appended during the pass borrows from", () => Effect.scoped( Effect.gen(function* () { diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 9f6f2596d443e..210322aed686f 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -100,6 +100,20 @@ export const layer: Layer.Layer, ) { const referenced = yield* readPromptReferences(throughSequence, head, betweenPages); + // Stream heads at or below the boundary are kept, so the anchors they borrow must stay too. + const keptHeadReferences = yield* sql<{ readonly prompt_sequence: number }>` + SELECT json_extract(event.payload_json, '$.promptSequence') AS prompt_sequence + FROM ( + SELECT MAX(sequence) AS sequence FROM orchestration_events + GROUP BY aggregate_kind, stream_id + ) AS heads + JOIN orchestration_events AS event ON event.sequence = heads.sequence + WHERE heads.sequence <= ${throughSequence} + AND event.event_type IN ('subagent.updated', 'turn-item.updated') + AND json_type(event.payload_json, '$.promptSequence') = 'integer' + `.pipe(NodeSqliteClient.readOnly); + for (const row of keptHeadReferences) referenced.add(row.prompt_sequence); + yield* betweenPages; let referencesThrough = head; let deleted = 0; let cursor = 0; diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 8ed7cb55580ee..a6b6e0851477f 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -1091,3 +1091,38 @@ it.effect.each(["thread", "shell"] as const)( assert.deepEqual(Exit.isFailure(exit) ? Cause.squash(exit.cause) : exit, "snapshot"); }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); + +// A thread idle since before the boundary missed nothing; resuming must not resend its body. +it.effect("thread resume at its own high-water below the pruned boundary skips the snapshot", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-10-13T00:00:00.000Z"; + const threadId = ThreadId.make("thread:idle"); + const [last] = yield* sql<{ sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + 'idle:0', 'thread', ${threadId}, 0, 'provider-session.detached', + ${now}, 'server', ${JSON.stringify({ providerSessionId: "session:idle", detachedAt: now })}, + '{}', 2 + ) RETURNING sequence + `; + yield* sql` + INSERT INTO orchestration_v2_projection_metadata ( + projection_name, schema_version, last_sequence, updated_at + ) VALUES (${OrchestrationEventStore.EVENT_LOG_PRUNED_THROUGH}, 2, ${last!.sequence + 10}, ${now}) + `; + const exit = yield* subscribeOrchestrationV2Thread({ + threadId, + afterSequence: last!.sequence, + }).pipe( + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + getThreadSnapshot: () => Effect.die("snapshot"), + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.exit, + ); + assert.notDeepEqual(Exit.isFailure(exit) ? Cause.squash(exit.cause) : "resumed", "snapshot"); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 0e1bd74339d1c..109900d8ad526 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -982,7 +982,11 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs }), ), ); - if (input.afterSequence > highWater || input.afterSequence < prunedThrough) { + // An idle thread's cursor sits at its own high-water: nothing was missed, so replay nothing. + if ( + input.afterSequence > highWater || + (input.afterSequence < highWater && input.afterSequence < prunedThrough) + ) { return yield* snapshotThenLive(); } const stats = yield* applicationEvents diff --git a/docs/internals/overview.md b/docs/internals/overview.md index 2ae85b443b1c0..c4a7def7553e9 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -63,9 +63,10 @@ command-id retries. [ProjectionMaintenance](../../apps/server/src/orchestration-v2/ProjectionMaintenance.ts) deletes older events the projections cover, so projections cannot be rebuilt from the log. It keeps each stream's newest event (high-water marks, stream versions, snapshot sequences), `thread.created` -(the legacy importer's marker), the first `running` update of each run attempt (Codex usage -history), and prompt anchors that retained subagent rows reference. A client cursor below the -recorded pruned-through sequence resumes from a snapshot. Command receipts stay for idempotency; a +(the legacy importer's marker), every `running` run update (Codex usage history reads the first +of each attempt), and prompt anchors that retained subagent rows reference. A client whose cursor +is below both its thread's newest event and the recorded pruned-through sequence missed pruned +events, so it resumes from a snapshot; an idle thread's cursor resumes without one. Command receipts stay for idempotency; a retry older than the retained history can find its events gone and must answer from projections. Retained events must remain decodable on replay. Changing a schema affects old environments at From e188f19585a2f3f9c34d2ad7cce5d256c6640ef8 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:02:36 -0800 Subject: [PATCH 3/5] fix(server): recheck the pruned boundary after reading a resume replay A prune pass can delete replayed rows after a resume passes the boundary check. Thread and shell resumes now read the replay first and fall back to a snapshot if the published boundary moved past the cursor meanwhile; pruning publishes the boundary before it deletes, so this catches every concurrent pass. Co-Authored-By: Claude Opus 5.5 --- apps/server/src/ws.test.ts | 81 ++++++++++++++++++++++++++++++++++++++ apps/server/src/ws.ts | 57 +++++++++++++++++++-------- docs/internals/overview.md | 6 ++- 3 files changed, 125 insertions(+), 19 deletions(-) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index a6b6e0851477f..4aeb25072769c 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -33,6 +33,7 @@ import * as TestClock from "effect/testing/TestClock"; import { Headers } from "effect/http"; import { RpcGroup, RpcMessage, RpcServer } from "effect/rpc"; import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; import * as PreviewAutomationBroker from "./mcp/PreviewAutomationBroker.ts"; import * as ServerBrowser from "./preview/ServerBrowser.ts"; @@ -1092,6 +1093,86 @@ it.effect.each(["thread", "shell"] as const)( }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); +// A prune pass can delete replayed rows after the boundary check; the read must not be trusted. +it.effect.each(["thread", "shell"] as const)( + "%s resume loads a snapshot when pruning advances during the replay read", + (scope) => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-10-13T00:00:00.000Z"; + const threadId = ThreadId.make("thread:prune-race"); + const insert = (version: number) => sql<{ sequence: number }>` + INSERT INTO orchestration_events ( + event_id, aggregate_kind, stream_id, stream_version, event_type, + occurred_at, actor_kind, payload_json, metadata_json, application_event_version + ) VALUES ( + ${`prune-race:${version}`}, 'thread', ${threadId}, ${version}, 'provider-session.detached', + ${now}, 'server', ${JSON.stringify({ providerSessionId: "session:race", detachedAt: now })}, + '{}', 2 + ) RETURNING sequence + `; + const [first] = yield* insert(0); + const [second] = yield* insert(1); + yield* insert(2); + + const readStarted = yield* Deferred.make(); + const releaseRead = yield* Deferred.make(); + let gated = false; + const gateReplayRead: Statement.Transformer = (statement) => { + const [query] = statement.compile(); + // The paged replay read (readApplicationRows), not the stats or high-water queries. + if (!gated && /^\s*SELECT\s+sequence,\s+event_id,/.test(query)) { + gated = true; + return Deferred.succeed(readStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseRead)), + Effect.as(statement), + ); + } + return Effect.succeed(statement); + }; + const subscription = Effect.gen(function* () { + const stream = + scope === "thread" + ? yield* subscribeOrchestrationV2Thread({ threadId, afterSequence: first!.sequence }) + : yield* subscribeOrchestrationV2Shell({ afterSequence: first!.sequence }); + return stream; + }).pipe( + Effect.provideService(Statement.CurrentTransformer, gateReplayRead), + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + getThreadSnapshot: () => Effect.die("snapshot"), + listActiveShellIndex: () => Effect.die("snapshot"), + readShellSnapshot: () => Effect.die("snapshot"), + listActiveSubagentThreadIds: () => Effect.succeed([]), + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provideService(ProjectStore.ProjectStoreV2, { + listShells: () => Effect.succeed([]), + } as unknown as ProjectStore.ProjectStoreV2["Service"]), + Effect.provideService( + ProjectService.ProjectService, + {} as ProjectService.ProjectService["Service"], + ), + Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, { + subscribeChanges: PubSub.subscribe(yield* PubSub.unbounded()), + } as ProjectEnrichment.ProjectEnrichmentService["Service"]), + Effect.exit, + Effect.forkChild, + ); + const fiber = yield* subscription; + yield* Deferred.await(readStarted); + // A pass publishes its boundary, then deletes the rows it covers. + yield* sql` + INSERT INTO orchestration_v2_projection_metadata ( + projection_name, schema_version, last_sequence, updated_at + ) VALUES (${OrchestrationEventStore.EVENT_LOG_PRUNED_THROUGH}, 2, ${second!.sequence}, ${now}) + `; + yield* sql`DELETE FROM orchestration_events WHERE sequence = ${second!.sequence}`; + yield* Deferred.succeed(releaseRead, undefined); + const exit = yield* Fiber.join(fiber); + assert.deepEqual(Exit.isFailure(exit) ? Cause.squash(exit.cause) : exit, "snapshot"); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +); + // A thread idle since before the boundary missed nothing; resuming must not resend its body. it.effect("thread resume at its own high-water below the pruned boundary skips the snapshot", () => Effect.gen(function* () { diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 109900d8ad526..6b2880a1f4bf9 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1027,9 +1027,25 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs // Retained creation events may describe deleted threads; replace replay only if a snapshot exists. if (shell !== null) return yield* snapshotThenLive(); } - const replay = coalesceLiveUpdatedEvents( - yield* loadReplayThrough(input.afterSequence, highWater), - ); + const replayed = yield* loadReplayThrough(input.afterSequence, highWater); + // Pruning publishes its boundary before deleting, so a pass during the read shows up here. + if ( + input.afterSequence < highWater && + input.afterSequence < + (yield* applicationEvents.prunedThroughSequence.pipe( + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: `Failed to prepare orchestration V2 thread ${input.threadId} replay`, + cause, + }), + ), + )) + ) { + return yield* snapshotThenLive(); + } + const replay = coalesceLiveUpdatedEvents(replayed); const plan = decideThreadResume({ afterSequence: input.afterSequence, highWater, @@ -1240,31 +1256,38 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc }); } - const highWater = yield* applicationEvents.latestApplicationSequence; - if (!(yield* canReplayPersistedRange(input.afterSequence, highWater, SHELL_RESUME_MAX_GAP))) { + const fromSnapshot = Effect.gen(function* () { const loaded = yield* loadSnapshot(); return composeShellStreamWithEnrichment({ initial: initialSnapshotItems(loaded), tail: completionThenLive(loaded.snapshot.snapshotSequence), enrichment: enrichmentRefreshes, }); + }); + const highWater = yield* applicationEvents.latestApplicationSequence; + if (!(yield* canReplayPersistedRange(input.afterSequence, highWater, SHELL_RESUME_MAX_GAP))) { + return yield* fromSnapshot; } subagentThreadIds = new Set(yield* threadManagement.listActiveSubagentThreadIds()); const loaded = yield* loadProjectMetadataSnapshot(highWater); - const replay = toShellStream( - applicationEvents - .readApplicationEvents({ - afterSequence: input.afterSequence, - throughSequence: highWater, - // A row a newer build wrote (e.g. after a downgrade) must not fail - // every shell resubscribe. The live tail below resumes from - // highWater, so a dropped row never hides later events. - skipUnknownEventTypes: true, - }) - .pipe(Stream.filter(filterSubagentTranscript)), - ); + // Read before replying: the range is budgeted above, and a prune pass during the read + // publishes its boundary first, so the recheck below catches deleted rows. + const replayed = yield* applicationEvents + .readApplicationEvents({ + afterSequence: input.afterSequence, + throughSequence: highWater, + // A row a newer build wrote (e.g. after a downgrade) must not fail + // every shell resubscribe. The live tail below resumes from + // highWater, so a dropped row never hides later events. + skipUnknownEventTypes: true, + }) + .pipe(Stream.filter(filterSubagentTranscript), Stream.runCollect); + if (input.afterSequence < (yield* applicationEvents.prunedThroughSequence)) { + return yield* fromSnapshot; + } + const replay = toShellStream(Stream.fromIterable(replayed)); return composeShellStreamWithEnrichment({ initial: initialEnrichmentItems(loaded), tail: Stream.concat(Stream.concat(replay, completionMarker), liveFrom(highWater)), diff --git a/docs/internals/overview.md b/docs/internals/overview.md index c4a7def7553e9..37500faaa9fc4 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -66,8 +66,10 @@ stream's newest event (high-water marks, stream versions, snapshot sequences), ` (the legacy importer's marker), every `running` run update (Codex usage history reads the first of each attempt), and prompt anchors that retained subagent rows reference. A client whose cursor is below both its thread's newest event and the recorded pruned-through sequence missed pruned -events, so it resumes from a snapshot; an idle thread's cursor resumes without one. Command receipts stay for idempotency; a -retry older than the retained history can find its events gone and must answer from projections. +events, so it resumes from a snapshot; an idle thread's cursor resumes without one. Resumes +recheck that sequence after reading the replay, since a pass can prune during the read. Command +receipts stay for idempotency: a retry older than the retained history is still deduplicated, +but its replay comes back without the pruned events. Retained events must remain decodable on replay. Changing a schema affects old environments at startup as well as live RPC traffic. Compatibility work must account for stored history, not just From 0b5b885ba2e339edee97b92a538b2013dbd5fea8 Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:07:16 -0800 Subject: [PATCH 4/5] test(server): require idle-thread resume to succeed, not merely skip the snapshot Co-Authored-By: Claude Opus 5.5 --- apps/server/src/ws.test.ts | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/apps/server/src/ws.test.ts b/apps/server/src/ws.test.ts index 4aeb25072769c..94aa91e16ffd8 100644 --- a/apps/server/src/ws.test.ts +++ b/apps/server/src/ws.test.ts @@ -1201,9 +1201,11 @@ it.effect("thread resume at its own high-water below the pruned boundary skips t Effect.provideService(ThreadManagement.ThreadManagementService, { ensureLegacyTranscript: () => Effect.void, getThreadSnapshot: () => Effect.die("snapshot"), + streamStoredEventsFrom: () => Stream.empty, } as unknown as ThreadManagement.ThreadManagementService["Service"]), Effect.exit, ); - assert.notDeepEqual(Exit.isFailure(exit) ? Cause.squash(exit.cause) : "resumed", "snapshot"); + // A snapshot load dies with "snapshot", so success means the resume replayed nothing. + assert.isTrue(Exit.isSuccess(exit)); }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), ); From e2ecefda3a9416614fe3203d13d7d114e03961af Mon Sep 17 00:00:00 2001 From: Julius Marminge <51714798+juliusmarminge@users.noreply.github.com> Date: Mon, 2 Nov 2026 18:31:33 -0800 Subject: [PATCH 5/5] docs(server): qualify which command receipts survive pruning Co-Authored-By: Claude Opus 5.5 --- docs/internals/overview.md | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/docs/internals/overview.md b/docs/internals/overview.md index 37500faaa9fc4..f24e3625d92f8 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -67,9 +67,10 @@ stream's newest event (high-water marks, stream versions, snapshot sequences), ` of each attempt), and prompt anchors that retained subagent rows reference. A client whose cursor is below both its thread's newest event and the recorded pruned-through sequence missed pruned events, so it resumes from a snapshot; an idle thread's cursor resumes without one. Resumes -recheck that sequence after reading the replay, since a pass can prune during the read. Command +recheck that sequence after reading the replay, since a pass can prune during the read. V2 command receipts stay for idempotency: a retry older than the retained history is still deduplicated, -but its replay comes back without the pruned events. +but its replay comes back without the pruned events. Receipts of imported legacy commands are +removed once their threads are imported. Retained events must remain decodable on replay. Changing a schema affects old environments at startup as well as live RPC traffic. Compatibility work must account for stored history, not just