From 35c5f43daf2d477673d186724d487e094102ca9c Mon Sep 17 00:00:00 2001 From: Yannic Charlon <52761674+JustYannicc@users.noreply.github.com> Date: Sat, 31 Oct 2026 13:19:24 +0100 Subject: [PATCH 1/5] perf(server): prune thread event history older than a day --- apps/server/src/cli/doctor.ts | 2 +- .../EventHistoryRetention.test.ts | 177 ++++++++++++ .../EventHistoryRetentionAnchors.test.ts | 272 ++++++++++++++++++ .../EventHistoryRetentionTestSupport.ts | 212 ++++++++++++++ .../orchestration-v2/ProjectionMaintenance.ts | 85 +++++- .../src/persistence/DatabaseStorage.test.ts | 2 +- .../server/src/persistence/DatabaseStorage.ts | 9 +- apps/server/src/persistence/Migrations.ts | 2 + .../Migrations/077_EventHistoryRetention.ts | 10 + .../persistence/OrchestrationEventStore.ts | 20 ++ .../reconcileV2PreviewMigration.test.ts | 2 + .../src/ws.eventHistoryRetention.test.ts | 165 +++++++++++ apps/server/src/ws.ts | 87 ++++-- docs/internals/overview.md | 9 +- docs/user/background-service.md | 10 +- 15 files changed, 1021 insertions(+), 43 deletions(-) create mode 100644 apps/server/src/orchestration-v2/EventHistoryRetention.test.ts create mode 100644 apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts create mode 100644 apps/server/src/orchestration-v2/EventHistoryRetentionTestSupport.ts create mode 100644 apps/server/src/persistence/Migrations/077_EventHistoryRetention.ts create mode 100644 apps/server/src/ws.eventHistoryRetention.test.ts diff --git a/apps/server/src/cli/doctor.ts b/apps/server/src/cli/doctor.ts index f7afaf9e4d0ca..9cc3fcba01fac 100644 --- a/apps/server/src/cli/doctor.ts +++ b/apps/server/src/cli/doctor.ts @@ -108,7 +108,7 @@ export const runDoctor = Effect.fn("runDoctor")(function* ( .pipe( Effect.tap((result) => Console.log( - `Done: ${formatBytes(result.beforeBytes)} -> ${formatBytes(result.afterBytes)}, ${result.deletedEventCount} superseded events removed.`, + `Done: ${formatBytes(result.beforeBytes)} -> ${formatBytes(result.afterBytes)}, ${result.deletedEventCount} old history events removed.`, ), ), ), diff --git a/apps/server/src/orchestration-v2/EventHistoryRetention.test.ts b/apps/server/src/orchestration-v2/EventHistoryRetention.test.ts new file mode 100644 index 0000000000000..144f0e755e98f --- /dev/null +++ b/apps/server/src/orchestration-v2/EventHistoryRetention.test.ts @@ -0,0 +1,177 @@ +import { assert, it } from "@effect/vitest"; +import { ThreadId } from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Deferred from "effect/Deferred"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as Statement from "effect/sql/Statement"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; +import * as ThreadManagement from "./ThreadManagementService.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 { subscribeOrchestrationV2Thread } from "../ws.ts"; +import { + createdEvent, + layerTest, + makeThread, + metadataUpdatedEvent, + visitedEvent, +} from "./EventHistoryRetentionTestSupport.ts"; + +it.effect("prunes expired thread history while preserving current projections", () => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const events = yield* EventStore.EventStoreV2; + const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const createdAt = yield* DateTime.now; + const threadId = ThreadId.make("thread:event-history-retention"); + const thread = makeThread(threadId, createdAt); + const created = createdEvent(thread, createdAt); + const oldMetadata = metadataUpdatedEvent(thread, "old", "Old title", createdAt); + const [createdStored, oldMetadataStored] = yield* sink.write({ + events: [created, oldMetadata], + }); + + yield* TestClock.adjust(Duration.hours(25)); + const recentAt = yield* DateTime.now; + const recent = visitedEvent(thread, "recent", recentAt); + yield* sink.write({ events: [recent] }); + const before = yield* projections.getThreadProjection(threadId); + + const result = yield* maintenance.compactEventStore(); + const retained = yield* events.read({ threadId }).pipe(Stream.runCollect); + + assert.equal(result.deletedEventCount, 1); + assert.deepEqual( + retained.map((stored) => stored.event.id), + [created.id, recent.id], + ); + assert.deepEqual(yield* projections.getThreadProjection(threadId), before); + assert.equal(yield* applicationEvents.replayFloor(threadId), oldMetadataStored!.sequence); + assert.equal(yield* applicationEvents.replayFloor(), oldMetadataStored!.sequence); + assert.isTrue((yield* maintenance.verify).valid); + assert.isTrue(Exit.isFailure(yield* Effect.exit(maintenance.rebuild))); + assert.deepEqual(yield* projections.getThreadProjection(threadId), before); + + const resume = (afterSequence: number) => + subscribeOrchestrationV2Thread({ + threadId, + afterSequence, + requestCompletionMarker: true, + }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.map((item) => item.kind), + Stream.runCollect, + ), + ), + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + getThreadShell: projections.getThreadShell, + getThreadSnapshot: projections.getThreadSnapshot, + streamStoredEventsFrom: () => Stream.never, + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + ); + const resumed = yield* resume(createdStored!.sequence); + assert.deepEqual(Array.from(resumed), ["snapshot", "synchronized"]); + const resumedAtFloor = yield* resume(oldMetadataStored!.sequence); + assert.deepEqual(Array.from(resumedAtFloor), ["event", "synchronized"]); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("thread resume falls back when retention commits during event replay", () => + Effect.scoped( + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const createdAt = yield* DateTime.now; + const threadId = ThreadId.make("thread:event-history-retention-race"); + const thread = makeThread(threadId, createdAt); + const [createdStored] = yield* sink.write({ + events: [ + createdEvent(thread, createdAt), + metadataUpdatedEvent(thread, "old", "Old title", createdAt), + ], + }); + yield* TestClock.adjust(Duration.hours(25)); + const recentAt = yield* DateTime.now; + yield* sink.write({ events: [visitedEvent(thread, "recent", recentAt)] }); + + const replayReadStarted = yield* Deferred.make(); + const releaseReplayRead = yield* Deferred.make(); + let blocked = false; + const blockReplayRead: Statement.Transformer = (statement) => { + const [query] = statement.compile(); + if (!blocked && query.includes("event_id") && query.includes("ORDER BY sequence ASC")) { + blocked = true; + return Deferred.succeed(replayReadStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseReplayRead)), + Effect.as(statement), + ); + } + return Effect.succeed(statement); + }; + const resume = subscribeOrchestrationV2Thread({ + threadId, + afterSequence: createdStored!.sequence, + requestCompletionMarker: true, + }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.map((item) => item.kind), + Stream.runCollect, + ), + ), + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + getThreadShell: projections.getThreadShell, + getThreadSnapshot: projections.getThreadSnapshot, + streamStoredEventsFrom: () => Stream.never, + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + Effect.provideService(Statement.CurrentTransformer, blockReplayRead), + ); + const resumedFiber = yield* Effect.forkScoped(resume); + yield* Deferred.await(replayReadStarted); + const result = yield* maintenance.compactEventStore(); + yield* Deferred.succeed(releaseReplayRead, undefined); + + assert.equal(result.deletedEventCount, 1); + assert.deepEqual(Array.from(yield* Fiber.join(resumedFiber)), ["snapshot", "synchronized"]); + }).pipe(Effect.provide(layerTest)), + ), +); + +it.effect("retention stays off when the replay floor table is missing", () => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const events = yield* EventStore.EventStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const sql = yield* SqlClient.SqlClient; + const now = yield* DateTime.now; + const thread = makeThread(ThreadId.make("thread:retention-without-floor"), now); + const oldEvent = metadataUpdatedEvent(thread, "old", "Old title", now); + yield* sink.write({ events: [createdEvent(thread, now), oldEvent] }); + yield* sql`DROP TABLE orchestration_v2_event_retention`; + yield* TestClock.adjust(Duration.hours(25)); + + assert.equal((yield* maintenance.compactEventStore()).deletedEventCount, 0); + const retained = yield* events.read({ threadId: thread.id }).pipe(Stream.runCollect); + assert.include( + retained.map((stored) => stored.event.id), + oldEvent.id, + ); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts b/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts new file mode 100644 index 0000000000000..86e5d34020fb8 --- /dev/null +++ b/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts @@ -0,0 +1,272 @@ +import { assert, it } from "@effect/vitest"; +import { + CommandId, + EventId, + MessageId, + type OrchestrationV2DomainEvent, + ProjectId, + RunAttemptId, + RunId, + ThreadId, + type OrchestrationV2Run, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; +import * as Effect from "effect/Effect"; +import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; +import * as ThreadManagement from "./ThreadManagementService.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 { + createdEvent, + layerTest, + makeThread, + messageEvent, + metadataUpdatedEvent, + promptEvents, + providerInstanceId, + runningEvent, + visitedEvent, +} from "./EventHistoryRetentionTestSupport.ts"; +import { subscribeOrchestrationV2Thread } from "../ws.ts"; + +it.effect("retention keeps replay anchors and leaves project and thread projections intact", () => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const events = yield* EventStore.EventStoreV2; + const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const sql = yield* SqlClient.SqlClient; + const oldAt = yield* DateTime.now; + + const anchoredThreadId = ThreadId.make("thread:retention-anchors"); + const anchoredThread = makeThread(anchoredThreadId, oldAt); + const run: OrchestrationV2Run = { + id: RunId.make(`run:${anchoredThreadId}`), + threadId: anchoredThreadId, + ordinal: 1, + providerInstanceId, + modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" }, + providerThreadId: null, + userMessageId: MessageId.make(`message:${anchoredThreadId}:run`), + rootNodeId: null, + activeAttemptId: RunAttemptId.make(`run-attempt:${anchoredThreadId}`), + status: "running", + queuePosition: null, + requestedAt: oldAt, + startedAt: oldAt, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }; + yield* sink.write({ + events: [ + createdEvent(anchoredThread, oldAt), + runningEvent(anchoredThreadId, run, "first", oldAt), + runningEvent(anchoredThreadId, run, "later", oldAt), + ...promptEvents(anchoredThreadId, oldAt, 0), + ...promptEvents(anchoredThreadId, oldAt, 1), + ...promptEvents(anchoredThreadId, oldAt, 2), + ], + }); + const commandId = CommandId.make("command:retention:receipt"); + const commandEvent = messageEvent(anchoredThreadId, "receipt", oldAt); + yield* sink.commitCommand({ + commandId, + threadId: anchoredThreadId, + commandType: "thread.send", + acceptedAt: oldAt, + events: [commandEvent], + effects: [], + }); + + const importingThreadId = ThreadId.make("thread:retention-importing"); + const importingThread = makeThread(importingThreadId, oldAt); + const importedMessages = [ + messageEvent(importingThreadId, "one", oldAt), + messageEvent(importingThreadId, "two", oldAt), + ]; + yield* sink.write({ events: [createdEvent(importingThread, oldAt), ...importedMessages] }); + yield* sql` + INSERT INTO orchestration_v2_legacy_imports ( + thread_id, source_updated_at, shell_imported_at, transcript_imported_at, + imported_message_count, last_error + ) VALUES ( + ${importingThreadId}, ${DateTime.formatIso(oldAt)}, ${DateTime.formatIso(oldAt)}, NULL, 0, NULL + ) + `; + + const highWaterThreadId = ThreadId.make("thread:retention-old-high-water"); + const highWaterThread = makeThread(highWaterThreadId, oldAt); + const oldLatest = metadataUpdatedEvent(highWaterThread, "latest", "Latest old state", oldAt); + yield* sink.write({ events: [createdEvent(highWaterThread, oldAt), oldLatest] }); + + const archivedThreadId = ThreadId.make("thread:retention-archived"); + const archivedThread = makeThread(archivedThreadId, oldAt); + const archivedEvent = { + id: EventId.make(`event:${archivedThreadId}:archived`), + type: "thread.archived", + threadId: archivedThreadId, + providerInstanceId, + occurredAt: oldAt, + payload: { ...archivedThread, archivedAt: oldAt }, + } satisfies OrchestrationV2DomainEvent; + yield* sink.write({ events: [createdEvent(archivedThread, oldAt), archivedEvent] }); + + const projectId = ProjectId.make("project:retention-project"); + const projectEvent = yield* applicationEvents.appendProjectEvent({ + eventId: EventId.make("event:retention-project:old"), + type: "project.meta-updated", + aggregateKind: "project", + aggregateId: projectId, + occurredAt: DateTime.formatIso(oldAt), + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + payload: { projectId, title: "Old project", updatedAt: DateTime.formatIso(oldAt) }, + }); + + yield* TestClock.adjust(Duration.hours(25)); + const recentAt = yield* DateTime.now; + yield* sink.write({ + events: [ + visitedEvent(anchoredThread, "recent", recentAt), + visitedEvent(importingThread, "recent", recentAt), + visitedEvent({ ...archivedThread, archivedAt: oldAt }, "recent", recentAt), + ], + }); + + const before = [ + yield* projections.getThreadProjection(anchoredThreadId), + yield* projections.getThreadProjection(importingThreadId), + yield* projections.getThreadProjection(highWaterThreadId), + yield* projections.getThreadProjection(archivedThreadId), + ]; + assert.isAbove((yield* maintenance.compactEventStore()).deletedEventCount, 0); + + const anchored = yield* events.read({ threadId: anchoredThreadId }).pipe(Stream.runCollect); + const anchoredIds = new Set(anchored.map((stored) => stored.event.id)); + for (const id of [ + "event:thread:retention-anchors:run:first", + "event:thread:retention-anchors:run:later", + commandEvent.id, + "event:thread:retention-anchors:subagent:1", + "event:thread:retention-anchors:subagent-item:1", + ]) { + assert.include(anchoredIds, id); + } + const fullPrompt = anchored.find( + (stored) => stored.event.id === "event:thread:retention-anchors:subagent:1", + )?.event; + assert.equal( + fullPrompt?.type === "subagent.updated" ? fullPrompt.payload.prompt : null, + "Prompt B. ".repeat(40), + ); + + const importing = yield* events.read({ threadId: importingThreadId }).pipe(Stream.runCollect); + const importingIds = new Set(importing.map((stored) => stored.event.id)); + for (const event of importedMessages) assert.include(importingIds, event.id); + + const highWater = yield* events.read({ threadId: highWaterThreadId }).pipe(Stream.runCollect); + assert.include( + highWater.map((stored) => stored.event.id), + oldLatest.id, + ); + const archived = yield* events.read({ threadId: archivedThreadId }).pipe(Stream.runCollect); + assert.notInclude( + archived.map((stored) => stored.event.id), + archivedEvent.id, + ); + const projectReplay = yield* applicationEvents + .readApplicationEvents({ + afterSequence: projectEvent.sequence - 1, + throughSequence: projectEvent.sequence, + }) + .pipe(Stream.runCollect); + assert.deepEqual( + projectReplay.map((stored) => { + if (!("aggregateKind" in stored)) return null; + return stored.eventId; + }), + [projectEvent.eventId], + ); + + assert.deepEqual( + [ + yield* projections.getThreadProjection(anchoredThreadId), + yield* projections.getThreadProjection(importingThreadId), + yield* projections.getThreadProjection(highWaterThreadId), + yield* projections.getThreadProjection(archivedThreadId), + ], + before, + ); + assert.isTrue((yield* maintenance.verify).valid); + }).pipe(Effect.provide(layerTest)), +); + +it.effect("retention keeps a deleted thread tombstone for old cursors", () => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const events = yield* EventStore.EventStoreV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread:retention-deleted"); + const thread = makeThread(threadId, now); + const [createdStored] = yield* sink.write({ + events: [createdEvent(thread, now), messageEvent(threadId, "before-delete", now)], + }); + + yield* TestClock.adjust(Duration.hours(25)); + const deletedAt = yield* DateTime.now; + const deleted = { + id: EventId.make(`event:${threadId}:deleted`), + type: "thread.deleted", + threadId, + providerInstanceId, + occurredAt: deletedAt, + payload: { ...thread, deletedAt }, + } satisfies OrchestrationV2DomainEvent; + yield* sink.write({ events: [deleted] }); + + assert.equal((yield* maintenance.compactEventStore()).deletedEventCount, 1); + const retained = yield* events.read({ threadId }).pipe(Stream.runCollect); + assert.deepEqual( + retained.map((stored) => stored.event.id), + [createdStored!.event.id, deleted.id], + ); + assert.isNull(yield* projections.getThreadShell(threadId)); + + const resumed = yield* subscribeOrchestrationV2Thread({ + threadId, + afterSequence: createdStored!.sequence, + requestCompletionMarker: true, + }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.runCollect, + ), + ), + Effect.provideService(ThreadManagement.ThreadManagementService, { + ensureLegacyTranscript: () => Effect.void, + getThreadShell: projections.getThreadShell, + getThreadSnapshot: projections.getThreadSnapshot, + streamStoredEventsFrom: () => Stream.never, + } as unknown as ThreadManagement.ThreadManagementService["Service"]), + ); + assert.deepEqual( + Array.from(resumed, (item) => item.kind), + ["event", "synchronized"], + ); + assert.equal(resumed[0]?.kind === "event" ? resumed[0].event.type : null, "thread.deleted"); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/EventHistoryRetentionTestSupport.ts b/apps/server/src/orchestration-v2/EventHistoryRetentionTestSupport.ts new file mode 100644 index 0000000000000..4ef011fdc153a --- /dev/null +++ b/apps/server/src/orchestration-v2/EventHistoryRetentionTestSupport.ts @@ -0,0 +1,212 @@ +import { + EventId, + MessageId, + type ModelSelection, + type OrchestrationV2AppThread, + type OrchestrationV2DomainEvent, + type OrchestrationV2Run, + NodeId, + ProjectId, + ProviderDriverKind, + ProviderInstanceId, + ThreadId, + TurnItemId, +} from "@t3tools/contracts"; +import * as DateTime from "effect/DateTime"; +import * as Layer from "effect/Layer"; + +import * as OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; +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 eventStore = EventStore.layer.pipe(Layer.provideMerge(database)); +const projectionStore = ProjectionStore.layer.pipe(Layer.provideMerge(database)); +const stores = Layer.mergeAll(database, eventStore, projectionStore); + +export const layerTest = Layer.mergeAll( + stores, + OrchestrationEventStore.layer.pipe(Layer.provideMerge(database)), + EventSink.layer.pipe(Layer.provide(stores)), + ProjectionMaintenance.layer.pipe(Layer.provide(stores)), +); + +export const providerInstanceId = ProviderInstanceId.make("codex"); +const providerDriver = ProviderDriverKind.make("codex"); +const modelSelection = { + instanceId: providerInstanceId, + model: "gpt-5.4", +} satisfies ModelSelection; + +export function makeThread(threadId: ThreadId, now: DateTime.Utc): OrchestrationV2AppThread { + return { + id: threadId, + projectId: ProjectId.make(`project:${threadId}`), + title: "Event history retention", + 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, + }; +} + +export function createdEvent(thread: OrchestrationV2AppThread, now: DateTime.Utc) { + return { + id: EventId.make(`event:${thread.id}:created`), + type: "thread.created", + threadId: thread.id, + providerInstanceId, + occurredAt: now, + payload: thread, + } satisfies OrchestrationV2DomainEvent; +} + +export function metadataUpdatedEvent( + thread: OrchestrationV2AppThread, + suffix: string, + title: string, + now: DateTime.Utc, +) { + return { + id: EventId.make(`event:${thread.id}:metadata:${suffix}`), + type: "thread.metadata-updated", + threadId: thread.id, + providerInstanceId, + occurredAt: now, + payload: { ...thread, title, updatedAt: now }, + } satisfies OrchestrationV2DomainEvent; +} + +export function visitedEvent(thread: OrchestrationV2AppThread, suffix: string, now: DateTime.Utc) { + return { + id: EventId.make(`event:${thread.id}:visit:${suffix}`), + type: "thread.visited", + threadId: thread.id, + providerInstanceId, + occurredAt: now, + payload: { ...thread, updatedAt: now, lastVisitedAt: now }, + } satisfies OrchestrationV2DomainEvent; +} + +export function messageEvent(threadId: ThreadId, suffix: string, now: DateTime.Utc) { + return { + id: EventId.make(`event:${threadId}:message:${suffix}`), + type: "message.updated", + threadId, + occurredAt: now, + payload: { + createdBy: "user", + creationSource: "web", + id: MessageId.make(`message:${threadId}:${suffix}`), + threadId, + runId: null, + nodeId: null, + role: "user", + text: suffix, + attachments: [], + streaming: false, + createdAt: now, + updatedAt: now, + }, + } satisfies OrchestrationV2DomainEvent; +} + +export function runningEvent( + threadId: ThreadId, + run: OrchestrationV2Run, + suffix: string, + now: DateTime.Utc, +) { + return { + id: EventId.make(`event:${threadId}:run:${suffix}`), + type: "run.updated", + threadId, + runId: run.id, + occurredAt: now, + payload: run, + } satisfies OrchestrationV2DomainEvent; +} + +export function promptEvents(threadId: ThreadId, now: DateTime.Utc, revision: number) { + const id = NodeId.make(`subagent:${threadId}:retention`); + const prompt = revision === 0 ? "Prompt A. ".repeat(40) : "Prompt B. ".repeat(40); + const progress = `Review progress ${revision}`; + return [ + { + id: EventId.make(`event:${threadId}:subagent:${revision}`), + type: "subagent.updated", + threadId, + occurredAt: now, + payload: { + id, + threadId, + runId: null, + parentNodeId: NodeId.make(`node:${threadId}:parent`), + origin: "provider_native", + createdBy: "agent", + driver: providerDriver, + providerInstanceId, + providerThreadId: null, + childThreadId: null, + nativeTaskRef: { driver: providerDriver, nativeId: "retention-review", strength: "strong" }, + prompt, + progress, + title: "Review", + model: null, + status: "running", + result: null, + startedAt: now, + completedAt: null, + updatedAt: now, + }, + }, + { + id: EventId.make(`event:${threadId}:subagent-item:${revision}`), + type: "turn-item.updated", + threadId, + occurredAt: now, + payload: { + id: TurnItemId.make(`turn-item:${threadId}:subagent`), + threadId, + runId: null, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + 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; +} diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index aad9d355924be..16a83adf657a1 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -2,6 +2,7 @@ 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"; @@ -56,6 +57,8 @@ const encodeEntityKey = Schema.encodeSync( Schema.fromJsonString(Schema.Tuple([Schema.String, Schema.String, Schema.NullOr(Schema.String)])), ); +const EVENT_HISTORY_RETENTION = Duration.hours(24); + export const layer: Layer.Layer< ProjectionMaintenanceV2, never, @@ -113,9 +116,27 @@ export const layer: Layer.Layer< } satisfies ProjectionVerificationV2; }).pipe(sql.withTransaction); + const hasReplayFloorTable = sql<{ readonly present: number }>` + SELECT EXISTS ( + SELECT 1 FROM sqlite_master + WHERE type = 'table' AND name = 'orchestration_v2_event_retention' + ) AS present + `.pipe(Effect.map((rows) => rows[0]?.present === 1)); + const rebuild = Effect.gen(function* () { yield* sql.withTransaction( Effect.gen(function* () { + if (yield* hasReplayFloorTable) { + const pruned = yield* sql<{ readonly pruned: number }>` + SELECT EXISTS (SELECT 1 FROM orchestration_v2_event_retention) AS pruned + `; + if (pruned[0]?.pruned === 1) { + // Pruned history cannot rebuild the projections that now hold its state. + return yield* new ProjectionMaintenanceError({ + operation: "rebuild from pruned event history", + }); + } + } const throughSequence = yield* eventStore.latestSequence(); yield* sql`DELETE FROM orchestration_v2_projection_context_transfers`; yield* sql`DELETE FROM orchestration_v2_projection_context_handoffs`; @@ -227,7 +248,7 @@ export const layer: Layer.Layer< // 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. + // Keep first/latest keyed events for replay order and page every event to bound sparse scans. const compactEventStore = Effect.fn("ProjectionMaintenance.compactEventStore")(function* ( betweenPages: Effect.Effect = Effect.void, ) { @@ -241,6 +262,13 @@ export const layer: Layer.Layer< `.pipe(NodeSqliteClient.readOnly); let throughSequence = bounds[0]?.event_sequence ?? 0; let throughReceiptRowId = bounds[0]?.receipt_row_id ?? 0; + // Older doctor builds without migration 77 can compact only superseded history. + const retentionCutoff = (yield* hasReplayFloorTable) + ? DateTime.formatIso( + DateTime.subtractDuration(yield* DateTime.now, EVENT_HISTORY_RETENTION), + ) + : ""; + const newestStreamKeys = new Set(); const retainedThreadIds = new Set(); const retainedReadStateThreadIds = new Set(); // Retain each key's oldest candidate, deleting it only when an older event appears. @@ -257,6 +285,8 @@ export const layer: Layer.Layer< readonly event_type: string; readonly entity_id: string | null; readonly imported_legacy_thread: number; + readonly expired: number; + readonly retention_anchor: number; }>` SELECT event.sequence, @@ -264,6 +294,27 @@ export const layer: Layer.Layer< event.aggregate_kind, event.stream_id, event.event_type, + event.occurred_at < ${retentionCutoff} AS expired, + CASE + WHEN event.occurred_at >= ${retentionCutoff} THEN 0 + WHEN event.command_id IS NOT NULL AND EXISTS ( + SELECT 1 FROM orchestration_command_receipts AS receipt + WHERE receipt.command_id = event.command_id + ) THEN 1 + WHEN 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 NULL + ) THEN 1 + WHEN event.event_type = 'run.updated' + AND json_extract(event.payload_json, '$.status') = 'running' + THEN 1 + WHEN event.event_type IN ('subagent.updated', 'turn-item.updated') + AND json_type(event.payload_json, '$.prompt') = 'text' + AND json_type(event.payload_json, '$.promptSequence') IS NULL + THEN 1 + ELSE 0 + END AS retention_anchor, CASE WHEN event.application_event_version = 2 AND event.event_type IN ('message.updated', 'node.updated') @@ -296,7 +347,24 @@ export const layer: Layer.Layer< `.pipe(NodeSqliteClient.readOnly); const page = rows.slice(0, COMPACTION_PAGE_SIZE); const obsolete: number[] = []; + const replayFloors = new Map(); for (const row of page) { + const streamKey = `${row.aggregate_kind}\0${row.stream_id}`; + const isNewestInStream = !newestStreamKeys.has(streamKey); + newestStreamKeys.add(streamKey); + if ( + row.expired === 1 && + row.retention_anchor === 0 && + row.application_event_version === 2 && + row.aggregate_kind === "thread" && + !isNewestInStream && + row.event_type !== "thread.created" && + row.event_type !== "thread.deleted" + ) { + obsolete.push(row.sequence); + if (!replayFloors.has(row.stream_id)) replayFloors.set(row.stream_id, row.sequence); + continue; + } if (row.imported_legacy_thread === 1) { obsolete.push(row.sequence); } else if (row.application_event_version === 2) { @@ -341,7 +409,20 @@ export const layer: Layer.Layer< // Never hold a transaction or the reader lock across a yield or a rest. yield* Effect.yieldNow; if (obsolete.length > 0) { - yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`; + // Replay readers must see a deletion and its floor from the same commit. + yield* sql.withTransaction( + Effect.gen(function* () { + yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`; + for (const [threadId, floor] of replayFloors) { + yield* sql` + INSERT INTO orchestration_v2_event_retention (thread_id, replay_floor) + VALUES (${threadId}, ${floor}) + ON CONFLICT(thread_id) DO UPDATE SET + replay_floor = MAX(replay_floor, excluded.replay_floor) + `; + } + }), + ); deletedEventCount += obsolete.length; yield* Effect.yieldNow; } diff --git a/apps/server/src/persistence/DatabaseStorage.test.ts b/apps/server/src/persistence/DatabaseStorage.test.ts index 878f596600888..e864852f9e458 100644 --- a/apps/server/src/persistence/DatabaseStorage.test.ts +++ b/apps/server/src/persistence/DatabaseStorage.test.ts @@ -77,7 +77,7 @@ it.effect("remembers completion only after the paced pass finishes", () => const fs = yield* FileSystem.FileSystem; const path = yield* Path.Path; const config = yield* ServerConfig.ServerConfig; - const timestampPath = path.join(path.dirname(config.dbPath), "compacted-at"); + const timestampPath = path.join(path.dirname(config.dbPath), "history-retained-at"); const betweenPagesReached = yield* Deferred.make(); const completed = yield* Deferred.make(); const timestampWritten = yield* Deferred.make(); diff --git a/apps/server/src/persistence/DatabaseStorage.ts b/apps/server/src/persistence/DatabaseStorage.ts index bbb4dd1399efd..02c8b27920047 100644 --- a/apps/server/src/persistence/DatabaseStorage.ts +++ b/apps/server/src/persistence/DatabaseStorage.ts @@ -31,7 +31,7 @@ export class DatabaseStorage extends Context.Service< } >()("t3/persistence/DatabaseStorage") {} -// Deleted events have newer retained replacements; old cursors replay to the same state. +// Expired thread history raises a replay floor; older cursors receive a snapshot. const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; @@ -98,7 +98,8 @@ const make = Effect.gen(function* () { // Even a pass that deletes nothing scans the event store. Persist completion // so restarting does not trigger another scan before the six-hour interval. - const compactedAtPath = path.join(path.dirname(config.dbPath), "compacted-at"); + // A new marker starts retention shortly after upgrade instead of waiting six hours. + const historyRetainedAtPath = path.join(path.dirname(config.dbPath), "history-retained-at"); const compact = Effect.gen(function* () { let sliceStarted = yield* Clock.monotonicTimeNanos; const betweenPages = Effect.gen(function* () { @@ -110,7 +111,7 @@ const make = Effect.gen(function* () { }); const result = yield* maintenance.compactEventStore(betweenPages); const completedAt = yield* Clock.currentTimeMillis; - yield* fs.writeFileString(compactedAtPath, String(completedAt)).pipe(Effect.ignore); + yield* fs.writeFileString(historyRetainedAtPath, String(completedAt)).pipe(Effect.ignore); if (result.deletedEventCount > 0 || result.deletedReceiptCount > 0) { yield* Effect.logInfo("Compacted orchestration history", result); } @@ -119,7 +120,7 @@ const make = Effect.gen(function* () { Effect.logWarning("Orchestration history compaction failed", { cause }), ), ); - const lastCompactedAt = yield* fs.readFileString(compactedAtPath).pipe( + const lastCompactedAt = yield* fs.readFileString(historyRetainedAtPath).pipe( Effect.map(Number), Effect.orElseSucceed(() => Number.NaN), ); diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index e13c2b449d9e7..63024ea6e808f 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -79,6 +79,7 @@ import Migration0073 from "./Migrations/073_ActiveTurnItemIndexes.ts"; import Migration0074 from "./Migrations/074_ExcludeAttachmentPlaceholdersFromSearch.ts"; import Migration0075 from "./Migrations/075_SubagentDeliveryState.ts"; import Migration0076 from "./Migrations/076_ProjectionProjectsPinArchive.ts"; +import Migration0077 from "./Migrations/077_EventHistoryRetention.ts"; const migrationEntries = [ [1, "OrchestrationEvents", Migration0001], @@ -158,6 +159,7 @@ const migrationEntries = [ [74, "ExcludeAttachmentPlaceholdersFromSearch", Migration0074], [75, "SubagentDeliveryState", Migration0075], [76, "ProjectionProjectsPinArchive", Migration0076], + [77, "EventHistoryRetention", Migration0077], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/077_EventHistoryRetention.ts b/apps/server/src/persistence/Migrations/077_EventHistoryRetention.ts new file mode 100644 index 0000000000000..d69ccb31e0d84 --- /dev/null +++ b/apps/server/src/persistence/Migrations/077_EventHistoryRetention.ts @@ -0,0 +1,10 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* sql`CREATE TABLE orchestration_v2_event_retention ( + thread_id TEXT PRIMARY KEY, + replay_floor INTEGER NOT NULL + )`; +}); diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index 51c3861a9fca0..e00541845f6ad 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -93,6 +93,11 @@ export class OrchestrationEventStore extends Context.Service< readonly latestApplicationSequence: Effect.Effect; + /** Highest sequence removed from one thread, or any thread when omitted. */ + readonly replayFloor: ( + threadId?: ThreadId, + ) => Effect.Effect; + /** Unknown event types fail unless skipUnknownEventTypes logs and drops them; only client replay may opt in. */ readonly readApplicationEvents: (input: { readonly afterSequence: number; @@ -657,6 +662,20 @@ const makeEventStore = Effect.gen(function* () { Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.latestAgentSequence:query")), ); + const replayFloor: OrchestrationEventStore["Service"]["replayFloor"] = Effect.fn( + "OrchestrationEventStore.replayFloor", + )(function* (threadId?: ThreadId) { + const rows = yield* sql<{ readonly floor: number | null }>` + SELECT MAX(replay_floor) AS floor + FROM orchestration_v2_event_retention + ${threadId === undefined ? sql`` : sql`WHERE thread_id = ${threadId}`} + `.pipe( + NodeSqliteClient.readOnly, + Effect.mapError(toPersistenceSqlError("OrchestrationEventStore.replayFloor:query")), + ); + return rows[0]?.floor ?? 0; + }); + // The OR planner otherwise scans every V2 event instead of seeking the final sequence. const latestApplicationSequence = sql<{ readonly sequence: number | null }>` SELECT MAX(sequence) AS sequence @@ -747,6 +766,7 @@ const makeEventStore = Effect.gen(function* () { getReplayStats, latestAgentSequence, latestApplicationSequence, + replayFloor, readApplicationEvents: catchUpApplicationEvents, publishCommitted: (events) => PubSub.publishAll(committedEvents, events).pipe(Effect.asVoid), streamApplicationEvents, diff --git a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts index 08c5094bf49c8..06e8db784ff1c 100644 --- a/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts +++ b/apps/server/src/persistence/reconcileV2PreviewMigration.test.ts @@ -56,6 +56,7 @@ describe("V2 preview upgrade", () => { [74, "ExcludeAttachmentPlaceholdersFromSearch"], [75, "SubagentDeliveryState"], [76, "ProjectionProjectsPinArchive"], + [77, "EventHistoryRetention"], ]); assert.deepStrictEqual(yield* runMigrations(), []); assert.deepStrictEqual(yield* sql`SELECT * FROM orchestration_v2_legacy_imports`, imports); @@ -155,6 +156,7 @@ describe("V2 preview upgrade", () => { [74, "ExcludeAttachmentPlaceholdersFromSearch"], [75, "SubagentDeliveryState"], [76, "ProjectionProjectsPinArchive"], + [77, "EventHistoryRetention"], ]); }).pipe(Effect.provide(NodeSqliteClient.layer({ filename: ":memory:" }))), ); diff --git a/apps/server/src/ws.eventHistoryRetention.test.ts b/apps/server/src/ws.eventHistoryRetention.test.ts new file mode 100644 index 0000000000000..b94ef55ac39f0 --- /dev/null +++ b/apps/server/src/ws.eventHistoryRetention.test.ts @@ -0,0 +1,165 @@ +import { assert, it } from "@effect/vitest"; +import { ProjectId, ThreadId } from "@t3tools/contracts"; +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 PubSub from "effect/PubSub"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as Statement from "effect/sql/Statement"; + +import * as ProjectEnrichment from "./project/ProjectEnrichmentService.ts"; +import * as ProjectService from "./project/ProjectService.ts"; +import * as ProjectStore from "./orchestration-v2/ProjectStore.ts"; +import * as ThreadManagement from "./orchestration-v2/ThreadManagementService.ts"; +import * as OrchestrationEventStore from "./persistence/OrchestrationEventStore.ts"; +import * as SqlitePersistence from "./persistence/Sqlite.ts"; +import { subscribeOrchestrationV2Shell } from "./ws.ts"; + +const ReplayTestLayer = OrchestrationEventStore.layer.pipe( + Layer.provideMerge(SqlitePersistence.layerMemory), +); + +it.effect("shell resume below the history replay floor sends a snapshot", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-01-05T00:00:00.000Z"; + const projectId = ProjectId.make("project:retention-floor"); + const insert = (version: number) => 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 ( + ${`retention-floor:${version}`}, 'project', ${projectId}, ${version}, 'project.meta-updated', + ${now}, 'server', ${JSON.stringify({ projectId, updatedAt: now })}, '{}', 2 + ) RETURNING sequence + `; + const [before] = yield* insert(0); + const [after] = yield* insert(1); + const resume = subscribeOrchestrationV2Shell({ + afterSequence: before!.sequence, + requestCompletionMarker: true, + }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.map((item) => item.kind as string), + Stream.runCollect, + ), + ), + Effect.provideService(ThreadManagement.ThreadManagementService, { + listActiveShellIndex: () => Effect.succeed([]), + readShellSnapshot: () => Effect.succeed(Effect.succeed({ schemaVersion: 1, threads: [] })), + 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, { + getShell: () => Effect.succeedNone, + } as unknown as ProjectService.ProjectService["Service"]), + Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, { + subscribeChanges: PubSub.subscribe(yield* PubSub.unbounded()), + } as ProjectEnrichment.ProjectEnrichmentService["Service"]), + ); + assert.deepEqual(Array.from(yield* resume), ["project.removed", "synchronized"]); + yield* sql` + INSERT INTO orchestration_v2_event_retention (thread_id, replay_floor) + VALUES ('thread:pruned', ${after!.sequence}) + `; + assert.deepEqual(Array.from(yield* resume), ["snapshot", "synchronized"]); + }).pipe(Effect.scoped, Effect.provide(ReplayTestLayer)), +); + +it.effect("shell resume falls back when retention commits during the replay read", () => + Effect.scoped( + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const now = "2026-01-05T00:00:00.000Z"; + const projectId = ProjectId.make("project:retention-race"); + const threadId = ThreadId.make("thread:retention-race"); + const insertProject = (version: number) => 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 ( + ${`retention-race:project:${version}`}, 'project', ${projectId}, ${version}, 'project.meta-updated', + ${now}, 'server', ${JSON.stringify({ projectId, updatedAt: now })}, '{}', 2 + ) RETURNING sequence + `; + const [before] = yield* insertProject(0); + const [pruned] = yield* 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 ( + 'retention-race:thread', 'thread', ${threadId}, 0, 'thread.metadata-updated', + ${now}, 'server', '{}', '{}', 2 + ) RETURNING sequence + `; + yield* insertProject(1); + + const replayReadStarted = yield* Deferred.make(); + const releaseReplayRead = yield* Deferred.make(); + let blocked = false; + const blockReplayRead: Statement.Transformer = (statement) => { + const [query] = statement.compile(); + if ( + !blocked && + query.includes("FROM orchestration_events") && + query.includes("ORDER BY sequence ASC") + ) { + blocked = true; + return Deferred.succeed(replayReadStarted, undefined).pipe( + Effect.andThen(Deferred.await(releaseReplayRead)), + Effect.as(statement), + ); + } + return Effect.succeed(statement); + }; + const resume = subscribeOrchestrationV2Shell({ + afterSequence: before!.sequence, + requestCompletionMarker: true, + }).pipe( + Effect.flatMap((stream) => + stream.pipe( + Stream.takeUntil((item) => item.kind === "synchronized"), + Stream.map((item) => item.kind as string), + Stream.runCollect, + ), + ), + Effect.provideService(ThreadManagement.ThreadManagementService, { + listActiveShellIndex: () => Effect.succeed([]), + readShellSnapshot: () => + Effect.succeed(Effect.succeed({ schemaVersion: 1, threads: [] })), + 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, { + getShell: () => Effect.succeedNone, + } as unknown as ProjectService.ProjectService["Service"]), + Effect.provideService(ProjectEnrichment.ProjectEnrichmentService, { + subscribeChanges: PubSub.subscribe(yield* PubSub.unbounded()), + } as ProjectEnrichment.ProjectEnrichmentService["Service"]), + Effect.provideService(Statement.CurrentTransformer, blockReplayRead), + ); + const resumedFiber = yield* Effect.forkScoped(resume); + yield* Deferred.await(replayReadStarted); + yield* sql.withTransaction( + Effect.gen(function* () { + yield* sql`DELETE FROM orchestration_events WHERE sequence = ${pruned!.sequence}`; + yield* sql` + INSERT INTO orchestration_v2_event_retention (thread_id, replay_floor) + VALUES (${threadId}, ${pruned!.sequence}) + `; + }), + ); + yield* Deferred.succeed(releaseReplayRead, undefined); + + assert.deepEqual(Array.from(yield* Fiber.join(resumedFiber)), ["snapshot", "synchronized"]); + }).pipe(Effect.provide(ReplayTestLayer)), + ), +); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 4781fcf78dde6..da4df23d25334 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -726,6 +726,9 @@ const canReplayPersistedRange = Effect.fnUntraced(function* ( if (replayGap < 0 || replayGap > maxGap) { return false; } + if ((yield* applicationEvents.replayFloor()) > afterSequence) { + return false; + } const stats = yield* applicationEvents.getReplayStats({ afterSequence, throughSequence: headSequence, @@ -917,6 +920,35 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs if (input.afterSequence > highWater) { return yield* snapshotThenLive(); } + const threadShellExists = Effect.suspend(() => + threadManagement.getThreadShell(input.threadId), + ).pipe( + Effect.map((shell) => shell !== null), + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: `Failed to locate orchestration V2 thread ${input.threadId}`, + cause, + }), + ), + ); + const resumeAfter = input.afterSequence; + const replayExpired = applicationEvents.replayFloor(input.threadId).pipe( + Effect.map((floor) => floor > resumeAfter), + Effect.mapError( + (cause) => + new OrchestrationV2GetThreadProjectionError({ + threadId: input.threadId, + message: `Failed to read orchestration V2 thread ${input.threadId} replay floor`, + cause, + }), + ), + ); + const expiredBeforeReplay = yield* replayExpired; + if (expiredBeforeReplay && (yield* threadShellExists)) { + return yield* snapshotThenLive(); + } const stats = yield* applicationEvents .getAgentReplayStats({ threadId: input.threadId, @@ -941,23 +973,16 @@ export const subscribeOrchestrationV2Thread = Effect.fn("ws.orchestrationV2.subs ) { return yield* snapshotThenLive(); } - if (stats.hasCreateEvent) { - const shell = yield* threadManagement.getThreadShell(input.threadId).pipe( - Effect.mapError( - (cause) => - new OrchestrationV2GetThreadProjectionError({ - threadId: input.threadId, - message: `Failed to locate recreated orchestration V2 thread ${input.threadId}`, - cause, - }), - ), - ); + if (stats.hasCreateEvent && (yield* threadShellExists)) { // Retained creation events may describe deleted threads; replace replay only if a snapshot exists. - if (shell !== null) return yield* snapshotThenLive(); + return yield* snapshotThenLive(); } const replay = coalesceLiveUpdatedEvents( yield* loadReplayThrough(input.afterSequence, highWater), ); + if (!expiredBeforeReplay && (yield* replayExpired) && (yield* threadShellExists)) { + return yield* snapshotThenLive(); + } const plan = decideThreadResume({ afterSequence: input.afterSequence, highWater, @@ -1159,40 +1184,40 @@ export const subscribeOrchestrationV2Shell = Effect.fn("ws.orchestrationV2.subsc Stream.concat(completionMarker, liveFrom(afterSequence)); const stream = yield* Effect.gen(function* () { - if (input.afterSequence === undefined) { + const snapshotThenLive = Effect.gen(function* () { const loaded = yield* loadSnapshot(); return composeShellStreamWithEnrichment({ initial: initialSnapshotItems(loaded), tail: completionThenLive(loaded.snapshot.snapshotSequence), enrichment: enrichmentRefreshes, }); + }); + if (input.afterSequence === undefined) { + return yield* snapshotThenLive; } const highWater = yield* applicationEvents.latestApplicationSequence; if (!(yield* canReplayPersistedRange(input.afterSequence, highWater, SHELL_RESUME_MAX_GAP))) { - const loaded = yield* loadSnapshot(); - return composeShellStreamWithEnrichment({ - initial: initialSnapshotItems(loaded), - tail: completionThenLive(loaded.snapshot.snapshotSequence), - enrichment: enrichmentRefreshes, - }); + return yield* snapshotThenLive; } 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)), - ); + 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 ((yield* applicationEvents.replayFloor()) > input.afterSequence) { + return yield* snapshotThenLive; + } + 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 9e5cf6c9d8a6c..6a26f9e6071e0 100644 --- a/docs/internals/overview.md +++ b/docs/internals/overview.md @@ -44,7 +44,7 @@ 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 +Events record orchestration decisions, and persisted projections hold the current read state. 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, @@ -62,6 +62,13 @@ Persisted events must remain decodable on replay. Changing a schema affects old startup as well as live RPC traffic. Compatibility work must account for stored history, not just what the newest client sends. +The log is not kept forever. [Compaction](../../apps/server/src/orchestration-v2/ProjectionMaintenance.ts) +deletes thread events older than a day, except anchors that other readers need, and raises the +thread's replay floor in the same transaction. A cursor below that floor gets a snapshot. Projections +cannot be rebuilt from events once anything has been pruned, so future data migrations must update +projections directly. Builds from before retention ignore replay floors and must not open a pruned +database. + ## Turn completion and checkpoints A provider turn ending and its follow-up work settling are separate milestones. Orchestration diff --git a/docs/user/background-service.md b/docs/user/background-service.md index c36527b2c425c..3c3132907650f 100644 --- a/docs/user/background-service.md +++ b/docs/user/background-service.md @@ -32,7 +32,10 @@ before updating; to match a remote client's version, follow [Updating T3 Code](./updating.md). Pass an exact version (`t3 update 0.0.42`) to pin one, `--channel nightly` to -switch trains, or `--allow-downgrade` to move backwards. `preview` is a +switch trains, or `--allow-downgrade` to move backwards. After one-day history +cleanup has pruned events, do not downgrade to a version from before that +cleanup; it cannot resume from the replay floor and may show incomplete +conversation history. `preview` is a maintainers' test train: its builds can be broken and are never offered as updates, so the installer and `t3 update` ask for confirmation before installing one. @@ -161,8 +164,9 @@ 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 keeps your conversations and removes internal event history it no +longer needs, such as streaming updates older than a day, while it runs. +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 20667eafddeefffdf908782ec1c736818319f786 Mon Sep 17 00:00:00 2001 From: Yannic Charlon <52761674+JustYannicc@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:06:59 +0100 Subject: [PATCH 2/5] fix(server): retain only referenced prompt anchors --- .../EventHistoryRetentionAnchors.test.ts | 11 ++++++ .../orchestration-v2/ProjectionMaintenance.ts | 36 +++++++++++++++---- .../persistence/OrchestrationEventStore.ts | 2 +- 3 files changed, 41 insertions(+), 8 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts b/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts index 86e5d34020fb8..e2304c41ed424 100644 --- a/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts +++ b/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts @@ -138,6 +138,7 @@ it.effect("retention keeps replay anchors and leaves project and thread projecti const recentAt = yield* DateTime.now; yield* sink.write({ events: [ + ...promptEvents(anchoredThreadId, recentAt, 3), visitedEvent(anchoredThread, "recent", recentAt), visitedEvent(importingThread, "recent", recentAt), visitedEvent({ ...archivedThread, archivedAt: oldAt }, "recent", recentAt), @@ -163,6 +164,16 @@ it.effect("retention keeps replay anchors and leaves project and thread projecti ]) { assert.include(anchoredIds, id); } + for (const suffix of ["subagent:0", "subagent-item:0"]) { + assert.notInclude(anchoredIds, `event:thread:retention-anchors:${suffix}`); + } + const compressedPrompt = anchored.find( + (stored) => stored.event.id === "event:thread:retention-anchors:subagent:3", + )?.event; + assert.equal( + compressedPrompt?.type === "subagent.updated" ? compressedPrompt.payload.prompt : null, + "Prompt B. ".repeat(40), + ); const fullPrompt = anchored.find( (stored) => stored.event.id === "event:thread:retention-anchors:subagent:1", )?.event; diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 16a83adf657a1..bd40e45e78e24 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -260,7 +260,8 @@ export const layer: Layer.Layer< 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; + const initialHighWater = bounds[0]?.event_sequence ?? 0; + let throughSequence = initialHighWater; let throughReceiptRowId = bounds[0]?.receipt_row_id ?? 0; // Older doctor builds without migration 77 can compact only superseded history. const retentionCutoff = (yield* hasReplayFloorTable) @@ -268,6 +269,7 @@ export const layer: Layer.Layer< DateTime.subtractDuration(yield* DateTime.now, EVENT_HISTORY_RETENTION), ) : ""; + const retainedPromptSequences = new Set(); const newestStreamKeys = new Set(); const retainedThreadIds = new Set(); const retainedReadStateThreadIds = new Set(); @@ -287,6 +289,7 @@ export const layer: Layer.Layer< readonly imported_legacy_thread: number; readonly expired: number; readonly retention_anchor: number; + readonly prompt_sequence: number | null; }>` SELECT event.sequence, @@ -295,6 +298,9 @@ export const layer: Layer.Layer< event.stream_id, event.event_type, event.occurred_at < ${retentionCutoff} AS expired, + CASE WHEN event.event_type IN ('subagent.updated', 'turn-item.updated') + THEN json_extract(event.payload_json, '$.promptSequence') + ELSE NULL END AS prompt_sequence, CASE WHEN event.occurred_at >= ${retentionCutoff} THEN 0 WHEN event.command_id IS NOT NULL AND EXISTS ( @@ -309,10 +315,6 @@ export const layer: Layer.Layer< WHEN event.event_type = 'run.updated' AND json_extract(event.payload_json, '$.status') = 'running' THEN 1 - WHEN event.event_type IN ('subagent.updated', 'turn-item.updated') - AND json_type(event.payload_json, '$.prompt') = 'text' - AND json_type(event.payload_json, '$.promptSequence') IS NULL - THEN 1 ELSE 0 END AS retention_anchor, CASE @@ -355,6 +357,7 @@ export const layer: Layer.Layer< if ( row.expired === 1 && row.retention_anchor === 0 && + !retainedPromptSequences.has(row.sequence) && row.application_event_version === 2 && row.aggregate_kind === "thread" && !isNewestInStream && @@ -365,6 +368,7 @@ export const layer: Layer.Layer< if (!replayFloors.has(row.stream_id)) replayFloors.set(row.stream_id, row.sequence); continue; } + if (row.prompt_sequence !== null) retainedPromptSequences.add(row.prompt_sequence); if (row.imported_legacy_thread === 1) { obsolete.push(row.sequence); } else if (row.application_event_version === 2) { @@ -412,7 +416,25 @@ export const layer: Layer.Layer< // Replay readers must see a deletion and its floor from the same commit. yield* sql.withTransaction( Effect.gen(function* () { - yield* sql`DELETE FROM orchestration_events WHERE sequence IN ${sql.in(obsolete)}`; + // A writer can add a prompt reference after the read page was released. + const deleted = yield* sql<{ sequence: number }>` + DELETE FROM orchestration_events AS candidate + WHERE sequence IN ${sql.in(obsolete)} + AND NOT ( + candidate.event_type IN ('subagent.updated', 'turn-item.updated') + AND COALESCE(json_type(candidate.payload_json, '$.prompt') = 'text', 0) + AND json_type(candidate.payload_json, '$.promptSequence') IS NULL + AND EXISTS ( + SELECT 1 FROM orchestration_events AS reference + WHERE reference.sequence > ${initialHighWater} + AND reference.stream_id = candidate.stream_id + AND reference.event_type = candidate.event_type + AND json_extract(reference.payload_json, '$.promptSequence') = candidate.sequence + ) + ) + RETURNING sequence + `; + deletedEventCount += deleted.length; for (const [threadId, floor] of replayFloors) { yield* sql` INSERT INTO orchestration_v2_event_retention (thread_id, replay_floor) @@ -423,7 +445,7 @@ export const layer: Layer.Layer< } }), ); - deletedEventCount += obsolete.length; + yield* Effect.yieldNow; } throughSequence = diff --git a/apps/server/src/persistence/OrchestrationEventStore.ts b/apps/server/src/persistence/OrchestrationEventStore.ts index e00541845f6ad..e6014960e202b 100644 --- a/apps/server/src/persistence/OrchestrationEventStore.ts +++ b/apps/server/src/persistence/OrchestrationEventStore.ts @@ -455,7 +455,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 only while their full-prompt anchor exists. const storedPayload = hint !== undefined && subagent !== undefined ? sql`CASE WHEN EXISTS ( From 17abbe4d2387c4c442e7f7b2f514c7cc9dd6603c Mon Sep 17 00:00:00 2001 From: Yannic Charlon <52761674+JustYannicc@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:13:34 +0100 Subject: [PATCH 3/5] fix(server): keep retention discovery on the reader worker --- .../src/orchestration-v2/FoundationPersistence.test.ts | 6 ++++-- apps/server/src/orchestration-v2/ProjectionMaintenance.ts | 2 +- 2 files changed, 5 insertions(+), 3 deletions(-) diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 770da19398485..eb0b7960bed71 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -287,10 +287,12 @@ it.effect("discovers on the reader worker and deletes on the writer", () => assert.equal(result.deletedEventCount, 148); assert.equal(result.deletedReceiptCount, 100); assert.deepEqual( - reads.filter((query) => /\bSELECT\b/.test(query)), + reads.filter((query) => /^\s*SELECT\b/.test(query)), [], ); - assert(writes.some((query) => query.includes("DELETE FROM orchestration_events"))); + assert( + [...reads, ...writes].some((query) => /^\s*DELETE FROM orchestration_events\b/.test(query)), + ); assert(writes.some((query) => query.includes("DELETE FROM orchestration_command_receipts"))); }).pipe( Effect.provide( diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index bd40e45e78e24..1f87cb8f55bd4 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -264,7 +264,7 @@ export const layer: Layer.Layer< let throughSequence = initialHighWater; let throughReceiptRowId = bounds[0]?.receipt_row_id ?? 0; // Older doctor builds without migration 77 can compact only superseded history. - const retentionCutoff = (yield* hasReplayFloorTable) + const retentionCutoff = (yield* hasReplayFloorTable.pipe(NodeSqliteClient.readOnly)) ? DateTime.formatIso( DateTime.subtractDuration(yield* DateTime.now, EVENT_HISTORY_RETENTION), ) From be66da5bc2569222f1755996afe0c5c1ae845273 Mon Sep 17 00:00:00 2001 From: Yannic Charlon <52761674+JustYannicc@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:18:55 +0100 Subject: [PATCH 4/5] docs: explain retention downgrade risk in user terms --- docs/user/background-service.md | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/docs/user/background-service.md b/docs/user/background-service.md index 3c3132907650f..a2dedb1bc6e10 100644 --- a/docs/user/background-service.md +++ b/docs/user/background-service.md @@ -34,8 +34,7 @@ before updating; to match a remote client's version, follow Pass an exact version (`t3 update 0.0.42`) to pin one, `--channel nightly` to switch trains, or `--allow-downgrade` to move backwards. After one-day history cleanup has pruned events, do not downgrade to a version from before that -cleanup; it cannot resume from the replay floor and may show incomplete -conversation history. `preview` is a +cleanup. An older version may show incomplete conversation history. `preview` is a maintainers' test train: its builds can be broken and are never offered as updates, so the installer and `t3 update` ask for confirmation before installing one. From d1ae8dcc920f363ecc6785b043133e7110d8bd15 Mon Sep 17 00:00:00 2001 From: Yannic Charlon <52761674+JustYannicc@users.noreply.github.com> Date: Sat, 31 Oct 2026 14:31:21 +0100 Subject: [PATCH 5/5] fix(server): record replay floors only for deleted events --- .../EventHistoryRetentionAnchors.test.ts | 36 +++++++++++++++++++ .../orchestration-v2/ProjectionMaintenance.ts | 14 ++++++-- 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts b/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts index e2304c41ed424..4220666b3991c 100644 --- a/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts +++ b/apps/server/src/orchestration-v2/EventHistoryRetentionAnchors.test.ts @@ -16,6 +16,7 @@ import * as Effect from "effect/Effect"; 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 OrchestrationEventStore from "../persistence/OrchestrationEventStore.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; @@ -281,3 +282,38 @@ it.effect("retention keeps a deleted thread tombstone for old cursors", () => assert.equal(resumed[0]?.kind === "event" ? resumed[0].event.type : null, "thread.deleted"); }).pipe(Effect.provide(layerTest)), ); + +it.effect("a prompt reference added before deletion does not advance the replay floor", () => + Effect.gen(function* () { + const sink = yield* EventSink.EventSinkV2; + const maintenance = yield* ProjectionMaintenance.ProjectionMaintenanceV2; + const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const oldAt = yield* DateTime.now; + const thread = makeThread(ThreadId.make("thread:prompt-reference-race"), oldAt); + yield* sink.write({ + events: [createdEvent(thread, oldAt), ...promptEvents(thread.id, oldAt, 1)], + }); + yield* TestClock.adjust(Duration.hours(25)); + const now = yield* DateTime.now; + yield* sink.write({ events: [visitedEvent(thread, "recent", now)] }); + let inserted = false; + const addReference: Statement.Transformer = (statement) => { + const [query] = statement.compile(); + if (!inserted && query.includes("DELETE FROM orchestration_events AS candidate")) { + inserted = true; + return sink + .write({ events: promptEvents(thread.id, now, 2) }) + .pipe(Effect.orDie, Effect.as(statement)); + } + return Effect.succeed(statement); + }; + const result = yield* maintenance + .compactEventStore() + .pipe(Effect.provideService(Statement.CurrentTransformer, addReference)); + assert.isTrue(inserted); + assert.equal(result.deletedEventCount, 0); + assert.equal(yield* applicationEvents.replayFloor(thread.id), 0); + assert.deepEqual(yield* sql`SELECT * FROM orchestration_v2_event_retention`, []); + }).pipe(Effect.provide(layerTest)), +); diff --git a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts index 1f87cb8f55bd4..ba61ae00caea1 100644 --- a/apps/server/src/orchestration-v2/ProjectionMaintenance.ts +++ b/apps/server/src/orchestration-v2/ProjectionMaintenance.ts @@ -349,7 +349,7 @@ export const layer: Layer.Layer< `.pipe(NodeSqliteClient.readOnly); const page = rows.slice(0, COMPACTION_PAGE_SIZE); const obsolete: number[] = []; - const replayFloors = new Map(); + const retentionCandidates = new Map(); for (const row of page) { const streamKey = `${row.aggregate_kind}\0${row.stream_id}`; const isNewestInStream = !newestStreamKeys.has(streamKey); @@ -365,7 +365,7 @@ export const layer: Layer.Layer< row.event_type !== "thread.deleted" ) { obsolete.push(row.sequence); - if (!replayFloors.has(row.stream_id)) replayFloors.set(row.stream_id, row.sequence); + retentionCandidates.set(row.sequence, row.stream_id); continue; } if (row.prompt_sequence !== null) retainedPromptSequences.add(row.prompt_sequence); @@ -435,6 +435,16 @@ export const layer: Layer.Layer< RETURNING sequence `; deletedEventCount += deleted.length; + const replayFloors = new Map(); + for (const row of deleted) { + const threadId = retentionCandidates.get(row.sequence); + if (threadId !== undefined) { + replayFloors.set( + threadId, + Math.max(replayFloors.get(threadId) ?? 0, row.sequence), + ); + } + } for (const [threadId, floor] of replayFloors) { yield* sql` INSERT INTO orchestration_v2_event_retention (thread_id, replay_floor)