From a8e763a74deab34c0e65d1782b2567119a11522f Mon Sep 17 00:00:00 2001 From: shivam <91240327+shivamhwp@users.noreply.github.com> Date: Fri, 13 Nov 2026 11:07:41 +0000 Subject: [PATCH 1/3] fix(server): forks of forks keep their deleted source's files Main keeps a deleted source's files while a live fork directly shows them. Forks further down lost them, a deleted source's files were never freed when its last fork went, and a normal fork could start while its source was being deleted. - The fork walk follows every run fork below a thread, through deleted forks, using the context-transfer index. - Deleting a fork also frees what its deleted sources kept only for it. - Thread and project deletions plan and commit one at a time, so sibling deletions see each other. - Every fork takes its source's lock and is refused once it is deleted. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/Orchestrator.ts | 38 +-- .../src/orchestration-v2/ProjectionStore.ts | 170 +++++++++---- .../orchestration-v2/ThreadCommandExecutor.ts | 24 +- .../src/orchestration-v2/runtimeLayer.test.ts | 231 +++++++++++++----- .../src/project/AgentSessionImporter.test.ts | 7 +- apps/server/src/project/ProjectService.ts | 2 +- 6 files changed, 341 insertions(+), 131 deletions(-) diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index defb1a42c892e..d77b8cb1874c3 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -5441,9 +5441,9 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), ), ); - // Only side chat forks lock the source; validate the modes copied here, not an earlier read. + // Read under the source's lock; validate the modes copied here, not an earlier read. yield* refuseAboveDispatchModeLimit(command, command.sourceThreadId, sourceProjection.thread); - if (command.sideChat === true && sourceProjection.thread.deletedAt !== null) { + if (sourceProjection.thread.deletedAt !== null) { return yield* new OrchestratorDispatchError({ commandId: command.commandId, commandType: command.type, @@ -15210,28 +15210,30 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio commandThreadId(command), dispatchWithReceiptEffect(command), ); - // A parent's lock comes before its side chats' locks, so a delete cascade and a side chat - // fork or command never interleave. A fork from a side chat or onto an existing thread would - // invert that order; dispatch rejects both, so they skip the source lock. - if ( - command.type === "thread.fork" && - command.sideChat === true && - command.sourceThreadId !== command.targetThreadId - ) { + // A fork holds its source's lock, so a source deletion and its fork never interleave. A + // parent's lock comes before its side chats' locks. A side chat fork from a side chat or a + // fork onto an existing thread would invert that order; dispatch rejects both, so they skip it. + if (command.type === "thread.fork" && command.sourceThreadId !== command.targetThreadId) { const source = yield* findThread(command.sourceThreadId); const target = yield* findThread(command.targetThreadId); - return yield* source === null || source.lineage.sideChat === true || target !== null + const skipsSourceLock = + source === null || + target !== null || + (command.sideChat === true && source.lineage.sideChat === true); + return yield* skipsSourceLock ? dispatch : threadDispatch.withLock(command.sourceThreadId, dispatch); } if (command.type === "thread.delete") - return yield* threadDispatch.withLock( - command.threadId, - liveSideChatIds(command.threadId).pipe( - Effect.flatMap((sideChatIds) => - sideChatIds.reduce( - (locked, sideChatId) => threadDispatch.withLock(sideChatId, locked), - dispatchWithReceiptEffect(command, sideChatIds), + return yield* threadDispatch.withDeletionLock( + threadDispatch.withLock( + command.threadId, + liveSideChatIds(command.threadId).pipe( + Effect.flatMap((sideChatIds) => + sideChatIds.reduce( + (locked, sideChatId) => threadDispatch.withLock(sideChatId, locked), + dispatchWithReceiptEffect(command, sideChatIds), + ), ), ), ), diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 841cd0a0b5b7d..e408f3d6b8afb 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6182,32 +6182,62 @@ export const layer: Layer.Layer = Effect.mapError(controlReadError(threadId)), ); - // A live run fork shows its source's history through the fork run without copying it. - const shownByLiveFork = (threadId: ThreadId, runIdColumn: string) => sql` - EXISTS ( - SELECT 1 - FROM orchestration_v2_projection_threads AS fork - JOIN orchestration_v2_projection_runs AS fork_run - ON fork_run.run_id = json_extract(fork.payload_json, '$.forkedFrom.runId') - LEFT JOIN orchestration_v2_projection_runs AS item_run - ON item_run.run_id = ${sql.literal(runIdColumn)} - WHERE fork_run.thread_id = ${threadId} - AND fork.deleted_at IS NULL - AND json_extract(fork.payload_json, '$.forkedFrom.type') = 'run' - AND json_extract(fork.payload_json, '$.forkedFrom.threadId') = ${threadId} - AND json_extract(fork.payload_json, '$.lineage.sideChat') IS NOT 1 - AND (${sql.literal(runIdColumn)} IS NULL OR item_run.ordinal <= fork_run.ordinal) - ) + // A run fork shows its source's history through the fork run without copying it, and every + // run fork below it shows that history too. Returns the last source run live forks show, with + // and without `leavingForkId`. + const lastRunsShownByLiveForks = (threadId: ThreadId, leavingForkId: ThreadId) => + sql<{ with_fork: number | null; without_fork: number | null }>` + WITH RECURSIVE showing(thread_id, deleted_at, through) AS ( + SELECT fork.thread_id, fork.deleted_at, fork_run.ordinal + FROM orchestration_v2_projection_context_transfers AS transfer + JOIN orchestration_v2_projection_threads AS fork + ON fork.thread_id = transfer.target_thread_id + JOIN orchestration_v2_projection_runs AS fork_run + ON fork_run.run_id = json_extract(fork.payload_json, '$.forkedFrom.runId') + WHERE transfer.source_thread_id = ${threadId} + AND transfer.type = 'fork' + AND fork_run.thread_id = ${threadId} + AND json_extract(fork.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(fork.payload_json, '$.forkedFrom.threadId') = ${threadId} + AND json_extract(fork.payload_json, '$.lineage.sideChat') IS NOT 1 + UNION + SELECT child.thread_id, child.deleted_at, showing.through + FROM showing + JOIN orchestration_v2_projection_context_transfers AS transfer + ON transfer.source_thread_id = showing.thread_id AND transfer.type = 'fork' + JOIN orchestration_v2_projection_threads AS child + ON child.thread_id = transfer.target_thread_id + WHERE json_extract(child.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(child.payload_json, '$.forkedFrom.threadId') = showing.thread_id + AND json_extract(child.payload_json, '$.lineage.sideChat') IS NOT 1 + ) + SELECT + MAX(through) AS with_fork, + MAX(CASE WHEN thread_id <> ${leavingForkId} THEN through END) AS without_fork + FROM showing + WHERE deleted_at IS NULL + `.pipe(Effect.map((rows) => rows[0] ?? { with_fork: null, without_fork: null })); + + // A row with no run predates every run (-1), so any live fork shows it. + const runOrdinalBetween = (runIdColumn: string, from: number, through: number) => sql` + COALESCE(( + SELECT run.ordinal FROM orchestration_v2_projection_runs AS run + WHERE run.run_id = ${sql.literal(runIdColumn)} + ), -1) BETWEEN ${from} AND ${through} `; - const getThreadAttachmentIds: ProjectionStoreV2Shape["getThreadAttachmentIds"] = (threadId) => - Effect.all([ + const runAttachmentIds = Effect.fn("ProjectionStore.runAttachmentIds")(function* ( + threadId: ThreadId, + from: number, + through: number, + ) { + const [messages, renders, answerFiles] = yield* 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} - AND NOT ${shownByLiveFork(threadId, "message.run_id")} + AND ${runOrdinalBetween("message.run_id", from, through)} `, sql<{ payload_json: string }>` SELECT item.payload_json @@ -6217,7 +6247,7 @@ export const layer: Layer.Layer = AND (item.payload_json LIKE '%htmlRender%' OR item.payload_json LIKE ${`%${MCP_APP_OUTPUT_KEY}%`} OR item.payload_json LIKE '%preview_recording_stop%') - AND NOT ${shownByLiveFork(threadId, "item.run_id")} + AND ${runOrdinalBetween("item.run_id", from, through)} `, sql<{ id: string }>` SELECT DISTINCT json_extract(attachment.value, '$.id') AS id @@ -6226,37 +6256,85 @@ export const layer: Layer.Layer = json_each(question.value) AS attachment WHERE item.thread_id = ${threadId} AND item.type = 'user_input_request' - AND NOT ${shownByLiveFork(threadId, "item.run_id")} + AND ${runOrdinalBetween("item.run_id", from, through)} `, + ]); + return [ + ...messages.map((row) => row.id), + ...answerFiles.map((row) => row.id), + ...threadHtmlRenderAttachmentIds( + threadId, + renders.map((row) => { + const item = parseEncodedPayload(row.payload_json); + return { + toolName: typeof item.toolName === "string" ? item.toolName : null, + output: item.output, + }; + }), + ), + ]; + }); + + // Deleting a fork also frees what its deleted sources kept only for it. + const getThreadAttachmentIds: ProjectionStoreV2Shape["getThreadAttachmentIds"] = Effect.fn( + "ProjectionStore.getThreadAttachmentIds", + )( + function* (threadId) { + const deletedSources = yield* sql<{ thread_id: ThreadId }>` + WITH RECURSIVE source(thread_id) AS ( + SELECT json_extract(thread.payload_json, '$.forkedFrom.threadId') + FROM orchestration_v2_projection_threads AS thread + WHERE thread.thread_id = ${threadId} + AND json_extract(thread.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(thread.payload_json, '$.lineage.sideChat') IS NOT 1 + UNION + SELECT json_extract(thread.payload_json, '$.forkedFrom.threadId') + FROM orchestration_v2_projection_threads AS thread + JOIN source USING (thread_id) + WHERE thread.deleted_at IS NOT NULL + AND json_extract(thread.payload_json, '$.forkedFrom.type') = 'run' + AND json_extract(thread.payload_json, '$.lineage.sideChat') IS NOT 1 + ) + SELECT thread.thread_id + FROM source + JOIN orchestration_v2_projection_threads AS thread USING (thread_id) + WHERE thread.deleted_at IS NOT NULL + `; + const afterRun = (ordinal: number | null) => (ordinal === null ? -1 : ordinal + 1); + const own = yield* lastRunsShownByLiveForks(threadId, threadId); + const released = yield* Effect.forEach(deletedSources, ({ thread_id: source }) => + lastRunsShownByLiveForks(source, threadId).pipe( + Effect.flatMap((shown) => + shown.with_fork === null + ? Effect.succeed([]) + : runAttachmentIds(source, afterRun(shown.without_fork), shown.with_fork), + ), + ), + ); // A guest's waiting files are pending uploads that no message row holds yet. - sql<{ id: string }>` - SELECT DISTINCT json_extract(attachment.value, '$.id') AS id - FROM orchestration_v2_projection_threads AS thread, - json_each(thread.payload_json, '$.guestShare.messages') AS message, - json_each(message.value, '$.attachments') AS attachment - WHERE thread.thread_id = ${threadId} - AND json_extract(message.value, '$.status') = 'pending' - `, - ]).pipe( - Effect.map(([messages, renders, answerFiles, guestFiles]) => [ + const guestFiles = yield* sql<{ id: string }>` + SELECT DISTINCT json_extract(attachment.value, '$.id') AS id + FROM orchestration_v2_projection_threads AS thread, + json_each(thread.payload_json, '$.guestShare.messages') AS message, + json_each(message.value, '$.attachments') AS attachment + WHERE thread.thread_id = ${threadId} + AND json_extract(message.value, '$.status') = 'pending' + `; + return [ ...new Set([ - ...messages.map((row) => row.id), - ...guestFiles.map((row) => row.id), - ...answerFiles.map((row) => row.id), - ...threadHtmlRenderAttachmentIds( + ...(yield* runAttachmentIds( threadId, - renders.map((row) => { - const item = parseEncodedPayload(row.payload_json); - return { - toolName: typeof item.toolName === "string" ? item.toolName : null, - output: item.output, - }; - }), - ), + afterRun(own.with_fork), + Number.MAX_SAFE_INTEGER, + )), + ...guestFiles.map((row) => row.id), + ...released.flat(), ]), - ]), - Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause })), - ); + ]; + }, + (effect, threadId) => + effect.pipe(Effect.mapError((cause) => new ProjectionStoreReadError({ threadId, cause }))), + ); const getMessagesById = Effect.fn("ProjectionStore.getMessagesById")( function* (threadId: ThreadId, messageIds: ReadonlyArray) { diff --git a/apps/server/src/orchestration-v2/ThreadCommandExecutor.ts b/apps/server/src/orchestration-v2/ThreadCommandExecutor.ts index 8bd1cc38d2de2..aa9a1f1ca7ab3 100644 --- a/apps/server/src/orchestration-v2/ThreadCommandExecutor.ts +++ b/apps/server/src/orchestration-v2/ThreadCommandExecutor.ts @@ -1,11 +1,31 @@ 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 & { + /** + * One thread deletion plans and commits at a time: each decides what its deleted fork + * sources keep for the forks still alive. Take it before any thread lock. + */ + readonly withDeletionLock: (effect: Effect.Effect) => Effect.Effect; + } >()("t3/orchestration-v2/ThreadCommandExecutor") {} -export const layer = Layer.effect(ThreadCommandExecutor, KeyedLock.make()); +export const make = Effect.gen(function* () { + const deletion = yield* Semaphore.make(1); + return ThreadCommandExecutor.of({ + ...(yield* KeyedLock.make()), + withDeletionLock: Effect.fn("ThreadCommandExecutor.withDeletionLock")(function* ( + effect: Effect.Effect, + ) { + return yield* deletion.withPermit(effect); + }), + }); +}); + +export const layer = Layer.effect(ThreadCommandExecutor, make); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index e62cd8725de92..fa2ee8bea7fc8 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 Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Path from "effect/Path"; import * as Queue from "effect/Queue"; +import * as Ref from "effect/Ref"; import * as Stream from "effect/Stream"; import * as TestClock from "effect/testing/TestClock"; import * as SqlClient from "effect/sql/SqlClient"; @@ -462,6 +463,7 @@ const layerProjectDeletionTest = Layer.mergeAll( RuntimeLayer.layer.pipe(Layer.provide(RuntimeLayer.layerProjectService)), RuntimeLayer.layerProjectService, RuntimeLayer.layerEventSink, + ProjectionStore.layer, ThreadCommandExecutor.layer, ).pipe( Layer.provide( @@ -491,11 +493,11 @@ const layerProjectDeletionTest = Layer.mergeAll( Layer.provide(McpProviderSessions.layer), Layer.provide(SqlitePersistence.layerMemory), Layer.provide(layerCheckpointStoreTest), - Layer.provide(layerServerConfig), + Layer.provideMerge(layerServerConfig), Layer.provide(ServerSettings.layerTest()), Layer.provide(layerTestProviderInstanceRegistry), Layer.provide(layerGitWorkflowTest), - Layer.provide(layerPlatformTest), + Layer.provideMerge(layerPlatformTest), ); it.layer(layerProjectDeletionTest)("project deletion during thread commands", (it) => { @@ -995,6 +997,167 @@ it.layer(layerProjectDeletionTest)("project deletion during thread commands", (i }).pipe(Effect.ensuring(Effect.sync(() => lockSpy.mockRestore()))); }), ); + + it.effect("deleted sources keep the html_render pages any live fork shows", () => + Effect.gen(function* () { + const projects = yield* ProjectService.ProjectService; + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projections = yield* ProjectionStore.ProjectionStoreV2; + const executor = yield* ThreadCommandExecutor.ThreadCommandExecutor; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const fileSystem = yield* FileSystem.FileSystem; + const projectId = ProjectId.make("html-fork-project"); + const sourceThreadId = ThreadId.make("html-fork-source"); + const forkThreadId = ThreadId.make("html-fork-target"); + yield* projects.create({ + commandId: CommandId.make("html-fork-project"), + projectId, + title: "Page family", + workspaceRoot: "/work/html-fork", + }); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make("html-fork-create"), + threadId: sourceThreadId, + projectId, + title: "Page source", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + const inherited = yield* writeHtmlRenderRun({ + threadId: sourceThreadId, + key: "inherited", + ordinal: 1, + }); + const fork = (sourceThreadId: ThreadId, runId: RunId, targetThreadId: ThreadId) => + orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make(`html-fork:${targetThreadId}`), + sourceThreadId, + targetThreadId, + sourcePoint: { type: "run", runId }, + createdBy: "user", + creationSource: "web", + }); + yield* fork(sourceThreadId, inherited.runId, forkThreadId); + const forkPage = yield* writeHtmlRenderRun({ + threadId: forkThreadId, + key: "fork", + ordinal: 1, + }); + const [first, second] = [ThreadId.make("html-fork-first"), ThreadId.make("html-fork-second")]; + yield* fork(forkThreadId, forkPage.runId, first); + // Project deletion goes newest first, so it reaches `second` before `first`. + yield* TestClock.adjust("1 second"); + yield* fork(forkThreadId, forkPage.runId, second); + const firstRun = yield* writeHtmlRenderRun({ threadId: first, key: "first", ordinal: 1 }); + const later = yield* writeHtmlRenderRun({ + threadId: sourceThreadId, + key: "later", + ordinal: 2, + }); + const deleteThread = (threadId: ThreadId) => + orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make(`html-fork-delete:${threadId}`), + threadId, + }); + const pagesOnDisk = Effect.gen(function* () { + yield* worker.drain(); + return yield* Effect.forEach([inherited, forkPage, later], (page) => + fileSystem.exists(page.pagePath), + ); + }); + yield* deleteThread(forkThreadId); + yield* deleteThread(sourceThreadId); + assert.deepEqual(yield* pagesOnDisk, [true, true, false]); + + // `first`'s delete pauses after reading what to free. A fork of it and the project delete, + // which reaches `second`, must each wait on a lock it holds rather than decide on the same + // state; `blocked` hears each request for one. + const firstPlanned = yield* Deferred.make(); + const releaseFirst = yield* Deferred.make(); + const blocked = yield* Queue.unbounded(); + const readAttachmentIds = projections.getThreadAttachmentIds; + const attachmentSpy = vi + .spyOn(projections, "getThreadAttachmentIds") + .mockImplementation((threadId) => + threadId === first + ? readAttachmentIds(threadId).pipe( + Effect.tap(() => Deferred.succeed(firstPlanned, undefined)), + Effect.tap(() => Deferred.await(releaseFirst)), + ) + : readAttachmentIds(threadId), + ); + const deletionHeld = yield* Ref.make(false); + const requestsHeld = (held: Effect.Effect) => + Effect.all([Deferred.isDone(firstPlanned), held]).pipe( + Effect.flatMap(([planned, isHeld]) => + planned && isHeld ? Queue.offer(blocked, undefined) : Effect.void, + ), + ); + const { withLock, withDeletionLock } = executor; + const observeLock: ThreadCommandExecutor.ThreadCommandExecutor["Service"]["withLock"] = ( + key, + effect, + ) => requestsHeld(Effect.succeed(key === first)).pipe(Effect.andThen(withLock(key, effect))); + const observeDeletion: ThreadCommandExecutor.ThreadCommandExecutor["Service"]["withDeletionLock"] = + (effect) => + requestsHeld(Ref.get(deletionHeld)).pipe( + Effect.andThen( + withDeletionLock( + Ref.set(deletionHeld, true).pipe( + Effect.andThen(effect), + Effect.ensuring(Ref.set(deletionHeld, false)), + ), + ), + ), + ); + const lockSpy = vi.spyOn(executor, "withLock").mockImplementation(observeLock); + const deletionSpy = vi + .spyOn(executor, "withDeletionLock") + .mockImplementation(observeDeletion); + const forkOfDeleted = yield* Effect.gen(function* () { + const firstFiber = yield* deleteThread(first).pipe( + Effect.forkChild({ startImmediately: true }), + ); + yield* Deferred.await(firstPlanned); + const forkFiber = yield* fork(first, firstRun.runId, ThreadId.make("html-fork-late")).pipe( + Effect.match({ + onSuccess: () => "created", + onFailure: (error) => ("cause" in error ? String(error.cause) : error._tag), + }), + Effect.forkChild({ startImmediately: true }), + ); + yield* Effect.raceFirst(Queue.take(blocked), Fiber.await(forkFiber)); + const projectFiber = yield* projects + .delete({ commandId: CommandId.make("html-fork-project-delete"), projectId, force: true }) + .pipe(Effect.forkChild({ startImmediately: true })); + yield* Effect.raceFirst(Queue.take(blocked), Fiber.await(projectFiber)); + yield* Deferred.succeed(releaseFirst, undefined); + yield* Fiber.join(firstFiber); + yield* Fiber.join(projectFiber); + return yield* Fiber.join(forkFiber); + }).pipe( + Effect.ensuring( + Effect.sync(() => { + lockSpy.mockRestore(); + deletionSpy.mockRestore(); + attachmentSpy.mockRestore(); + }), + ), + ); + assert.deepEqual( + { afterLastForks: yield* pagesOnDisk, forkOfDeleted }, + { afterLastForks: [false, false, false], forkOfDeleted: `Thread ${first} was deleted.` }, + ); + }), + ); }); const layerSharedApplicationDataPlaneTest = Layer.mergeAll( @@ -2874,6 +3037,11 @@ const writeHtmlRenderRun = Effect.fn("test.writeHtmlRenderRun")(function* (input const runId = RunId.make(`html-fork-run:${input.key}`); const messageId = MessageId.make(`html-fork-message:${input.key}`); const attachmentId = createAttachmentId(input.threadId, "html")!; + const fileSystem = yield* FileSystem.FileSystem; + const { attachmentsDir } = yield* ServerConfig.ServerConfig; + const pagePath = (yield* Path.Path).join(attachmentsDir, `${attachmentId}.html`); + yield* fileSystem.makeDirectory(attachmentsDir, { recursive: true }); + yield* fileSystem.writeFileString(pagePath, "

