From a3fb7665f55e54e8c6fc22821acee64059ba87b2 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 08:28:00 +0000 Subject: [PATCH 1/4] test(server): reproduce #30017 attachment cleanup under live run forks Deleting a thread queues cleanup of every attachment in its message rows, even when a live run fork still shows that prompt by reading through the source. These tests reproduce pingdotgg/t3code#30017: the source delete must leave the fork's inherited image alone, and deleting the fork afterwards must queue its cleanup. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/runtimeLayer.test.ts | 186 ++++++++++++++++++ 1 file changed, 186 insertions(+) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index b38fb737c91db..0385b2291231c 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -47,6 +47,7 @@ import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as SqlClient from "effect/sql/SqlClient"; +import { createAttachmentId } from "../attachmentStore.ts"; import * as CheckpointStore from "../checkpointing/CheckpointStore.ts"; import * as GitWorkflow from "../git/GitWorkflowService.ts"; import * as ServerConfig from "../config.ts"; @@ -2405,6 +2406,140 @@ it.layer(layerLegacyImportTest)("OrchestrationV2 legacy import", (it) => { ); }); +// A run fork stores no message rows; it shows its source's prompt, image included, by reading through. +const seedImagePromptRunFork = Effect.fn("test.seedImagePromptRunFork")(function* (key: string) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const sourceThreadId = ThreadId.make(`image-fork-source:${key}`); + const forkThreadId = ThreadId.make(`image-fork-target:${key}`); + const runId = RunId.make(`image-fork-run:${key}`); + const messageId = MessageId.make(`image-fork-message:${key}`); + const attachmentId = createAttachmentId(sourceThreadId, "png")!; + const text = "What is in this screenshot?"; + const attachments = [ + { + type: "image" as const, + id: attachmentId, + name: "screenshot.png", + mimeType: "image/png", + sizeBytes: 8, + }, + ]; + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`image-fork-create:${key}`), + threadId: sourceThreadId, + projectId: ProjectId.make(`image-fork-project:${key}`), + title: "Image source", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + yield* eventSink.write({ + events: [ + { + id: EventId.make(`image-fork-run:${key}`), + type: "run.created", + threadId: sourceThreadId, + runId, + occurredAt: now, + payload: { + id: runId, + threadId: sourceThreadId, + ordinal: 1, + providerInstanceId: modelSelection.instanceId, + modelSelection, + providerThreadId: null, + userMessageId: messageId, + rootNodeId: null, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }, + { + id: EventId.make(`image-fork-message:${key}`), + type: "message.updated", + threadId: sourceThreadId, + runId, + occurredAt: now, + payload: { + id: messageId, + threadId: sourceThreadId, + runId, + nodeId: null, + role: "user", + text, + attachments, + streaming: false, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + }, + }, + { + id: EventId.make(`image-fork-prompt:${key}`), + type: "turn-item.updated", + threadId: sourceThreadId, + runId, + occurredAt: now, + payload: { + id: TurnItemId.make(`image-fork-prompt:${key}`), + threadId: sourceThreadId, + runId, + nodeId: null, + providerThreadId: null, + providerTurnId: null, + nativeItemRef: null, + parentItemId: null, + ordinal: 1, + status: "completed", + title: null, + startedAt: now, + completedAt: now, + updatedAt: now, + type: "user_message", + createdBy: "user", + creationSource: "web", + messageId, + inputIntent: "turn_start", + text, + attachments, + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make(`image-fork-fork:${key}`), + sourceThreadId, + targetThreadId: forkThreadId, + sourcePoint: { type: "run", runId }, + createdBy: "user", + creationSource: "web", + }); + return { sourceThreadId, forkThreadId, attachmentId }; +}); + +const queuedAttachmentCleanup = Effect.fn("test.queuedAttachmentCleanup")(function* ( + commandId: CommandId, +) { + const outbox = yield* EffectOutbox.EffectOutboxV2; + return (yield* outbox.listByCommandId(commandId)).flatMap((effect) => + effect.request.type === "attachment.cleanup" ? effect.request.attachmentIds : [], + ); +}); + it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect("applies lifecycle commands idempotently and emits archive/removal shell deltas", () => Effect.gen(function* () { @@ -2722,6 +2857,57 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("deleting a thread keeps the attachments a live run fork still shows", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { sourceThreadId, forkThreadId, attachmentId } = + yield* seedImagePromptRunFork("delete-source"); + const deleteCommandId = CommandId.make("image-fork-delete-source"); + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: deleteCommandId, + threadId: sourceThreadId, + }); + const fork = yield* orchestrator.getThreadProjection(forkThreadId); + assert.isNull(fork.thread.deletedAt); + assert.include( + fork.visibleTurnItems.flatMap(({ item }) => + item.type === "user_message" ? item.attachments.map((file) => file.id) : [], + ), + attachmentId, + ); + assert.notInclude( + yield* queuedAttachmentCleanup(deleteCommandId), + attachmentId, + "deleting the source queued cleanup of an image its live fork still shows", + ); + }), + ); + + it.effect("deleting the last thread that shows an inherited attachment cleans it up", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const { sourceThreadId, forkThreadId, attachmentId } = + yield* seedImagePromptRunFork("delete-both"); + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make("image-fork-delete-both-source"), + threadId: sourceThreadId, + }); + const deleteForkCommandId = CommandId.make("image-fork-delete-both-fork"); + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: deleteForkCommandId, + threadId: forkThreadId, + }); + assert.include( + yield* queuedAttachmentCleanup(deleteForkCommandId), + attachmentId, + "deleting the last fork that shows the inherited image left it uncleaned", + ); + }), + ); + it.effect("keeps the branch pull request when linking another pull request", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 332fc589b4d856ad17a23af18c3f3ba5d9f59c75 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 08:39:57 +0000 Subject: [PATCH 2/4] fix(server): deleting a thread keeps attachments a live run fork shows A run fork shows its source's prompt files through the source thread, but getThreadAttachmentIds listed only the deleted thread's own files, so deleting the source removed images the fork still displayed. It now walks the run-fork family (forkedFrom up, indexed fork transfers down), applies each fork's run cutoff, and returns the files the thread shows that no other live thread does, so deleting the last thread that shows an inherited file cleans it up. Question-answer files follow the same walk. Fixes pingdotgg/t3code#30017 Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/ProjectionStore.ts | 157 ++++++++++++++++-- .../src/orchestration-v2/runtimeLayer.test.ts | 78 +++++++++ 2 files changed, 218 insertions(+), 17 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 3cffce15b70d6..4dc7b6e7bf37f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -1364,6 +1364,53 @@ function inheritedRunFork(thread: Pick; + readonly files: ReadonlyArray<{ + readonly threadId: ThreadId; + readonly runOrdinal: number | null; + readonly id: string; + }>; +}): Array { + const threads = new Map(input.threads.map((thread) => [thread.id, thread])); + const filesByThread = Map.groupBy(input.files, (file) => file.threadId); + const shownBy = (threadId: ThreadId) => { + const shown = new Set(); + const seen = new Set(); + let current: ThreadId | undefined = threadId; + let cutoff = Number.POSITIVE_INFINITY; + while (current !== undefined && !seen.has(current)) { + seen.add(current); + for (const file of filesByThread.get(current) ?? []) { + // Runless rows count as shown: over-keeping only delays cleanup. + if (file.runOrdinal === null || file.runOrdinal <= cutoff) shown.add(file.id); + } + const source: RunForkFamilyThread["source"] = threads.get(current)?.source ?? null; + if (source === null) break; + cutoff = source.runOrdinal; + current = source.threadId; + } + return shown; + }; + const kept = new Set( + input.threads.flatMap((thread) => + thread.deleted || thread.id === input.threadId ? [] : [...shownBy(thread.id)], + ), + ); + return [...shownBy(input.threadId)].filter((id) => !kept.has(id)); +} + function buildVisibleTurnItems(input: { readonly projection: OrchestrationV2ThreadProjection; readonly sourceProjection: OrchestrationV2ThreadProjection | null; @@ -5440,28 +5487,105 @@ export const layer: Layer.Layer = Effect.mapError(controlReadError(threadId)), ); + // The thread, its run-fork ancestors, and every run fork below them, found through the indexed + // fork transfers rather than a scan of all threads. Deleted members stay: forks read through them. + const selectRunForkFamily = (threadId: ThreadId) => + sql<{ + thread_id: ThreadId; + deleted_at: string | null; + source_thread_id: ThreadId | null; + source_run_ordinal: number | null; + }>` + WITH RECURSIVE ancestor(thread_id) AS ( + SELECT ${threadId} + UNION + SELECT json_extract(thread.payload_json, '$.forkedFrom.threadId') + FROM ancestor + JOIN orchestration_v2_projection_threads AS thread ON thread.thread_id = ancestor.thread_id + WHERE json_extract(thread.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(thread.payload_json, '$.lineage.sideChat') IS NOT 1 + ), family(thread_id) AS ( + SELECT thread_id FROM ancestor + UNION + SELECT fork.thread_id + FROM family + JOIN orchestration_v2_projection_context_transfers AS transfer + ON transfer.source_thread_id = family.thread_id AND transfer.type = 'fork' + JOIN orchestration_v2_projection_threads AS fork + ON fork.thread_id = transfer.target_thread_id + WHERE json_extract(fork.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(fork.payload_json, '$.forkedFrom.threadId') = family.thread_id + AND json_extract(fork.payload_json, '$.lineage.sideChat') IS NOT 1 + ) + SELECT + thread.thread_id, + thread.deleted_at, + source_run.thread_id AS source_thread_id, + source_run.ordinal AS source_run_ordinal + FROM family + JOIN orchestration_v2_projection_threads AS thread ON thread.thread_id = family.thread_id + LEFT JOIN orchestration_v2_projection_runs AS source_run + ON json_extract(thread.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(thread.payload_json, '$.lineage.sideChat') IS NOT 1 + AND source_run.run_id = json_extract(thread.payload_json, '$.forkedFrom.runId') + AND source_run.thread_id = json_extract(thread.payload_json, '$.forkedFrom.threadId') + `; + + // Prompt files and question-answer files, with the run that bounds which forks show them. + const selectThreadFiles = (threadIds: ReadonlyArray) => + sql<{ thread_id: ThreadId; run_ordinal: number | null; id: string }>` + SELECT message.thread_id, run.ordinal AS run_ordinal, + json_extract(attachment.value, '$.id') AS id + FROM orchestration_v2_projection_messages AS message + JOIN json_each(message.payload_json, '$.attachments') AS attachment + LEFT JOIN orchestration_v2_projection_runs AS run ON run.run_id = message.run_id + WHERE message.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) + UNION + SELECT item.thread_id, run.ordinal AS run_ordinal, + json_extract(attachment.value, '$.id') AS id + FROM orchestration_v2_projection_turn_items AS item + JOIN json_each(item.payload_json, '$.questionAnswer.attachmentsByQuestionId') AS question + JOIN json_each(question.value) AS attachment + LEFT JOIN orchestration_v2_projection_runs AS run ON run.run_id = item.run_id + WHERE item.thread_id IN (SELECT value FROM json_each(${encodeIdList(threadIds)})) + AND item.type = 'user_input_request' + `; + const getThreadAttachmentIds: ProjectionStoreV2Shape["getThreadAttachmentIds"] = (threadId) => Effect.all([ - sql<{ id: string }>` - SELECT DISTINCT json_extract(attachment.value, '$.id') AS id - FROM orchestration_v2_projection_messages AS message, - json_each(message.payload_json, '$.attachments') AS attachment - WHERE message.thread_id = ${threadId} - `, + selectRunForkFamily(threadId).pipe( + Effect.flatMap((family) => + selectThreadFiles([threadId, ...family.map((thread) => thread.thread_id)]).pipe( + Effect.map((files) => + unsharedThreadFileIds({ + threadId, + threads: family.map((thread) => ({ + id: thread.thread_id, + deleted: thread.deleted_at !== null, + source: + thread.source_thread_id === null || thread.source_run_ordinal === null + ? null + : { + threadId: thread.source_thread_id, + runOrdinal: thread.source_run_ordinal, + }, + })), + files: files.map((file) => ({ + threadId: file.thread_id, + runOrdinal: file.run_ordinal, + id: file.id, + })), + }), + ), + ), + ), + ), sql<{ payload_json: string }>` SELECT payload_json FROM orchestration_v2_projection_turn_items WHERE thread_id = ${threadId} AND type = 'dynamic_tool' AND (payload_json LIKE '%htmlRender%' OR payload_json LIKE ${`%${MCP_APP_OUTPUT_KEY}%`}) - `, - sql<{ id: string }>` - SELECT DISTINCT json_extract(attachment.value, '$.id') AS id - FROM orchestration_v2_projection_turn_items AS item, - json_each(item.payload_json, '$.questionAnswer.attachmentsByQuestionId') AS question, - json_each(question.value) AS attachment - WHERE item.thread_id = ${threadId} - AND item.type = 'user_input_request' `, // A guest's waiting files are pending uploads that no message row holds yet. sql<{ id: string }>` @@ -5473,11 +5597,10 @@ export const layer: Layer.Layer = AND json_extract(message.value, '$.status') = 'pending' `, ]).pipe( - Effect.map(([messages, renders, answerFiles, guestFiles]) => [ + Effect.map(([files, renders, guestFiles]) => [ ...new Set([ - ...messages.map((row) => row.id), + ...files, ...guestFiles.map((row) => row.id), - ...answerFiles.map((row) => row.id), ...threadHtmlRenderAttachmentIds( threadId, renders.map((row) => { diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 0385b2291231c..13e32873c02b4 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -2908,6 +2908,84 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("deleting a source releases the files it added after its fork's run", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const { sourceThreadId, attachmentId } = yield* seedImagePromptRunFork("after-fork-run"); + const laterRunId = RunId.make("image-fork-later-run"); + const laterMessageId = MessageId.make("image-fork-later-message"); + const laterAttachmentId = createAttachmentId(sourceThreadId, "png")!; + yield* eventSink.write({ + events: [ + { + id: EventId.make("image-fork-later-run"), + type: "run.created", + threadId: sourceThreadId, + runId: laterRunId, + occurredAt: now, + payload: { + id: laterRunId, + threadId: sourceThreadId, + ordinal: 2, + providerInstanceId: modelSelection.instanceId, + modelSelection, + providerThreadId: null, + userMessageId: laterMessageId, + rootNodeId: null, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }, + { + id: EventId.make("image-fork-later-message"), + type: "message.updated", + threadId: sourceThreadId, + runId: laterRunId, + occurredAt: now, + payload: { + id: laterMessageId, + threadId: sourceThreadId, + runId: laterRunId, + nodeId: null, + role: "user", + text: "And this one?", + attachments: [ + { + type: "image", + id: laterAttachmentId, + name: "later.png", + mimeType: "image/png", + sizeBytes: 8, + }, + ], + streaming: false, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + }, + }, + ], + }); + const deleteCommandId = CommandId.make("image-fork-after-fork-run-delete"); + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: deleteCommandId, + threadId: sourceThreadId, + }); + const cleaned = yield* queuedAttachmentCleanup(deleteCommandId); + assert.include(cleaned, laterAttachmentId); + assert.notInclude(cleaned, attachmentId); + }), + ); + it.effect("keeps the branch pull request when linking another pull request", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; From 55ad1ddb39c97a5f21cf6df36805b9cda75c897b Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 3 Nov 2026 09:01:58 +0000 Subject: [PATCH 3/4] fix(server): serialize thread deletions so a shared file is never leaked Deletion decides which files to release from reads taken before it commits. Deleting a source and its last run fork at the same time let each delete see the other thread as live, so neither released the file. A process-wide deletion permit on ThreadCommandExecutor now covers planning through commit for thread.delete (with its side-chat cascade) and project delete's per-thread delete. Also covers a two-level run-fork chain. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 2 +- .../orchestration-v2/ThreadCommandExecutor.ts | 19 ++- .../src/orchestration-v2/runtimeLayer.test.ts | 152 ++++++++++++++++++ apps/server/src/project/ProjectService.ts | 36 +++-- 4 files changed, 190 insertions(+), 19 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 600764704e57f..f39c47bd87b68 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -13713,7 +13713,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio Effect.flatMap((sideChatIds) => sideChatIds.reduce( (locked, sideChatId) => threadDispatch.withLock(sideChatId, locked), - dispatchWithReceiptEffect(command, sideChatIds), + threadDispatch.withDeletionLock(dispatchWithReceiptEffect(command, sideChatIds)), ), ), ), diff --git a/apps/server/src/orchestration-v2/ThreadCommandExecutor.ts b/apps/server/src/orchestration-v2/ThreadCommandExecutor.ts index 8bd1cc38d2de2..3d3c6c9ff999c 100644 --- a/apps/server/src/orchestration-v2/ThreadCommandExecutor.ts +++ b/apps/server/src/orchestration-v2/ThreadCommandExecutor.ts @@ -1,11 +1,26 @@ import type { ThreadId } from "@t3tools/contracts"; import * as KeyedLock from "@t3tools/shared/KeyedLock"; import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Semaphore from "effect/Semaphore"; export class ThreadCommandExecutor extends Context.Service< ThreadCommandExecutor, - KeyedLock.KeyedLock + KeyedLock.KeyedLock & { + /** + * Serializes deletion planning through commit because run forks show files their sources own. + * Take it inside the thread lock; never take a thread lock while holding it. + */ + readonly withDeletionLock: (effect: Effect.Effect) => Effect.Effect; + } >()("t3/orchestration-v2/ThreadCommandExecutor") {} -export const layer = Layer.effect(ThreadCommandExecutor, KeyedLock.make()); +export const layer = Layer.effect( + ThreadCommandExecutor, + Effect.gen(function* () { + const threads = yield* KeyedLock.make(); + const deletion = yield* Semaphore.make(1); + return { ...threads, withDeletionLock: deletion.withPermit }; + }), +); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 13e32873c02b4..5ef1965f0ef2b 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -2986,6 +2986,158 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("concurrently deleting a source and its last fork still cleans the shared file", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const { sourceThreadId, forkThreadId, attachmentId } = + yield* seedImagePromptRunFork("concurrent-delete"); + const sourceRead = yield* Deferred.make(); + const releaseSource = yield* Deferred.make(); + const getThreadAttachmentIds = projectionStore.getThreadAttachmentIds; + const withDeletionLock = executor.withDeletionLock; + let deletionLockRequests = 0; + // The source's delete holds its read until the fork's delete has read too or is queued behind it. + const readSpy = vi + .spyOn(projectionStore, "getThreadAttachmentIds") + .mockImplementation((threadId) => + getThreadAttachmentIds(threadId).pipe( + Effect.tap(() => + threadId === sourceThreadId + ? Deferred.succeed(sourceRead, undefined).pipe( + Effect.andThen(Deferred.await(releaseSource)), + ) + : Deferred.succeed(releaseSource, undefined), + ), + ), + ); + const observeDeletionLock: ThreadCommandExecutor.ThreadCommandExecutor["Service"]["withDeletionLock"] = + (effect) => + ++deletionLockRequests === 2 + ? Deferred.succeed(releaseSource, undefined).pipe( + Effect.andThen(withDeletionLock(effect)), + ) + : withDeletionLock(effect); + const lockSpy = vi + .spyOn(executor, "withDeletionLock") + .mockImplementation(observeDeletionLock); + const deleteSourceCommandId = CommandId.make("image-fork-concurrent-delete-source"); + const deleteForkCommandId = CommandId.make("image-fork-concurrent-delete-fork"); + yield* Effect.gen(function* () { + const deleteSource = yield* orchestrator + .dispatch({ + type: "thread.delete", + commandId: deleteSourceCommandId, + threadId: sourceThreadId, + }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Deferred.await(sourceRead); + const deleteFork = yield* orchestrator + .dispatch({ + type: "thread.delete", + commandId: deleteForkCommandId, + threadId: forkThreadId, + }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Fiber.join(deleteSource); + yield* Fiber.join(deleteFork); + }).pipe( + Effect.ensuring( + Effect.sync(() => { + readSpy.mockRestore(); + lockSpy.mockRestore(); + }), + ), + ); + assert.notInclude(yield* queuedAttachmentCleanup(deleteSourceCommandId), attachmentId); + assert.include( + yield* queuedAttachmentCleanup(deleteForkCommandId), + attachmentId, + "concurrent deletes of a source and its last fork leaked the shared image", + ); + }), + ); + + it.effect("a fork of a fork keeps the file its grandparent shows until it is deleted", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const now = yield* DateTime.now; + const { sourceThreadId, forkThreadId, attachmentId } = + yield* seedImagePromptRunFork("fork-chain"); + const forkRunId = RunId.make("image-fork-chain-fork-run"); + const forkMessageId = MessageId.make("image-fork-chain-fork-message"); + const grandchildThreadId = ThreadId.make("image-fork-chain-grandchild"); + yield* eventSink.write({ + events: [ + { + id: EventId.make("image-fork-chain-fork-run"), + type: "run.created", + threadId: forkThreadId, + runId: forkRunId, + occurredAt: now, + payload: { + id: forkRunId, + threadId: forkThreadId, + ordinal: 2, + providerInstanceId: modelSelection.instanceId, + modelSelection, + providerThreadId: null, + userMessageId: forkMessageId, + rootNodeId: null, + activeAttemptId: null, + status: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + checkpointId: null, + contextHandoffId: null, + }, + }, + { + id: EventId.make("image-fork-chain-fork-message"), + type: "message.updated", + threadId: forkThreadId, + runId: forkRunId, + occurredAt: now, + payload: { + id: forkMessageId, + threadId: forkThreadId, + runId: forkRunId, + nodeId: null, + role: "user", + text: "Now crop it.", + attachments: [], + streaming: false, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + }, + }, + ], + }); + yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make("image-fork-chain-fork-of-fork"), + sourceThreadId: forkThreadId, + targetThreadId: grandchildThreadId, + sourcePoint: { type: "run", runId: forkRunId }, + createdBy: "user", + creationSource: "web", + }); + const deleteThread = Effect.fn("test.deleteThread")(function* (threadId: ThreadId) { + const commandId = CommandId.make(`image-fork-chain-delete:${threadId}`); + yield* orchestrator.dispatch({ type: "thread.delete", commandId, threadId }); + return yield* queuedAttachmentCleanup(commandId); + }); + assert.notInclude(yield* deleteThread(sourceThreadId), attachmentId); + assert.notInclude(yield* deleteThread(forkThreadId), attachmentId); + assert.include(yield* deleteThread(grandchildThreadId), attachmentId); + }), + ); + it.effect("keeps the branch pull request when linking another pull request", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/project/ProjectService.ts b/apps/server/src/project/ProjectService.ts index 914456a500c51..3ebe0d8a8cacb 100644 --- a/apps/server/src/project/ProjectService.ts +++ b/apps/server/src/project/ProjectService.ts @@ -427,22 +427,26 @@ export const make = Effect.gen(function* () { commandId: CommandId.make(`${input.commandId}:delete-thread:${threadId}`), threadId, }; - const now = yield* DateTime.now; - const plan = yield* planThreadDeletion({ - command, - projection, - attachmentIds: yield* threadProjections.getThreadAttachmentIds(threadId), - now, - idAllocator, - }); - const committed = yield* eventSink.commitCommand({ - commandId: command.commandId, - commandType: command.type, - threadId, - acceptedAt: now, - events: plan.events, - effects: plan.effects, - }); + const committed = yield* threadCommands.withDeletionLock( + Effect.gen(function* () { + const now = yield* DateTime.now; + const plan = yield* planThreadDeletion({ + command, + projection, + attachmentIds: yield* threadProjections.getThreadAttachmentIds(threadId), + now, + idAllocator, + }); + return yield* eventSink.commitCommand({ + commandId: command.commandId, + commandType: command.type, + threadId, + acceptedAt: now, + events: plan.events, + effects: plan.effects, + }); + }), + ); if ( committed.receipt.threadId !== command.threadId || committed.receipt.commandType !== command.type From 0917124ec702e43f44995b929dac569f79e14b8c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 10 Nov 2026 23:36:07 +0000 Subject: [PATCH 4/4] test(server): trim fork attachment cleanup tests to the regressions they catch AI-assisted; verified with the tests and tools listed above. --- .../src/orchestration-v2/runtimeLayer.test.ts | 353 +++++------------- 1 file changed, 89 insertions(+), 264 deletions(-) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 4dac682747aff..ccafe793918af 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -2498,52 +2498,36 @@ it.layer(layerLegacyImportTest)("OrchestrationV2 legacy import", (it) => { ); }); -// A run fork stores no message rows; it shows its source's prompt, image included, by reading through. -const seedImagePromptRunFork = Effect.fn("test.seedImagePromptRunFork")(function* (key: string) { - const orchestrator = yield* Orchestrator.OrchestratorV2; +const imageAttachment = (threadId: ThreadId, name: string) => ({ + type: "image" as const, + id: createAttachmentId(threadId, "png")!, + name, + mimeType: "image/png", + sizeBytes: 8, +}); + +const writeCompletedUserRun = Effect.fn("test.writeCompletedUserRun")(function* (input: { + readonly threadId: ThreadId; + readonly key: string; + readonly ordinal: number; + readonly attachments: ReadonlyArray>; +}) { const eventSink = yield* EventSink.EventSinkV2; const now = yield* DateTime.now; - const sourceThreadId = ThreadId.make(`image-fork-source:${key}`); - const forkThreadId = ThreadId.make(`image-fork-target:${key}`); - const runId = RunId.make(`image-fork-run:${key}`); - const messageId = MessageId.make(`image-fork-message:${key}`); - const attachmentId = createAttachmentId(sourceThreadId, "png")!; - const text = "What is in this screenshot?"; - const attachments = [ - { - type: "image" as const, - id: attachmentId, - name: "screenshot.png", - mimeType: "image/png", - sizeBytes: 8, - }, - ]; - yield* orchestrator.dispatch({ - type: "thread.create", - commandId: CommandId.make(`image-fork-create:${key}`), - threadId: sourceThreadId, - projectId: ProjectId.make(`image-fork-project:${key}`), - title: "Image source", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - createdBy: "user", - creationSource: "web", - }); + const runId = RunId.make(`image-fork-run:${input.key}`); + const messageId = MessageId.make(`image-fork-message:${input.key}`); yield* eventSink.write({ events: [ { - id: EventId.make(`image-fork-run:${key}`), + id: EventId.make(`image-fork-run:${input.key}`), type: "run.created", - threadId: sourceThreadId, + threadId: input.threadId, runId, occurredAt: now, payload: { id: runId, - threadId: sourceThreadId, - ordinal: 1, + threadId: input.threadId, + ordinal: input.ordinal, providerInstanceId: modelSelection.instanceId, modelSelection, providerThreadId: null, @@ -2559,19 +2543,19 @@ const seedImagePromptRunFork = Effect.fn("test.seedImagePromptRunFork")(function }, }, { - id: EventId.make(`image-fork-message:${key}`), + id: EventId.make(`image-fork-message:${input.key}`), type: "message.updated", - threadId: sourceThreadId, + threadId: input.threadId, runId, occurredAt: now, payload: { id: messageId, - threadId: sourceThreadId, + threadId: input.threadId, runId, nodeId: null, role: "user", - text, - attachments, + text: "What is in this screenshot?", + attachments: [...input.attachments], streaming: false, createdBy: "user", creationSource: "web", @@ -2579,38 +2563,37 @@ const seedImagePromptRunFork = Effect.fn("test.seedImagePromptRunFork")(function updatedAt: now, }, }, - { - id: EventId.make(`image-fork-prompt:${key}`), - type: "turn-item.updated", - threadId: sourceThreadId, - runId, - occurredAt: now, - payload: { - id: TurnItemId.make(`image-fork-prompt:${key}`), - threadId: sourceThreadId, - runId, - nodeId: null, - providerThreadId: null, - providerTurnId: null, - nativeItemRef: null, - parentItemId: null, - ordinal: 1, - status: "completed", - title: null, - startedAt: now, - completedAt: now, - updatedAt: now, - type: "user_message", - createdBy: "user", - creationSource: "web", - messageId, - inputIntent: "turn_start", - text, - attachments, - }, - }, ], }); + return runId; +}); + +// A run fork stores no message rows; it shows its source's prompt, image included, by reading through. +const seedImagePromptRunFork = Effect.fn("test.seedImagePromptRunFork")(function* (key: string) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const sourceThreadId = ThreadId.make(`image-fork-source:${key}`); + const forkThreadId = ThreadId.make(`image-fork-target:${key}`); + const attachment = imageAttachment(sourceThreadId, "screenshot.png"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`image-fork-create:${key}`), + threadId: sourceThreadId, + projectId: ProjectId.make(`image-fork-project:${key}`), + title: "Image source", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + const runId = yield* writeCompletedUserRun({ + threadId: sourceThreadId, + key, + ordinal: 1, + attachments: [attachment], + }); yield* orchestrator.dispatch({ type: "thread.fork", commandId: CommandId.make(`image-fork-fork:${key}`), @@ -2620,7 +2603,14 @@ const seedImagePromptRunFork = Effect.fn("test.seedImagePromptRunFork")(function createdBy: "user", creationSource: "web", }); - return { sourceThreadId, forkThreadId, attachmentId }; + return { sourceThreadId, forkThreadId, attachmentId: attachment.id }; +}); + +const deleteThreadCleanup = Effect.fn("test.deleteThreadCleanup")(function* (threadId: ThreadId) { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const commandId = CommandId.make(`image-fork-delete:${threadId}`); + yield* orchestrator.dispatch({ type: "thread.delete", commandId, threadId }); + return yield* queuedAttachmentCleanup(commandId); }); const queuedAttachmentCleanup = Effect.fn("test.queuedAttachmentCleanup")(function* ( @@ -2963,132 +2953,46 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); - it.effect("deleting a thread keeps the attachments a live run fork still shows", () => + it.effect("deleting a source keeps what its live fork shows and releases later files", () => Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const { sourceThreadId, forkThreadId, attachmentId } = - yield* seedImagePromptRunFork("delete-source"); - const deleteCommandId = CommandId.make("image-fork-delete-source"); - yield* orchestrator.dispatch({ - type: "thread.delete", - commandId: deleteCommandId, + const { sourceThreadId, attachmentId } = yield* seedImagePromptRunFork("after-fork-run"); + const later = imageAttachment(sourceThreadId, "later.png"); + yield* writeCompletedUserRun({ threadId: sourceThreadId, + key: "after-fork-run-later", + ordinal: 2, + attachments: [later], }); - const fork = yield* orchestrator.getThreadProjection(forkThreadId); - assert.isNull(fork.thread.deletedAt); - assert.include( - fork.visibleTurnItems.flatMap(({ item }) => - item.type === "user_message" ? item.attachments.map((file) => file.id) : [], - ), - attachmentId, - ); - assert.notInclude( - yield* queuedAttachmentCleanup(deleteCommandId), - attachmentId, - "deleting the source queued cleanup of an image its live fork still shows", - ); + const cleaned = yield* deleteThreadCleanup(sourceThreadId); + assert.include(cleaned, later.id); + assert.notInclude(cleaned, attachmentId, "deleted an image its live fork still shows"); }), ); - it.effect("deleting the last thread that shows an inherited attachment cleans it up", () => + it.effect("a fork of a fork keeps the file its grandparent shows until it is deleted", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; const { sourceThreadId, forkThreadId, attachmentId } = - yield* seedImagePromptRunFork("delete-both"); - yield* orchestrator.dispatch({ - type: "thread.delete", - commandId: CommandId.make("image-fork-delete-both-source"), - threadId: sourceThreadId, - }); - const deleteForkCommandId = CommandId.make("image-fork-delete-both-fork"); - yield* orchestrator.dispatch({ - type: "thread.delete", - commandId: deleteForkCommandId, + yield* seedImagePromptRunFork("fork-chain"); + const grandchildThreadId = ThreadId.make("image-fork-chain-grandchild"); + const forkRunId = yield* writeCompletedUserRun({ threadId: forkThreadId, + key: "fork-chain-fork", + ordinal: 2, + attachments: [], }); - assert.include( - yield* queuedAttachmentCleanup(deleteForkCommandId), - attachmentId, - "deleting the last fork that shows the inherited image left it uncleaned", - ); - }), - ); - - it.effect("deleting a source releases the files it added after its fork's run", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const eventSink = yield* EventSink.EventSinkV2; - const now = yield* DateTime.now; - const { sourceThreadId, attachmentId } = yield* seedImagePromptRunFork("after-fork-run"); - const laterRunId = RunId.make("image-fork-later-run"); - const laterMessageId = MessageId.make("image-fork-later-message"); - const laterAttachmentId = createAttachmentId(sourceThreadId, "png")!; - yield* eventSink.write({ - events: [ - { - id: EventId.make("image-fork-later-run"), - type: "run.created", - threadId: sourceThreadId, - runId: laterRunId, - occurredAt: now, - payload: { - id: laterRunId, - threadId: sourceThreadId, - ordinal: 2, - providerInstanceId: modelSelection.instanceId, - modelSelection, - providerThreadId: null, - userMessageId: laterMessageId, - rootNodeId: null, - activeAttemptId: null, - status: "completed", - requestedAt: now, - startedAt: now, - completedAt: now, - checkpointId: null, - contextHandoffId: null, - }, - }, - { - id: EventId.make("image-fork-later-message"), - type: "message.updated", - threadId: sourceThreadId, - runId: laterRunId, - occurredAt: now, - payload: { - id: laterMessageId, - threadId: sourceThreadId, - runId: laterRunId, - nodeId: null, - role: "user", - text: "And this one?", - attachments: [ - { - type: "image", - id: laterAttachmentId, - name: "later.png", - mimeType: "image/png", - sizeBytes: 8, - }, - ], - streaming: false, - createdBy: "user", - creationSource: "web", - createdAt: now, - updatedAt: now, - }, - }, - ], - }); - const deleteCommandId = CommandId.make("image-fork-after-fork-run-delete"); yield* orchestrator.dispatch({ - type: "thread.delete", - commandId: deleteCommandId, - threadId: sourceThreadId, + type: "thread.fork", + commandId: CommandId.make("image-fork-chain-fork-of-fork"), + sourceThreadId: forkThreadId, + targetThreadId: grandchildThreadId, + sourcePoint: { type: "run", runId: forkRunId }, + createdBy: "user", + creationSource: "web", }); - const cleaned = yield* queuedAttachmentCleanup(deleteCommandId); - assert.include(cleaned, laterAttachmentId); - assert.notInclude(cleaned, attachmentId); + assert.notInclude(yield* deleteThreadCleanup(sourceThreadId), attachmentId); + assert.notInclude(yield* deleteThreadCleanup(forkThreadId), attachmentId); + assert.include(yield* deleteThreadCleanup(grandchildThreadId), attachmentId); }), ); @@ -3165,85 +3069,6 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); - it.effect("a fork of a fork keeps the file its grandparent shows until it is deleted", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const eventSink = yield* EventSink.EventSinkV2; - const now = yield* DateTime.now; - const { sourceThreadId, forkThreadId, attachmentId } = - yield* seedImagePromptRunFork("fork-chain"); - const forkRunId = RunId.make("image-fork-chain-fork-run"); - const forkMessageId = MessageId.make("image-fork-chain-fork-message"); - const grandchildThreadId = ThreadId.make("image-fork-chain-grandchild"); - yield* eventSink.write({ - events: [ - { - id: EventId.make("image-fork-chain-fork-run"), - type: "run.created", - threadId: forkThreadId, - runId: forkRunId, - occurredAt: now, - payload: { - id: forkRunId, - threadId: forkThreadId, - ordinal: 2, - providerInstanceId: modelSelection.instanceId, - modelSelection, - providerThreadId: null, - userMessageId: forkMessageId, - rootNodeId: null, - activeAttemptId: null, - status: "completed", - requestedAt: now, - startedAt: now, - completedAt: now, - checkpointId: null, - contextHandoffId: null, - }, - }, - { - id: EventId.make("image-fork-chain-fork-message"), - type: "message.updated", - threadId: forkThreadId, - runId: forkRunId, - occurredAt: now, - payload: { - id: forkMessageId, - threadId: forkThreadId, - runId: forkRunId, - nodeId: null, - role: "user", - text: "Now crop it.", - attachments: [], - streaming: false, - createdBy: "user", - creationSource: "web", - createdAt: now, - updatedAt: now, - }, - }, - ], - }); - yield* orchestrator.dispatch({ - type: "thread.fork", - commandId: CommandId.make("image-fork-chain-fork-of-fork"), - sourceThreadId: forkThreadId, - targetThreadId: grandchildThreadId, - sourcePoint: { type: "run", runId: forkRunId }, - createdBy: "user", - creationSource: "web", - }); - const deleteThread = Effect.fn("test.deleteThread")(function* (threadId: ThreadId) { - const commandId = CommandId.make(`image-fork-chain-delete:${threadId}`); - yield* orchestrator.dispatch({ type: "thread.delete", commandId, threadId }); - return yield* queuedAttachmentCleanup(commandId); - }); - assert.notInclude(yield* deleteThread(sourceThreadId), attachmentId); - assert.notInclude(yield* deleteThread(forkThreadId), attachmentId); - assert.include(yield* deleteThread(grandchildThreadId), attachmentId); - }), - ); - it.effect("keeps the branch pull request when linking another pull request", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2;