diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index eb1209158cc74..0a250bb2b865c 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -117,6 +117,7 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ Schema.Struct({ type: Schema.Literal("attachment.cleanup"), attachmentIds: Schema.Array(Schema.String), + skipReferenced: Schema.optional(Schema.Boolean), }), Schema.Struct({ type: Schema.Literal("thread-title.generate"), diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index c753e07d4689a..fd598e95ec11a 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -31,6 +31,7 @@ import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; import * as CheckpointService from "./CheckpointService.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; import * as EffectWorker from "./EffectWorker.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; import * as RunFinalizationService from "./RunFinalizationService.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderTurnControlService from "./ProviderTurnControlService.ts"; @@ -168,6 +169,7 @@ function layerExecutorFor(input: { Layer.mergeAll( layerDependencies, Layer.mock(ThreadManagementService.ThreadManagementService)(input.threads ?? {}), + Layer.mock(ProjectionStore.ProjectionStoreV2)({}), ServerSettings.layerTest( input.continueAfterRestart === true ? { continueThreadsAfterServerUpdate: true } : {}, ), diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index bbb214542540d..19a387eb8a954 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -22,6 +22,7 @@ import { import * as RunFinalizationService from "./RunFinalizationService.ts"; import * as ResourceCleanupService from "./ResourceCleanupService.ts"; import * as EffectOutbox from "./EffectOutbox.ts"; +import * as ProjectionStore from "./ProjectionStore.ts"; import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; import * as ProviderTurnControlService from "./ProviderTurnControlService.ts"; @@ -84,6 +85,7 @@ export const layerExecutor: Layer.Layer< | RuntimeRequestService.RuntimeRequestServiceV2 | ThreadTitleRegenerationService.ThreadTitleRegenerationService | ThreadManagementService.ThreadManagementService + | ProjectionStore.ProjectionStoreV2 | ServerSettings.ServerSettingsService > = Layer.effect( OrchestrationEffectExecutorV2, @@ -98,6 +100,7 @@ export const layerExecutor: Layer.Layer< const threadTitleRegeneration = yield* ThreadTitleRegenerationService.ThreadTitleRegenerationService; const threads = yield* ThreadManagementService.ThreadManagementService; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; const settings = yield* ServerSettings.ServerSettingsService; return OrchestrationEffectExecutorV2.of({ execute: Effect.fn("OrchestrationEffectExecutorV2.execute")(function (effect, options) { @@ -584,8 +587,21 @@ export const layerExecutor: Layer.Layer< }), ), ); - case "attachment.cleanup": - return resourceCleanup.cleanupAttachments(effect.request.attachmentIds).pipe( + case "attachment.cleanup": { + const request = effect.request; + return Effect.gen(function* () { + // Another message or answer may share the file, or a stale queued edit may restore it. + const referenced = new Set( + request.skipReferenced + ? yield* projectionStore.getThreadAttachmentIds(effect.threadId, { + includeShownByLiveFork: true, + }) + : [], + ); + yield* resourceCleanup.cleanupAttachments( + request.attachmentIds.filter((id) => !referenced.has(id)), + ); + }).pipe( Effect.mapError( (cause) => new OrchestrationEffectExecutionError({ @@ -595,6 +611,7 @@ export const layerExecutor: Layer.Layer< }), ), ); + } case "thread-title.generate": return threadTitleRegeneration .execute({ diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 6049beffadcfe..6d78d1008e991 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -11529,6 +11529,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const dispatchQueuedRunEdit = Effect.fn("Orchestrator.dispatchQueuedRunEdit")(function* ( command: Extract, events: Ref.Ref>, + effects: Ref.Ref>, ) { if (command.text.trim().length === 0) { return yield* new OrchestratorDispatchError({ @@ -11622,6 +11623,28 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }), }); } + // Thread deletion only cleans files the thread still references. + const keptAttachmentIds = new Set( + (command.attachments ?? queuedMessage.attachments).map((attachment) => attachment.id), + ); + const droppedAttachmentIds = queuedMessage.attachments + .map((attachment) => attachment.id) + .filter((id) => !keptAttachmentIds.has(id)); + if (droppedAttachmentIds.length > 0) { + yield* Ref.update(effects, (existing) => [ + ...existing, + { + id: `effect:${command.commandId}:attachment.cleanup`, + commandId: command.commandId, + threadId: command.threadId, + request: { + type: "attachment.cleanup", + attachmentIds: droppedAttachmentIds, + skipReferenced: true, + }, + } satisfies PendingOrchestrationEffectV2, + ]); + } }); const subagentCount = (threadId: ThreadId) => @@ -15113,7 +15136,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* dispatchQueuedRunCancel(command, events); break; case "queued-run.edit": - yield* dispatchQueuedRunEdit(command, events); + yield* dispatchQueuedRunEdit(command, events, effects); break; case "checkpoint.rollback": yield* dispatchCheckpointRollback(command, events, effects); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 32007af08573f..5ee086f569c41 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -501,8 +501,10 @@ export interface ProjectionStoreV2Shape { readonly searchThread: ( input: OrchestrationV2SearchThreadInput, ) => Effect.Effect; + /** Files thread deletion may remove; `includeShownByLiveFork` adds ones a live fork shows. */ readonly getThreadAttachmentIds: ( threadId: ThreadId, + options?: { readonly includeShownByLiveFork?: boolean }, ) => Effect.Effect, ProjectionStoreV2Error>; readonly getThreadHistoryPage: ( threadId: ThreadId, @@ -6252,14 +6254,27 @@ export const layer: Layer.Layer = ) `; - const getThreadAttachmentIds: ProjectionStoreV2Shape["getThreadAttachmentIds"] = (threadId) => + // Thread deletion keeps what a live fork still shows; a reference check must count it. + const exceptShownByLiveFork = ( + threadId: ThreadId, + runIdColumn: string, + options: { readonly includeShownByLiveFork?: boolean } | undefined, + ) => + options?.includeShownByLiveFork === true + ? sql`` + : sql`AND NOT ${shownByLiveFork(threadId, runIdColumn)}`; + + const getThreadAttachmentIds: ProjectionStoreV2Shape["getThreadAttachmentIds"] = ( + threadId, + options, + ) => 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")} + ${exceptShownByLiveFork(threadId, "message.run_id", options)} `, sql<{ payload_json: string }>` SELECT item.payload_json @@ -6269,7 +6284,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")} + ${exceptShownByLiveFork(threadId, "item.run_id", options)} `, sql<{ id: string }>` SELECT DISTINCT json_extract(attachment.value, '$.id') AS id @@ -6278,7 +6293,7 @@ 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")} + ${exceptShownByLiveFork(threadId, "item.run_id", options)} `, // A guest's waiting files are pending uploads that no message row holds yet. sql<{ id: string }>` diff --git a/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts b/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts index a5e852d0e133f..3ca71f384eba1 100644 --- a/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts +++ b/apps/server/src/orchestration-v2/RunFinalizationQueue.test.ts @@ -97,6 +97,7 @@ it.effect.each([ Layer.mock(ThreadManagement.ThreadManagementService)({ recoverDelegatedTask: () => Effect.void, }), + Layer.mock(ProjectionStore.ProjectionStoreV2)({}), ServerSettings.layerTest(), Layer.mock(ProviderTurnStart.ProviderTurnStartServiceV2)({ start: () => Effect.sync(() => calls.push("start")).pipe(Effect.asVoid), diff --git a/apps/server/src/orchestration-v2/ThreadMessageIntake.test.ts b/apps/server/src/orchestration-v2/ThreadMessageIntake.test.ts index bec9c60c7a014..a92cdd22aa515 100644 --- a/apps/server/src/orchestration-v2/ThreadMessageIntake.test.ts +++ b/apps/server/src/orchestration-v2/ThreadMessageIntake.test.ts @@ -10,6 +10,7 @@ import { EventId, GuestMessageId, MessageId, + RunId, RuntimeRequestId, ThreadId, TurnItemId, @@ -426,3 +427,51 @@ it.effect("sends a guest message with the guest's files, never ones the client n expect(command.attachments).toEqual([guestFile]); }).pipe(Effect.provide(layerIntakeTest)), ); + +it.effect("refuses a queued edit that keeps a file another edit already deleted", () => + Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const storedFile = (name: string) => ({ + type: "image" as const, + id: ChatAttachmentId.make(createAttachmentId("thread-queued-edit")!), + name, + mimeType: "image/png", + sizeBytes: 3, + }); + const kept = storedFile("kept.png"); + const deleted = storedFile("deleted.png"); + NodeFS.writeFileSync( + NodePath.join(config.attachmentsDir, `${kept.id}.png`), + new Uint8Array([1, 2, 3]), + ); + const captured: OrchestrationV2ServerCommand[] = []; + const edit = (attachments: ReadonlyArray) => + dispatchCommand({ + type: "queued-run.edit", + commandId: CommandId.make(`queued-edit-${attachments.length}`), + threadId: ThreadId.make("thread-queued-edit"), + runId: RunId.make("run-queued-edit"), + text: "Edited from a stale view.", + attachments, + }).pipe( + Effect.provide( + Layer.mock(ThreadManagementService.ThreadManagementService)({ + dispatch: layerFailingDispatchHandler(captured), + hasReceipt: (commandId) => + Effect.succeed(captured.some((command) => command.commandId === commandId)), + }), + ), + Effect.result, + ); + + yield* edit([kept]); + expect(captured).toHaveLength(1); + const result = yield* edit([kept, deleted]); + expect(result._tag).toBe("Failure"); + expect(captured).toHaveLength(1); + // A retry of the accepted edit replays its receipt even after a later cleanup removed its file. + NodeFS.rmSync(NodePath.join(config.attachmentsDir, `${kept.id}.png`)); + yield* edit([kept]); + expect(captured).toHaveLength(2); + }).pipe(Effect.provide(layerIntakeTest)), +); diff --git a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts index c6a6cdbf7ac1e..6e9fd503baec5 100644 --- a/apps/server/src/orchestration-v2/ThreadMessageIntake.ts +++ b/apps/server/src/orchestration-v2/ThreadMessageIntake.ts @@ -2,6 +2,7 @@ import { remapComposerContextAttachments } from "@t3tools/shared/composerContext import { appendUserInputAttachmentPaths } from "../provider/userInputAttachments.ts"; import type { ChatAttachment, OrchestrationV2Command } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; import * as Schema from "effect/Schema"; import { resolveAttachmentPath } from "../attachmentStore.ts"; import * as ServerConfig from "../config.ts"; @@ -159,6 +160,24 @@ export const dispatchCommand = Effect.fn("ThreadMessageIntake.dispatchCommand")( (command.type !== "queued-run.edit" || command.attachments === undefined) ) return yield* threads.dispatch(command); + // An accepted edit's retry replays its receipt even if a later cleanup removed its files. + if (command.type === "queued-run.edit" && !(yield* threads.hasReceipt(command.commandId))) { + const config = yield* ServerConfig.ServerConfig; + const fileSystem = yield* FileSystem.FileSystem; + // A stale view can keep a file that another client's edit dropped and deleted. + for (const attachment of command.attachments ?? []) { + if (AttachmentClaims.attachmentIsPendingUpload(attachment)) continue; + const path = resolveAttachmentPath({ attachmentsDir: config.attachmentsDir, attachment }); + if ( + path !== null && + !(yield* fileSystem.exists(path).pipe(Effect.orElseSucceed(() => false))) + ) { + return yield* new AttachmentClaims.AttachmentClaimError({ + message: `Attachment '${attachment.name}' cannot be sent: attachment not found.`, + }); + } + } + } const claimed = yield* AttachmentClaims.claimPendingAttachments({ threadId: command.threadId, attachments: command.attachments ?? [], diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 9e054e3a0bf89..38aed90798a3f 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -491,7 +491,8 @@ const layerProjectDeletionTest = Layer.mergeAll( Layer.provide(McpProviderSessions.layer), Layer.provide(SqlitePersistence.layerMemory), Layer.provide(layerCheckpointStoreTest), - Layer.provide(layerServerConfig), + // Exposed so tests can reach the attachments directory that resource cleanup removes from. + Layer.provideMerge(layerServerConfig), Layer.provide(ServerSettings.layerTest()), Layer.provide(layerTestProviderInstanceRegistry), Layer.provide(layerGitWorkflowTest), @@ -680,6 +681,147 @@ it.layer(layerProjectDeletionTest)("project deletion during thread commands", (i }), ); + it.effect.each(["no fork", "a live fork"] as const)( + "a queued edit's cleanup keeps a file earlier history still shows, with %s", + (fork) => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const worker = yield* EffectWorker.OrchestrationEffectWorkerV2; + const config = yield* ServerConfig.ServerConfig; + const fileSystem = yield* FileSystem.FileSystem; + const path = yield* Path.Path; + const key = `queued-edit-shared-file-${fork.replaceAll(" ", "-")}`; + const threadId = ThreadId.make(key); + const storedFile = (name: string) => + ({ + type: "image", + id: createAttachmentId(threadId)!, + name: `${name}.png`, + mimeType: "image/png", + sizeBytes: 5, + }) as const; + const shared = storedFile("shared"); + const dropped = storedFile("dropped"); + const filePath = (file: typeof shared) => + path.join(config.attachmentsDir, `${file.id}.png`); + yield* fileSystem.writeFileString(filePath(shared), "bytes"); + yield* fileSystem.writeFileString(filePath(dropped), "bytes"); + yield* orchestrator.dispatch({ + type: "thread.create", + commandId: CommandId.make(`${key}:create`), + createdBy: "user", + creationSource: "web", + threadId, + projectId: ProjectId.make("queued-edit-shared-file-project"), + title: "Shared file", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + }); + const writeRun = Effect.fn("test.writeQueuedEditRun")(function* (input: { + readonly name: string; + readonly ordinal: number; + readonly status: "completed" | "queued"; + readonly attachments: ReadonlyArray; + }) { + const now = yield* DateTime.now; + const runId = RunId.make(`${key}:${input.name}`); + const messageId = MessageId.make(`${key}:${input.name}`); + const completed = input.status === "completed"; + yield* eventSink.write({ + events: [ + { + id: EventId.make(`${key}:${input.name}:run`), + type: "run.created", + threadId, + runId, + occurredAt: now, + payload: { + id: runId, + threadId, + ordinal: input.ordinal, + providerInstanceId: modelSelection.instanceId, + modelSelection, + providerThreadId: null, + userMessageId: messageId, + rootNodeId: null, + activeAttemptId: null, + status: input.status, + requestedAt: now, + startedAt: completed ? now : null, + completedAt: completed ? now : null, + checkpointId: null, + contextHandoffId: null, + }, + }, + { + id: EventId.make(`${key}:${input.name}:message`), + type: "message.updated", + threadId, + runId, + occurredAt: now, + payload: { + id: messageId, + threadId, + runId, + nodeId: null, + role: "user", + text: `Send ${input.name} work`, + attachments: input.attachments, + streaming: false, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + }, + }, + ], + }); + return runId; + }); + + const sourceRunId = yield* writeRun({ + name: "source", + ordinal: 1, + status: "completed", + attachments: [shared], + }); + if (fork !== "no fork") { + const forkThreadId = ThreadId.make(`${key}:fork`); + yield* orchestrator.dispatch({ + type: "thread.fork", + commandId: CommandId.make(`${key}:fork`), + sourceThreadId: threadId, + targetThreadId: forkThreadId, + sourcePoint: { type: "run", runId: sourceRunId }, + createdBy: "user", + creationSource: "web", + }); + } + const queuedRunId = yield* writeRun({ + name: "queued", + ordinal: 2, + status: "queued", + attachments: [shared, dropped], + }); + yield* orchestrator.dispatch({ + type: "queued-run.edit", + commandId: CommandId.make(`${key}:edit`), + threadId, + runId: queuedRunId, + text: "Send queued work without files", + attachments: [], + }); + yield* worker.drain(); + + assert.isFalse(yield* fileSystem.exists(filePath(dropped))); + assert.equal(yield* fileSystem.readFileString(filePath(shared)), "bytes"); + }).pipe(Effect.provide(NodeServices.layer)), + ); + it.effect("waits for an in-flight thread update before planning deletion", () => Effect.gen(function* () { const projects = yield* ProjectService.ProjectService; @@ -9352,7 +9494,23 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { it.effect("edits and removes queued runs", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; + const outbox = yield* EffectOutbox.EffectOutboxV2; const threadId = ThreadId.make("runtime-layer-queued-edit-thread"); + const attachment = (name: string) => + ({ + type: "image", + id: `runtime-layer-queued-edit-${name}`, + name: `${name}.png`, + mimeType: "image/png", + sizeBytes: 128, + }) as const; + const kept = attachment("kept"); + const shared = attachment("shared"); + const dropped = attachment("dropped"); + const effectRequests = (commandId: string) => + Effect.map(outbox.listByCommandId(CommandId.make(commandId)), (effects) => + effects.map((effect) => effect.request), + ); yield* orchestrator.dispatch({ type: "thread.create", @@ -9376,7 +9534,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { threadId, messageId: MessageId.make("runtime-layer-queued-edit-active-message"), text: "Keep the provider occupied.", - attachments: [], + attachments: [shared], modelSelection, dispatchMode: { type: "start_immediately" }, }); @@ -9421,22 +9579,33 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { threadId, runId: queuedRun.id, text: "Updated queued text with an attachment.", - attachments: [ - { - type: "image", - id: "runtime-layer-queued-edit-attachment", - name: "screenshot.png", - mimeType: "image/png", - sizeBytes: 128, - }, - ], + attachments: [kept, shared, dropped], + }); + yield* orchestrator.dispatch({ + type: "queued-run.edit", + commandId: CommandId.make("runtime-layer-queued-edit-drop-attachments"), + threadId, + runId: queuedRun.id, + text: "Updated queued text with an attachment.", + attachments: [kept], }); const afterAttachmentEdit = yield* orchestrator.getThreadProjection(threadId); assert.deepEqual( afterAttachmentEdit.messages .find((message) => message.id === queuedRun.userMessageId) ?.attachments.map((attachment) => attachment.id), - ["runtime-layer-queued-edit-attachment"], + [kept.id], + ); + assert.deepEqual( + yield* effectRequests("runtime-layer-queued-edit-drop-attachments"), + [ + { + type: "attachment.cleanup", + attachmentIds: [shared.id, dropped.id], + skipReferenced: true, + }, + ], + "dropped files are queued for a cleanup that skips files the thread still references", ); yield* orchestrator.dispatch({ @@ -9451,9 +9620,10 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { afterTextOnlyEdit.messages .find((message) => message.id === queuedRun.userMessageId) ?.attachments.map((attachment) => attachment.id), - ["runtime-layer-queued-edit-attachment"], + [kept.id], "an edit without attachments must leave the stored attachments untouched", ); + assert.deepEqual(yield* effectRequests("runtime-layer-queued-edit-text-only"), []); const emptyEditError = yield* orchestrator .dispatch({ diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 6866633397168..4f834535cc2d2 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -343,6 +343,7 @@ const layerEffectExecutorProvided = EffectWorker.layerExecutor.pipe( layerRuntimeRequestServiceProvided, layerThreadTitleRegenerationProvided, layerThreadManagementProvided, + ProjectionStore.layer, ), ), ); diff --git a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts index 9f035b7e039b7..0ba9137e45dbc 100644 --- a/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts +++ b/apps/server/src/orchestration-v2/testkit/ProviderReplayHarness.ts @@ -552,6 +552,7 @@ export function layerWithRegistry( layerThreadTitleRegenerationTest, layerServerSettings, layerThreadManagementProvided, + layerStores, ), ), );