Page

"); yield* eventSink.write({ events: [ { @@ -2955,67 +3123,10 @@ const writeHtmlRenderRun = Effect.fn("test.writeHtmlRenderRun")(function* (input }, ], }); - return { runId, attachmentId }; + return { runId, pagePath }; }); it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { - it.effect("deleting a source keeps the html_render page its live fork shows", () => - Effect.gen(function* () { - const orchestrator = yield* Orchestrator.OrchestratorV2; - const outbox = yield* EffectOutbox.EffectOutboxV2; - const sourceThreadId = ThreadId.make("html-fork-source"); - const forkThreadId = ThreadId.make("html-fork-target"); - yield* orchestrator.dispatch({ - type: "thread.create", - commandId: CommandId.make("html-fork-create"), - threadId: sourceThreadId, - projectId: ProjectId.make("html-fork-project"), - title: "Page source", - modelSelection, - runtimeMode: "full-access", - interactionMode: "default", - branch: null, - worktreePath: null, - createdBy: "user", - creationSource: "web", - }); - const inherited = yield* writeHtmlRenderRun({ - threadId: sourceThreadId, - key: "inherited", - ordinal: 1, - }); - yield* orchestrator.dispatch({ - type: "thread.fork", - commandId: CommandId.make("html-fork-fork"), - sourceThreadId, - targetThreadId: forkThreadId, - sourcePoint: { type: "run", runId: inherited.runId }, - createdBy: "user", - creationSource: "web", - }); - const later = yield* writeHtmlRenderRun({ - threadId: sourceThreadId, - key: "later", - ordinal: 2, - }); - const deleteCommandId = CommandId.make("html-fork-delete-source"); - yield* orchestrator.dispatch({ - type: "thread.delete", - commandId: deleteCommandId, - threadId: sourceThreadId, - }); - const cleaned = (yield* outbox.listByCommandId(deleteCommandId)).flatMap((effect) => - effect.request.type === "attachment.cleanup" ? effect.request.attachmentIds : [], - ); - assert.include(cleaned, later.attachmentId); - assert.notInclude( - cleaned, - inherited.attachmentId, - "deleted a page its live fork still shows", - ); - }), - ); - it.effect("applies lifecycle commands idempotently and emits archive/removal shell deltas", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/apps/server/src/project/AgentSessionImporter.test.ts b/apps/server/src/project/AgentSessionImporter.test.ts index b05a0a0504cdd..7e7040c091a4b 100644 --- a/apps/server/src/project/AgentSessionImporter.test.ts +++ b/apps/server/src/project/AgentSessionImporter.test.ts @@ -15,7 +15,6 @@ import { type OrchestrationV2ProviderThread, } from "@t3tools/contracts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; -import * as KeyedLock from "@t3tools/shared/KeyedLock"; import * as AcpRegistryRuntimeCoordinator from "@t3tools/provider-acp-registry/server/AcpRegistryRuntimeCoordinator"; import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as Context from "effect/Context"; @@ -182,7 +181,7 @@ const makeImporter = Effect.fn(function* ( candidates: ReadonlyArray>, beforeTransaction: Effect.Effect = Effect.void, getAdapter?: ProviderAdapterRegistry.ProviderAdapterRegistryV2Shape["get"], - threadCommands?: KeyedLock.KeyedLock, + threadCommands?: ThreadCommandExecutor.ThreadCommandExecutor["Service"], settings: Layer.Layer< ServerSettings.ServerSettingsService, ServerSettingsError @@ -259,7 +258,7 @@ const makeImporter = Effect.fn(function* ( ), Layer.succeed( ThreadCommandExecutor.ThreadCommandExecutor, - threadCommands ?? (yield* KeyedLock.make()), + threadCommands ?? (yield* ThreadCommandExecutor.make), ), ), ), @@ -685,7 +684,7 @@ const importDuringCommand = Effect.fn(function* ( const projections = yield* ProjectionStore.ProjectionStoreV2; const sink = yield* EventSink.EventSinkV2; const threadId = yield* seedOwner(candidate(CODEX_MODEL_SELECTION, name)); - const lock = yield* KeyedLock.make(); + const lock = yield* ThreadCommandExecutor.make; const loaded = yield* Deferred.make(); const importerCommitting = yield* Deferred.make(); const commandWritten = yield* Deferred.make(); diff --git a/apps/server/src/project/ProjectService.ts b/apps/server/src/project/ProjectService.ts index 464454ebba763..1026f82abe75a 100644 --- a/apps/server/src/project/ProjectService.ts +++ b/apps/server/src/project/ProjectService.ts @@ -517,7 +517,7 @@ export const make = Effect.gen(function* () { projectThreads, (thread) => threadCommands - .withLock(thread.id, deleteChildThread(input, thread.id)) + .withDeletionLock(threadCommands.withLock(thread.id, deleteChildThread(input, thread.id))) .pipe( Effect.mapError( (cause) => From 5b85e99ee7310387fb19c417b20fdb0d2a853321 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 02:44:20 +0000 Subject: [PATCH 2/3] style(server): format merged attachment retention lookup --- .../src/orchestration-v2/ProjectionStore.ts | 16 +++++++++------- 1 file changed, 9 insertions(+), 7 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 244820e218307..fafbc1e8ded67 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6408,14 +6408,16 @@ export const layer: Layer.Layer = `; const afterRun = (ordinal: number | null) => (ordinal === null ? -1 : ordinal + 1); const own = yield* lastRunsShownByLiveForks(threadId, threadId); - const released = yield* Effect.forEach(options?.includeShownByLiveFork ? [] : deletedSources, ({ thread_id: source }) => - lastRunsShownByLiveForks(source, threadId).pipe( - Effect.flatMap((shown) => - shown.with_fork === null - ? Effect.succeed([]) - : runAttachmentIds(source, afterRun(shown.without_fork), shown.with_fork), + const released = yield* Effect.forEach( + options?.includeShownByLiveFork ? [] : deletedSources, + ({ thread_id: source }) => + lastRunsShownByLiveForks(source, threadId).pipe( + Effect.flatMap((shown) => + shown.with_fork === null + ? Effect.succeed([]) + : runAttachmentIds(source, afterRun(shown.without_fork), shown.with_fork), + ), ), - ), ); // A guest's waiting files are pending uploads that no message row holds yet. const guestFiles = yield* sql<{ id: string }>` From 1b08ca5ecac9c04d8d405e57d727f4e289e8a282 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sun, 15 Nov 2026 02:53:10 +0000 Subject: [PATCH 3/3] fix(server): retain reused files while earlier forks show them --- .../src/orchestration-v2/ProjectionStore.ts | 15 ++++++-- .../src/orchestration-v2/runtimeLayer.test.ts | 35 +++++++++++++++++++ 2 files changed, 48 insertions(+), 2 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index fafbc1e8ded67..0c854428649dd 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -6381,6 +6381,17 @@ export const layer: Layer.Layer = ]; }); + const unshownRunAttachmentIds = Effect.fn("ProjectionStore.unshownRunAttachmentIds")(function* ( + threadId: ThreadId, + from: number, + through: number, + ) { + const candidates = yield* runAttachmentIds(threadId, from, through); + if (from <= -1 || candidates.length === 0) return candidates; + const retained = new Set(yield* runAttachmentIds(threadId, -1, from - 1)); + return candidates.filter((id) => !retained.has(id)); + }); + // Deleting a fork also frees what its deleted sources kept only for it. const getThreadAttachmentIds: ProjectionStoreV2Shape["getThreadAttachmentIds"] = Effect.fn( "ProjectionStore.getThreadAttachmentIds", @@ -6415,7 +6426,7 @@ export const layer: Layer.Layer = Effect.flatMap((shown) => shown.with_fork === null ? Effect.succeed([]) - : runAttachmentIds(source, afterRun(shown.without_fork), shown.with_fork), + : unshownRunAttachmentIds(source, afterRun(shown.without_fork), shown.with_fork), ), ), ); @@ -6430,7 +6441,7 @@ export const layer: Layer.Layer = `; return [ ...new Set([ - ...(yield* runAttachmentIds( + ...(yield* unshownRunAttachmentIds( threadId, options?.includeShownByLiveFork ? -1 : afterRun(own.with_fork), Number.MAX_SAFE_INTEGER, diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index ffeea4623620d..c3e52d780648b 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -821,6 +821,41 @@ it.layer(layerProjectDeletionTest)("project deletion during thread commands", (i assert.isFalse(yield* fileSystem.exists(filePath(dropped))); assert.equal(yield* fileSystem.readFileString(filePath(shared)), "bytes"); + + if (fork !== "no fork") { + const laterRunId = yield* writeRun({ + name: "later", + ordinal: 3, + status: "completed", + attachments: [shared], + }); + const laterForkId = ThreadId.make(`${key}:later-fork`); + yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make(`${key}:later-fork`), + sourceThreadId: threadId, + targetThreadId: laterForkId, + sourcePoint: { type: "run", runId: laterRunId }, + createdBy: "user", + creationSource: "web", + }); + for (const id of [threadId, laterForkId]) { + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make(`${key}:delete:${id}`), + threadId: id, + }); + yield* worker.drain(); + assert.equal(yield* fileSystem.readFileString(filePath(shared)), "bytes"); + } + yield* orchestrator.dispatch({ + type: "thread.delete", + commandId: CommandId.make(`${key}:delete-first-fork`), + threadId: ThreadId.make(`${key}:fork`), + }); + yield* worker.drain(); + assert.isFalse(yield* fileSystem.exists(filePath(shared))); + } }).pipe(Effect.provide(NodeServices.layer)), );