diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts index 0999000ed4f..0e09fde8836 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts @@ -2643,3 +2643,166 @@ 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: "approval", + kind: "approval.requested", + summary: "Command approval requested", + payload: { + requestId: "approval-recreate-1", + requestKind: "command", + }, + 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); + assert.equal(yield* countRows("projection_pending_approvals"), 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); + assert.equal(yield* countRows("projection_pending_approvals"), 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..2b0faf9de09 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, @@ -1339,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) ?? diff --git a/apps/server/src/orchestration/commandInvariants.ts b/apps/server/src/orchestration/commandInvariants.ts index b59ded77f4f..4d8857e4af5 100644 --- a/apps/server/src/orchestration/commandInvariants.ts +++ b/apps/server/src/orchestration/commandInvariants.ts @@ -156,7 +156,20 @@ 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 (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 fea36b5717f..8a708842668 100644 --- a/apps/server/src/orchestration/decider.delete.test.ts +++ b/apps/server/src/orchestration/decider.delete.test.ts @@ -214,4 +214,92 @@ 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 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"), + 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"); + // 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("still blocks creating a thread id that is live (not deleted)", () => + Effect.gen(function* () { + const readModel = yield* seedReadModel; + const error = yield* Effect.flip( + decideOrchestrationCommand({ + command: { + type: "thread.create", + commandId: asCommandId("cmd-thread-recreate-live"), + threadId: asThreadId("thread-delete-1"), + projectId: asProjectId("project-delete"), + title: "Duplicate Of Live Thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5-codex", + }, + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + runtimeMode: "approval-required", + branch: null, + worktreePath: null, + createdAt: "2026-01-01T00:00:00.000Z", + }, + readModel, + }), + ); + expect(error.message).toContain("already exists"); + }), + ); });