From 8a26586651affbf4c18dde420bf2a1db2770d7ad Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Mon, 20 Jul 2026 15:04:07 +0200 Subject: [PATCH 1/4] fix(server): allow re-creating a thread id after a soft delete MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When a draft thread's bootstrap turn start fails partway (for example, worktree preparation fails), the server cleans up by deleting the just-created thread. Thread deletion is a soft delete: the thread stays in the read model with a non-null deletedAt. Because requireThreadAbsent only checked for the thread's presence — not whether it was soft-deleted — a client retrying with the same draft thread id then failed with "Thread '' already exists and cannot be created twice.", leaving the user unable to start the conversation at all. Treat soft-deleted threads as absent so the retry succeeds. --- .../src/orchestration/commandInvariants.ts | 9 +++- .../src/orchestration/decider.delete.test.ts | 52 +++++++++++++++++++ 2 files changed, 60 insertions(+), 1 deletion(-) diff --git a/apps/server/src/orchestration/commandInvariants.ts b/apps/server/src/orchestration/commandInvariants.ts index b59ded77f4f..fe0496aaa80 100644 --- a/apps/server/src/orchestration/commandInvariants.ts +++ b/apps/server/src/orchestration/commandInvariants.ts @@ -156,7 +156,14 @@ export function requireThreadAbsent(input: { readonly command: OrchestrationCommand; readonly threadId: ThreadId; }): Effect.Effect { - if (!findThreadById(input.readModel, input.threadId)) { + const existing = findThreadById(input.readModel, input.threadId); + // A soft-deleted thread stays in the read model with `deletedAt` set, but it + // must not block re-creating a thread with the same id. This happens when a + // draft thread's bootstrap turn start fails partway (e.g. worktree prep + // fails): the server cleans up by deleting the just-created thread, and the + // client then retries with the same draft thread id. Treating the tombstone + // as absent lets the retry succeed instead of failing with "already exists". + if (!existing || existing.deletedAt !== null) { return Effect.void; } return Effect.fail( diff --git a/apps/server/src/orchestration/decider.delete.test.ts b/apps/server/src/orchestration/decider.delete.test.ts index fea36b5717f..6dfaeacfd24 100644 --- a/apps/server/src/orchestration/decider.delete.test.ts +++ b/apps/server/src/orchestration/decider.delete.test.ts @@ -214,4 +214,56 @@ it.layer(NodeServices.layer)("decider deletion flows", (it) => { expect(normalizeDeleteEvent(forcedResult)).toEqual(normalizeDeleteEvent(sequentialEvents)); }), ); + + it.effect("allows re-creating a thread id after it was soft-deleted", () => + Effect.gen(function* () { + const now = "2026-01-01T00:00:00.000Z"; + const threadId = asThreadId("thread-delete-1"); + let readModel = yield* seedReadModel; + let nextSequence = readModel.snapshotSequence; + + const projectDecided = function* (command: OrchestrationCommand) { + const decided = yield* decideOrchestrationCommand({ command, readModel }); + const events = Array.isArray(decided) ? decided : [decided]; + for (const event of events) { + nextSequence += 1; + readModel = yield* projectEvent(readModel, { ...event, sequence: nextSequence }); + } + return events; + }; + + // Soft-delete the thread (this is what the server does when a bootstrap + // turn start fails partway and cleans up the just-created thread). + yield* projectDecided({ + type: "thread.delete", + commandId: asCommandId("cmd-thread-delete-recreate"), + threadId, + }); + expect(readModel.threads.find((thread) => thread.id === threadId)?.deletedAt).not.toBeNull(); + + // Re-creating the same thread id (client retries with the same draft id) + // must succeed instead of failing with "already exists". + const recreatedEvents = yield* projectDecided({ + type: "thread.create", + commandId: asCommandId("cmd-thread-recreate"), + threadId, + projectId: asProjectId("project-delete"), + title: "Recreated Thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: now, + }); + + expect(recreatedEvents.map((event) => event.type)).toEqual(["thread.created"]); + const resurrected = readModel.threads.find((thread) => thread.id === threadId); + expect(resurrected?.deletedAt).toBeNull(); + expect(resurrected?.title).toBe("Recreated Thread"); + }), + ); }); From d96fd04ed4b40480afa2a361d7427c3a264a48fe Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Mon, 20 Jul 2026 15:22:24 +0200 Subject: [PATCH 2/4] fix(server): only resurrect content-free thread tombstones MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review: thread deletion is a soft delete and the persistent projection keeps child rows (messages, activities, proposed plans, checkpoints, sessions) keyed by threadId. Treating every soft-deleted thread as absent could resurrect an id whose child rows still exist, surfacing stale records on the re-created thread. Restrict reuse to a content-free tombstone — no messages, activities, proposed plans, checkpoints, session, or turn — which is exactly the state a draft thread is left in when a bootstrap turn start fails before the turn runs. A tombstone that owned content stays blocked. Added a negative test covering that case. --- .../src/orchestration/commandInvariants.ts | 41 ++++++++-- .../src/orchestration/decider.delete.test.ts | 74 ++++++++++++++++++- 2 files changed, 106 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration/commandInvariants.ts b/apps/server/src/orchestration/commandInvariants.ts index fe0496aaa80..b5056d8f3ef 100644 --- a/apps/server/src/orchestration/commandInvariants.ts +++ b/apps/server/src/orchestration/commandInvariants.ts @@ -151,19 +151,46 @@ export function requireThreadNotArchived(input: { ); } +/** + * A soft-deleted thread that never accumulated any content: no messages, + * activities, proposed plans, checkpoints, session, or turn. + * + * This is exactly the state a draft thread is left in when its bootstrap turn + * start fails before the turn ever runs — e.g. worktree preparation fails, so + * the server cleans up by deleting the just-created thread (see + * `dispatchBootstrapTurnStart` in `ws.ts`). The client then retries with the + * same draft thread id. + * + * Only such a content-free tombstone is safe to reuse: thread deletion is a + * soft delete, and the persistent projection keeps child rows keyed by + * `threadId` (messages, activities, proposed plans, checkpoints, sessions) even + * after deletion. Reusing the id of a tombstone that DID have content would + * surface those stale child records on the re-created thread, so those are left + * blocked. A content-free tombstone owns no child rows, so reuse is clean. + */ +function isReusableThreadTombstone(thread: OrchestrationThread): boolean { + return ( + thread.deletedAt !== null && + thread.latestTurn === null && + thread.session === null && + thread.messages.length === 0 && + thread.activities.length === 0 && + thread.proposedPlans.length === 0 && + thread.checkpoints.length === 0 + ); +} + export function requireThreadAbsent(input: { readonly readModel: OrchestrationReadModel; readonly command: OrchestrationCommand; readonly threadId: ThreadId; }): Effect.Effect { const existing = findThreadById(input.readModel, input.threadId); - // A soft-deleted thread stays in the read model with `deletedAt` set, but it - // must not block re-creating a thread with the same id. This happens when a - // draft thread's bootstrap turn start fails partway (e.g. worktree prep - // fails): the server cleans up by deleting the just-created thread, and the - // client then retries with the same draft thread id. Treating the tombstone - // as absent lets the retry succeed instead of failing with "already exists". - if (!existing || existing.deletedAt !== null) { + // Re-creating a thread id is allowed only when the existing thread is a + // content-free tombstone left behind by a failed bootstrap turn start (see + // isReusableThreadTombstone). A live thread — or a tombstone that still owns + // child projection rows — blocks creation. + if (!existing || isReusableThreadTombstone(existing)) { return Effect.void; } return Effect.fail( diff --git a/apps/server/src/orchestration/decider.delete.test.ts b/apps/server/src/orchestration/decider.delete.test.ts index 6dfaeacfd24..3375b379bb7 100644 --- a/apps/server/src/orchestration/decider.delete.test.ts +++ b/apps/server/src/orchestration/decider.delete.test.ts @@ -4,6 +4,7 @@ import { EventId, ProjectId, ThreadId, + TurnId, type OrchestrationCommand, type OrchestrationEvent, ProviderInstanceId, @@ -19,6 +20,7 @@ const asCommandId = (value: string): CommandId => CommandId.make(value); const asEventId = (value: string): EventId => EventId.make(value); const asProjectId = (value: string): ProjectId => ProjectId.make(value); const asThreadId = (value: string): ThreadId => ThreadId.make(value); +const asTurnId = (value: string): TurnId => TurnId.make(value); const seedReadModel = Effect.gen(function* () { const now = "2026-01-01T00:00:00.000Z"; @@ -232,8 +234,9 @@ it.layer(NodeServices.layer)("decider deletion flows", (it) => { return events; }; - // Soft-delete the thread (this is what the server does when a bootstrap - // turn start fails partway and cleans up the just-created thread). + // Soft-delete the freshly-created (content-free) thread — this is what the + // server does when a bootstrap turn start fails partway and cleans up the + // just-created thread, before any turn, message, or activity exists. yield* projectDecided({ type: "thread.delete", commandId: asCommandId("cmd-thread-delete-recreate"), @@ -266,4 +269,71 @@ it.layer(NodeServices.layer)("decider deletion flows", (it) => { expect(resurrected?.title).toBe("Recreated Thread"); }), ); + + it.effect("does not resurrect a soft-deleted thread that still owns content", () => + Effect.gen(function* () { + const now = "2026-01-01T00:00:00.000Z"; + const threadId = asThreadId("thread-delete-2"); + let readModel = yield* seedReadModel; + let nextSequence = readModel.snapshotSequence; + + const deleted = yield* decideOrchestrationCommand({ + command: { + type: "thread.delete", + commandId: asCommandId("cmd-thread-delete-content"), + threadId, + }, + readModel, + }); + for (const event of Array.isArray(deleted) ? deleted : [deleted]) { + nextSequence += 1; + readModel = yield* projectEvent(readModel, { ...event, sequence: nextSequence }); + } + + // Simulate a tombstone that owned content before deletion (here: a turn). + // Its child projection rows persist keyed by threadId, so the id must NOT + // be reused — otherwise the re-created thread would inherit stale records. + readModel = { + ...readModel, + threads: readModel.threads.map((thread) => + thread.id === threadId + ? { + ...thread, + latestTurn: { + turnId: asTurnId("turn-content"), + state: "completed" as const, + requestedAt: now, + startedAt: now, + completedAt: now, + assistantMessageId: null, + }, + } + : thread, + ), + }; + + const error = yield* Effect.flip( + decideOrchestrationCommand({ + command: { + type: "thread.create", + commandId: asCommandId("cmd-thread-recreate-content"), + threadId, + projectId: asProjectId("project-delete"), + title: "Recreated With Content", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: now, + }, + readModel, + }), + ); + expect(error.message).toContain("already exists"); + }), + ); }); From 28b58eaa66939009e2347f4c7a5a68b88157cb74 Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Mon, 20 Jul 2026 15:40:07 +0200 Subject: [PATCH 3/4] fix(server): purge stale child rows when recreating a thread id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review: instead of a content-sniffing heuristic in requireThreadAbsent (which the command read model cannot evaluate reliably — getCommandReadModel never loads messages/activities/checkpoints — and which wrongly blocked retries that recorded setup-script activities), allow re-creating any soft-deleted thread id and make the re-creation clean at the source. Each thread-keyed persistent projector (messages, proposed plans, activities, sessions, turns/checkpoints) now purges its own rows for the id on thread.created, matching the in-memory projector which already resets the thread to an empty record on re-create. The purge is a no-op for a brand-new id and only clears leftovers from a prior lifecycle, and because each projector touches only its own table it stays replay-safe under independent reprojection. Added a ProjectionPipeline test asserting a recreated thread inherits no stale messages/activities, and a decider test that a live (non-deleted) id is still rejected. --- .../Layers/ProjectionPipeline.test.ts | 158 ++++++++++++++++++ .../Layers/ProjectionPipeline.ts | 40 +++++ .../src/orchestration/commandInvariants.ts | 47 ++---- .../src/orchestration/decider.delete.test.ts | 60 ++----- 4 files changed, 224 insertions(+), 81 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 0999000ed4f..e6f65314e48 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -2643,3 +2643,161 @@ engineLayer("OrchestrationProjectionPipeline via engine dispatch", (it) => { }), ); }); + +it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-recreate-")))( + "OrchestrationProjectionPipeline", + (it) => { + it.effect("purges stale child rows when a soft-deleted thread id is recreated", () => + Effect.gen(function* () { + const projectionPipeline = yield* OrchestrationProjectionPipeline; + const eventStore = yield* OrchestrationEventStore; + const sql = yield* SqlClient.SqlClient; + const now = "2026-01-01T00:00:00.000Z"; + const threadId = ThreadId.make("thread-recreate"); + + const appendAndProject = (event: Parameters[0]) => + eventStore + .append(event) + .pipe(Effect.flatMap((savedEvent) => projectionPipeline.projectEvent(savedEvent))); + + yield* appendAndProject({ + type: "project.created", + eventId: EventId.make("evt-recreate-1"), + aggregateKind: "project", + aggregateId: ProjectId.make("project-recreate"), + occurredAt: now, + commandId: CommandId.make("cmd-recreate-1"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-recreate-1"), + metadata: {}, + payload: { + projectId: ProjectId.make("project-recreate"), + title: "Project Recreate", + workspaceRoot: "/tmp/project-recreate", + defaultModelSelection: null, + scripts: [], + createdAt: now, + updatedAt: now, + }, + }); + + const createThreadEvent = (eventId: string, commandId: string) => + ({ + type: "thread.created", + eventId: EventId.make(eventId), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make(commandId), + causationEventId: null, + correlationId: CorrelationId.make(commandId), + metadata: {}, + payload: { + threadId, + projectId: ProjectId.make("project-recreate"), + title: "Thread Recreate", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + runtimeMode: "full-access", + branch: null, + worktreePath: null, + createdAt: now, + updatedAt: now, + }, + }) satisfies Parameters[0]; + + // First lifecycle: create the thread and accumulate child rows. + yield* appendAndProject(createThreadEvent("evt-recreate-2", "cmd-recreate-2")); + + yield* appendAndProject({ + type: "thread.message-sent", + eventId: EventId.make("evt-recreate-3"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("cmd-recreate-3"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-recreate-3"), + metadata: {}, + payload: { + threadId, + messageId: MessageId.make("message-recreate"), + role: "user", + text: "first lifecycle message", + turnId: null, + streaming: false, + createdAt: now, + updatedAt: now, + }, + }); + + yield* appendAndProject({ + type: "thread.activity-appended", + eventId: EventId.make("evt-recreate-4"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("cmd-recreate-4"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-recreate-4"), + metadata: {}, + payload: { + threadId, + activity: { + id: EventId.make("activity-recreate"), + tone: "info", + kind: "setup-script.requested", + summary: "Starting setup script", + payload: {}, + turnId: null, + createdAt: now, + }, + }, + }); + + const countRows = (table: string) => + sql<{ readonly count: number }>` + SELECT COUNT(*) AS "count" FROM ${sql(table)} WHERE thread_id = ${threadId} + `.pipe(Effect.map((rows) => rows[0]?.count ?? 0)); + + assert.equal(yield* countRows("projection_thread_messages"), 1); + assert.equal(yield* countRows("projection_thread_activities"), 1); + + // Soft-delete (bootstrap cleanup), then recreate with the same id. + yield* appendAndProject({ + type: "thread.deleted", + eventId: EventId.make("evt-recreate-5"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: CommandId.make("cmd-recreate-5"), + causationEventId: null, + correlationId: CorrelationId.make("cmd-recreate-5"), + metadata: {}, + payload: { + threadId, + deletedAt: now, + }, + }); + + yield* appendAndProject(createThreadEvent("evt-recreate-6", "cmd-recreate-6")); + + // The re-created thread must not inherit stale child rows. + assert.equal(yield* countRows("projection_thread_messages"), 0); + assert.equal(yield* countRows("projection_thread_activities"), 0); + + const threadRows = yield* sql<{ + readonly threadId: string; + readonly deletedAt: string | null; + }>` + SELECT thread_id AS "threadId", deleted_at AS "deletedAt" + FROM projection_threads + WHERE thread_id = ${threadId} + `; + assert.deepEqual(threadRows, [{ threadId: "thread-recreate", deletedAt: null }]); + }), + ); + }, +); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index f12df850941..1bc1a97e817 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -811,6 +811,16 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti "applyThreadMessagesProjection", )(function* (event, attachmentSideEffects) { switch (event.type) { + case "thread.created": + // Re-creating a soft-deleted thread id reuses the same threadId (a + // draft's bootstrap retry). Clear any child rows left over from the + // prior lifecycle so the re-created thread starts clean, matching the + // in-memory projector which resets the thread to empty on create. + yield* projectionThreadMessageRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); + return; + case "thread.message-sent": { const existingMessage = yield* projectionThreadMessageRepository.getByMessageId({ messageId: event.payload.messageId, @@ -890,6 +900,13 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti "applyThreadProposedPlansProjection", )(function* (event, _attachmentSideEffects) { switch (event.type) { + case "thread.created": + // Clear stale proposed plans when a soft-deleted thread id is reused. + yield* projectionThreadProposedPlanRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); + return; + case "thread.proposed-plan-upserted": yield* projectionThreadProposedPlanRepository.upsert({ planId: event.payload.proposedPlan.id, @@ -941,6 +958,13 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti "applyThreadActivitiesProjection", )(function* (event, _attachmentSideEffects) { switch (event.type) { + case "thread.created": + // Clear stale activities when a soft-deleted thread id is reused. + yield* projectionThreadActivityRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); + return; + case "thread.activity-appended": yield* projectionThreadActivityRepository.upsert({ activityId: event.payload.activity.id, @@ -992,6 +1016,13 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const applyThreadSessionsProjection: ProjectorDefinition["apply"] = Effect.fn( "applyThreadSessionsProjection", )(function* (event, _attachmentSideEffects) { + if (event.type === "thread.created") { + // Clear a stale session when a soft-deleted thread id is reused. + yield* projectionThreadSessionRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); + return; + } if (event.type !== "thread.session-set") { return; } @@ -1011,6 +1042,15 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti "applyThreadTurnsProjection", )(function* (event, _attachmentSideEffects) { switch (event.type) { + case "thread.created": { + // Clear stale turns/checkpoints when a soft-deleted thread id is + // reused. + yield* projectionTurnRepository.deleteByThreadId({ + threadId: event.payload.threadId, + }); + return; + } + case "thread.turn-start-requested": { yield* projectionTurnRepository.replacePendingTurnStart({ threadId: event.payload.threadId, diff --git a/apps/server/src/orchestration/commandInvariants.ts b/apps/server/src/orchestration/commandInvariants.ts index b5056d8f3ef..4d8857e4af5 100644 --- a/apps/server/src/orchestration/commandInvariants.ts +++ b/apps/server/src/orchestration/commandInvariants.ts @@ -151,46 +151,25 @@ export function requireThreadNotArchived(input: { ); } -/** - * A soft-deleted thread that never accumulated any content: no messages, - * activities, proposed plans, checkpoints, session, or turn. - * - * This is exactly the state a draft thread is left in when its bootstrap turn - * start fails before the turn ever runs — e.g. worktree preparation fails, so - * the server cleans up by deleting the just-created thread (see - * `dispatchBootstrapTurnStart` in `ws.ts`). The client then retries with the - * same draft thread id. - * - * Only such a content-free tombstone is safe to reuse: thread deletion is a - * soft delete, and the persistent projection keeps child rows keyed by - * `threadId` (messages, activities, proposed plans, checkpoints, sessions) even - * after deletion. Reusing the id of a tombstone that DID have content would - * surface those stale child records on the re-created thread, so those are left - * blocked. A content-free tombstone owns no child rows, so reuse is clean. - */ -function isReusableThreadTombstone(thread: OrchestrationThread): boolean { - return ( - thread.deletedAt !== null && - thread.latestTurn === null && - thread.session === null && - thread.messages.length === 0 && - thread.activities.length === 0 && - thread.proposedPlans.length === 0 && - thread.checkpoints.length === 0 - ); -} - export function requireThreadAbsent(input: { readonly readModel: OrchestrationReadModel; readonly command: OrchestrationCommand; readonly threadId: ThreadId; }): Effect.Effect { const existing = findThreadById(input.readModel, input.threadId); - // Re-creating a thread id is allowed only when the existing thread is a - // content-free tombstone left behind by a failed bootstrap turn start (see - // isReusableThreadTombstone). A live thread — or a tombstone that still owns - // child projection rows — blocks creation. - if (!existing || isReusableThreadTombstone(existing)) { + // A soft-deleted thread (deletedAt set) must not block re-creating a thread + // with the same id. This happens when a draft thread's bootstrap turn start + // fails partway (e.g. worktree preparation fails): the server cleans up by + // deleting the just-created thread, and the client then retries with the same + // draft thread id. Treating the tombstone as absent lets the retry succeed + // instead of failing with "already exists". + // + // Re-creation is clean: the in-memory projector replaces the thread with a + // fresh, empty record on `thread.created`, and the persistent projectors each + // purge their own child rows for the id on `thread.created` (see + // ProjectionPipeline), so no stale messages/activities/turns/etc. from the + // prior lifecycle survive onto the re-created thread. + if (!existing || existing.deletedAt !== null) { return Effect.void; } return Effect.fail( diff --git a/apps/server/src/orchestration/decider.delete.test.ts b/apps/server/src/orchestration/decider.delete.test.ts index 3375b379bb7..8a708842668 100644 --- a/apps/server/src/orchestration/decider.delete.test.ts +++ b/apps/server/src/orchestration/decider.delete.test.ts @@ -4,7 +4,6 @@ import { EventId, ProjectId, ThreadId, - TurnId, type OrchestrationCommand, type OrchestrationEvent, ProviderInstanceId, @@ -20,7 +19,6 @@ const asCommandId = (value: string): CommandId => CommandId.make(value); const asEventId = (value: string): EventId => EventId.make(value); const asProjectId = (value: string): ProjectId => ProjectId.make(value); const asThreadId = (value: string): ThreadId => ThreadId.make(value); -const asTurnId = (value: string): TurnId => TurnId.make(value); const seedReadModel = Effect.gen(function* () { const now = "2026-01-01T00:00:00.000Z"; @@ -267,59 +265,27 @@ it.layer(NodeServices.layer)("decider deletion flows", (it) => { const resurrected = readModel.threads.find((thread) => thread.id === threadId); expect(resurrected?.deletedAt).toBeNull(); expect(resurrected?.title).toBe("Recreated Thread"); + // The in-memory projector resets the resurrected thread to empty; the + // persistent projectors purge their own child rows on thread.created + // (covered in ProjectionPipeline.test.ts) so no stale data survives. + expect(resurrected?.messages).toEqual([]); + expect(resurrected?.activities).toEqual([]); + expect(resurrected?.latestTurn).toBeNull(); + expect(resurrected?.session).toBeNull(); }), ); - it.effect("does not resurrect a soft-deleted thread that still owns content", () => + it.effect("still blocks creating a thread id that is live (not deleted)", () => Effect.gen(function* () { - const now = "2026-01-01T00:00:00.000Z"; - const threadId = asThreadId("thread-delete-2"); - let readModel = yield* seedReadModel; - let nextSequence = readModel.snapshotSequence; - - const deleted = yield* decideOrchestrationCommand({ - command: { - type: "thread.delete", - commandId: asCommandId("cmd-thread-delete-content"), - threadId, - }, - readModel, - }); - for (const event of Array.isArray(deleted) ? deleted : [deleted]) { - nextSequence += 1; - readModel = yield* projectEvent(readModel, { ...event, sequence: nextSequence }); - } - - // Simulate a tombstone that owned content before deletion (here: a turn). - // Its child projection rows persist keyed by threadId, so the id must NOT - // be reused — otherwise the re-created thread would inherit stale records. - readModel = { - ...readModel, - threads: readModel.threads.map((thread) => - thread.id === threadId - ? { - ...thread, - latestTurn: { - turnId: asTurnId("turn-content"), - state: "completed" as const, - requestedAt: now, - startedAt: now, - completedAt: now, - assistantMessageId: null, - }, - } - : thread, - ), - }; - + const readModel = yield* seedReadModel; const error = yield* Effect.flip( decideOrchestrationCommand({ command: { type: "thread.create", - commandId: asCommandId("cmd-thread-recreate-content"), - threadId, + commandId: asCommandId("cmd-thread-recreate-live"), + threadId: asThreadId("thread-delete-1"), projectId: asProjectId("project-delete"), - title: "Recreated With Content", + title: "Duplicate Of Live Thread", modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex", @@ -328,7 +294,7 @@ it.layer(NodeServices.layer)("decider deletion flows", (it) => { runtimeMode: "approval-required", branch: null, worktreePath: null, - createdAt: now, + createdAt: "2026-01-01T00:00:00.000Z", }, readModel, }), From 854c0d25bbd3237fe66dc75d4f11002cd9b5658f Mon Sep 17 00:00:00 2001 From: Ondra Pelech Date: Mon, 20 Jul 2026 15:48:29 +0200 Subject: [PATCH 4/4] fix(server): also purge stale pending approvals on thread recreate Address review: the pending-approvals projector was the one thread-keyed projection left unpurged on thread.created, so a reused thread id could carry a stale pendingApprovalCount into refreshThreadShellSummary. It clears its rows on thread.created too now (listByThreadId + deleteByRequestId, since the row is keyed by requestId). Extended the recreate pipeline test to cover pending approvals. --- .../Layers/ProjectionPipeline.test.ts | 13 +++++++++---- .../orchestration/Layers/ProjectionPipeline.ts | 16 ++++++++++++++++ 2 files changed, 25 insertions(+), 4 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index e6f65314e48..0e09fde8836 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -2747,10 +2747,13 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-recr threadId, activity: { id: EventId.make("activity-recreate"), - tone: "info", - kind: "setup-script.requested", - summary: "Starting setup script", - payload: {}, + tone: "approval", + kind: "approval.requested", + summary: "Command approval requested", + payload: { + requestId: "approval-recreate-1", + requestKind: "command", + }, turnId: null, createdAt: now, }, @@ -2764,6 +2767,7 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-recr assert.equal(yield* countRows("projection_thread_messages"), 1); assert.equal(yield* countRows("projection_thread_activities"), 1); + assert.equal(yield* countRows("projection_pending_approvals"), 1); // Soft-delete (bootstrap cleanup), then recreate with the same id. yield* appendAndProject({ @@ -2787,6 +2791,7 @@ it.layer(Layer.fresh(makeProjectionPipelinePrefixedTestLayer("t3-projection-recr // The re-created thread must not inherit stale child rows. assert.equal(yield* countRows("projection_thread_messages"), 0); assert.equal(yield* countRows("projection_thread_activities"), 0); + assert.equal(yield* countRows("projection_pending_approvals"), 0); const threadRows = yield* sql<{ readonly threadId: string; diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 1bc1a97e817..2b0faf9de09 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -1379,6 +1379,22 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti "applyPendingApprovalsProjection", )(function* (event, _attachmentSideEffects) { switch (event.type) { + case "thread.created": { + // Clear stale pending approvals when a soft-deleted thread id is + // reused. These rows are keyed by requestId (no thread-level delete), + // so drop each row the prior lifecycle left for this thread. + const staleApprovals = yield* projectionPendingApprovalRepository.listByThreadId({ + threadId: event.payload.threadId, + }); + yield* Effect.forEach( + staleApprovals, + (row) => + projectionPendingApprovalRepository.deleteByRequestId({ requestId: row.requestId }), + { concurrency: 1 }, + ); + return; + } + case "thread.activity-appended": { const requestId = extractActivityRequestId(event.payload.activity.payload) ??