# Files changed · PR \#10580 · pingdotgg/t3code [View on GitCafe](https://git.cafe/pingdotgg/t3code/pull/10580/diffs) Repository: [pingdotgg/t3code](https://git.cafe/pingdotgg/t3code) Visibility: public [refactor\(server\): share attachment message intake](https://git.cafe/pingdotgg/t3code/pull/10580.md) Version: 8 Head commit: 4cfe3581d49e2e0e43ff30bae928ca53c3a4bcaa Base commit: 80736156baa52edd357468cb49a39fd3782d4ad0 Comparison base: 80736156baa52edd357468cb49a39fd3782d4ad0 Showing up to 20 entries from this revision. If the pull request changes, reopen the first page before continuing. ## apps/server/src/orchestration-v2/AttachmentClaims.ts Status: modified; +1; −2 ```diff @@ -28,8 +28,7 @@ return parseThreadSegmentFromAttachmentId(attachment.id) === PENDING_ATTACHMENT_THREAD_SEGMENT; } -/** Best-effort removal of claimed copies after a failed dispatch: the pending +/** Remove partial claims only before dispatch, or after proving they were not accepted. */ - * upload remains the retry source, so only the thread-scoped copies go. */ export const releaseClaimedAttachments = Effect.fn("AttachmentClaims.releaseClaimedAttachments")( function* (claimedPaths: ReadonlyArray) { if (claimedPaths.length === 0) return; ``` ## apps/server/src/orchestration-v2/ThreadLaunchService.test.ts Status: modified; +143; −0 ```diff @@ -1,5 +1,12 @@ +import * as NodeServices from "@effect/platform-node/NodeServices"; +import * as FileSystem from "effect/FileSystem"; +import * as ServerConfig from "../config.ts"; +import { createPendingAttachmentId, resolveAttachmentPath } from "../attachmentStore.ts"; +import * as ThreadMessageIntake from "./ThreadMessageIntake.ts"; import { assert, it, vi } from "@effect/vitest"; import { + ChatAttachmentId, + type ChatAttachment, CommandId, DEFAULT_SERVER_SETTINGS, GitCommandError, @@ -1286,3 +1293,139 @@ assert.equal(launched.projection.messages[0]?.text, "No private workflow state"); }).pipe(Effect.provide(harness.layer)); }); + +it.effect("shared intake preserves durable attachment bytes after a lost launch result", () => { + const harness = makeHarness(); + const files = ServerConfig.layerTest(process.cwd(), { prefix: "t3-message-intake-" }).pipe( + Layer.provideMerge(NodeServices.layer), + ); + return Effect.gen(function* () { + const config = yield* ServerConfig.ServerConfig; + const fs = yield* FileSystem.FileSystem; + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const pendingId = createPendingAttachmentId(); + assert.isNotNull(pendingId); + const attachment: ChatAttachment = { + type: "image", + id: ChatAttachmentId.make(pendingId), + name: "image.png", + mimeType: "image/png", + sizeBytes: 4, + }; + const pendingPath = resolveAttachmentPath({ + attachmentsDir: config.attachmentsDir, + attachment, + }); + assert.isNotNull(pendingPath); + yield* fs.makeDirectory(config.attachmentsDir, { recursive: true }); + yield* fs.writeFile(pendingPath, new Uint8Array([1, 2, 3, 4])); + const input = { + ...launchInput({ command: "intake-launch", thread: "intake-thread" }), + initialMessage: { + messageId: MessageId.make("intake-first"), + text: "First", + attachments: [attachment], + }, + }; + const failed = yield* ThreadMessageIntake.launchThread(input).pipe( + Effect.provideService(ThreadLaunch.ThreadLaunchService, { + launch: (request) => + launches.launch(request).pipe( + Effect.andThen( + new ThreadLaunch.ThreadLaunchError({ + operation: "create-thread", + commandId: request.commandId, + projectId, + cause: "lost result after acceptance", + }), + ), + ), + }), + Effect.flip, + ); + assert.equal(failed._tag, "ThreadLaunchError"); + // The observer failed, but the real V2 message and its bytes were accepted. + const accepted = yield* threads.getThreadProjection(input.threadId); + const stored = accepted.messages.find( + (message) => message.id === input.initialMessage.messageId, + ); + assert.isDefined(stored); + assert.notEqual(stored.attachments[0]?.id, attachment.id); + const storedPath = resolveAttachmentPath({ + attachmentsDir: config.attachmentsDir, + attachment: stored.attachments[0]!, + }); + assert.isNotNull(storedPath); + assert.deepEqual(yield* fs.readFile(storedPath), new Uint8Array([1, 2, 3, 4])); + assert.deepEqual(yield* fs.readFile(pendingPath), new Uint8Array([1, 2, 3, 4])); + + const replayed = yield* ThreadMessageIntake.launchThread(input); + assert.equal(replayed.projection.messages[0]?.id, stored.id); + assert.deepEqual(replayed.projection.messages[0]?.attachments, stored.attachments); + const claimedFiles = Effect.map(fs.readDirectory(config.attachmentsDir), (files) => + files.filter((name) => !name.startsWith("pending-")), + ); + assert.equal((yield* claimedFiles).length, 1); + + const missingProject = yield* ThreadMessageIntake.launchThread({ + ...input, + commandId: CommandId.make("intake-no-project"), + projectId: ProjectId.make("missing-project"), + }).pipe(Effect.flip); + assert.equal(missingProject._tag, "ThreadLaunchError"); + assert.equal((yield* claimedFiles).length, 1); + const missingThread = yield* ThreadMessageIntake.dispatchCommand({ + type: "message.dispatch", + commandId: CommandId.make("intake-no-thread"), + threadId: ThreadId.make("missing-thread"), + messageId: MessageId.make("intake-missing"), + text: "Missing", + attachments: [attachment], + dispatchMode: { type: "start_immediately" }, + createdBy: "user", + creationSource: "web", + }).pipe(Effect.flip); + assert.equal(missingThread._tag, "OrchestratorProjectionError"); + assert.equal((yield* claimedFiles).length, 1); + + // Both ordinary command intake (RPC) and send intake (MCP) use the same store. + const dispatch = ThreadMessageIntake.dispatchCommand({ + type: "message.dispatch", + commandId: CommandId.make("intake-dispatch"), + threadId: input.threadId, + messageId: MessageId.make("intake-second"), + text: "Second", + attachments: [attachment], + dispatchMode: { type: "queue_after_active" }, + createdBy: "user", + creationSource: "web", + }); + yield* dispatch; + yield* dispatch; + const send = ThreadMessageIntake.sendToThread({ + commandId: CommandId.make("intake-send"), + projectId, + threadId: input.threadId, + messageId: MessageId.make("intake-third"), + text: "Third", + attachments: [attachment], + mode: "queue", + createdBy: "agent", + creationSource: "mcp", + }); + yield* send; + yield* send; + assert.equal((yield* claimedFiles).length, 3); + const final = yield* threads.getThreadProjection(input.threadId); + assert.equal(final.messages.length, 3); + for (const message of final.messages) { + const path = resolveAttachmentPath({ + attachmentsDir: config.attachmentsDir, + attachment: message.attachments[0]!, + }); + assert.isNotNull(path); + assert.deepEqual(yield* fs.readFile(path), new Uint8Array([1, 2, 3, 4])); + } + }).pipe(Effect.provide(Layer.mergeAll(harness.layer, files))); +}); ``` ## apps/server/src/orchestration-v2/ThreadMessageIntake.ts Status: added; +134; −0 ```diff @@ -1,0 +1,134 @@ +import type { ChatAttachment, OrchestrationV2Command } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; +import * as Schema from "effect/Schema"; +import { resolveAttachmentPath } from "../attachmentStore.ts"; +import * as ServerConfig from "../config.ts"; +import * as Orchestrator from "./Orchestrator.ts"; + +import * as AttachmentClaims from "./AttachmentClaims.ts"; +import * as ThreadLaunch from "./ThreadLaunchService.ts"; +import * as ThreadManagement from "./ThreadManagementService.ts"; + +// These dispatcher failures occur in receipt validation or planning, before +// commitCommand. Generic dispatch errors can follow a commit and remain uncertain. +function dispatchWasNotAccepted( + error: Orchestrator.OrchestratorV2Error | ThreadManagement.ThreadManagementError, +) { + switch (error._tag) { + case "OrchestratorProjectionError": + case "OrchestratorProviderAdapterError": + case "OrchestratorCommandPreviouslyRejectedError": + case "OrchestratorCommandIdConflictError": + return true; + default: + return false; + } +} +const isOrchestratorError = Schema.is(Orchestrator.OrchestratorV2Error); + +const releaseUnusedClaims = Effect.fn("ThreadMessageIntake.releaseUnusedClaims")(function* ( + claimed: AttachmentClaims.ClaimedAttachments, + accepted: ReadonlyArray, +) { + if (claimed.claimedPaths.length === 0) return; + const config = yield* ServerConfig.ServerConfig; + const retained = new Set( + accepted.map((attachment) => + resolveAttachmentPath({ + attachmentsDir: config.attachmentsDir, + attachment, + }), + ), + ); + yield* AttachmentClaims.releaseClaimedAttachments( + claimed.claimedPaths.filter((path) => !retained.has(path)), + ); +}); + +export const dispatchCommand = Effect.fn("ThreadMessageIntake.dispatchCommand")(function* ( + command: OrchestrationV2Command, +) { + const threads = yield* ThreadManagement.ThreadManagementService; + if (command.type !== "message.dispatch") return yield* threads.dispatch(command); + const claimed = yield* AttachmentClaims.claimPendingAttachments({ + threadId: command.threadId, + attachments: command.attachments, + }); + return yield* threads.dispatch({ ...command, attachments: claimed.attachments }).pipe( + Effect.tap((result) => + releaseUnusedClaims( + claimed, + result.storedEvents.flatMap(({ event }) => + event.type === "message.updated" ? event.payload.attachments : [], + ), + ), + ), + Effect.tapError((error) => + dispatchWasNotAccepted(error) + ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths) + : Effect.void, + ), + ); +}); + +export const sendToThread = Effect.fn("ThreadMessageIntake.sendToThread")(function* ( + input: ThreadManagement.ThreadManagementSendInput, +) { + const threads = yield* ThreadManagement.ThreadManagementService; + const claimed = yield* AttachmentClaims.claimPendingAttachments(input); + return yield* threads.sendToThread({ ...input, attachments: claimed.attachments }).pipe( + Effect.tap((result) => releaseUnusedClaims(claimed, result.message.attachments)), + Effect.tapError((error) => + dispatchWasNotAccepted(error) + ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths) + : Effect.void, + ), + ); +}); + +export const launchThread = Effect.fn("ThreadMessageIntake.launchThread")(function* ( + input: ThreadLaunch.ThreadLaunchInput, +) { + const launches = yield* ThreadLaunch.ThreadLaunchService; + if (!input.initialMessage?.attachments.some(AttachmentClaims.attachmentIsPendingUpload)) { + return yield* launches.launch(input); + } + if (input.threadId === undefined) { + return yield* new AttachmentClaims.AttachmentClaimError({ + message: "Uploaded attachments need a thread id at launch.", + }); + } + const claimed = yield* AttachmentClaims.claimPendingAttachments({ + threadId: input.threadId, + attachments: input.initialMessage.attachments, + }); + return yield* launches + .launch({ + ...input, + initialMessage: { ...input.initialMessage, attachments: claimed.attachments }, + }) + .pipe( + Effect.tap((result) => + releaseUnusedClaims( + claimed, + result.projection.messages.flatMap((message) => message.attachments), + ), + ), + Effect.tapError((error) => { + // Project/receipt reads precede message dispatch. The create-thread error + // also wraps post-message projection reads, so its tag alone is not proof. + const notAccepted = + error.operation === "resolve-project" || + error.operation === "read-receipt" || + ((error.operation === "create-thread" || error.operation === "dispatch-message") && + isOrchestratorError(error.cause) && + // Projection errors under create-thread can occur after the message commit. + (error.operation !== "create-thread" || + error.cause._tag !== "OrchestratorProjectionError") && + dispatchWasNotAccepted(error.cause)); + return notAccepted + ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths) + : Effect.void; + }), + ); +}); ``` ## apps/server/src/ws.ts Status: modified; +101; −148 ```diff @@ -91,7 +91,8 @@ import * as ExternalLauncher from "./process/externalLauncher.ts"; import * as ThreadManagementService from "./orchestration-v2/ThreadManagementService.ts"; import { ProviderSessionManagerV2 } from "./orchestration-v2/ProviderSessionManager.ts"; -import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts"; +import type { ThreadLaunchService } from "./orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadMessageIntake from "./orchestration-v2/ThreadMessageIntake.ts"; import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts"; import { archivedShellStreamItemFromThreadShell, @@ -121,11 +122,6 @@ import * as ProjectionSnapshotQuery from "./orchestration/Services/ProjectionSnapshotQuery.ts"; import * as OrchestrationEventStore from "./persistence/Services/OrchestrationEventStore.ts"; import { userFacingDispatchErrorMessage } from "./orchestration-v2/UserFacingErrors.ts"; -import { - attachmentIsPendingUpload, - claimPendingAttachments, - releaseClaimedAttachments, -} from "./orchestration-v2/AttachmentClaims.ts"; import { observeRpcEffect as instrumentRpcEffect, observeRpcStream as instrumentRpcStream, @@ -527,6 +523,12 @@ const currentSessionId = currentSession.sessionId; const sql = yield* SqlClient.SqlClient; const threadManagement = yield* ThreadManagementService.ThreadManagementService; + const intakeContext = yield* Effect.context< + | ThreadManagementService.ThreadManagementService + | ThreadLaunchService + | FileSystem.FileSystem + | ServerConfig.ServerConfig + >(); const applicationEvents = yield* OrchestrationEventStore.OrchestrationEventStore; const projectionSnapshotQuery = yield* ProjectionSnapshotQuery.ProjectionSnapshotQuery; const canReplayPersistedRange = Effect.fnUntraced(function* ( @@ -599,7 +601,6 @@ })), ), ); - const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService; const scheduledTasks = yield* ScheduledTasks.ScheduledTaskService; const pullRequests = yield* PullRequestService.PullRequestService; const usage = yield* UsageService.UsageService; @@ -1313,52 +1314,29 @@ [ORCHESTRATION_V2_WS_METHODS.dispatchCommand]: (command) => observeRpcEffect( ORCHESTRATION_V2_WS_METHODS.dispatchCommand, + startup + .enqueueCommand( + ThreadMessageIntake.dispatchCommand( + ThreadManagementService.withCreationProvenance(command, { + createdBy: "user", + creationSource: "creationSource" in command ? command.creationSource : "web", + }), + ).pipe(Effect.provide(intakeContext)), + ) + .pipe( + Effect.tap(() => recordClientCommandAnalytics(command)), + Effect.map((result) => ({ sequence: result.sequence })), + Effect.mapError((cause) => { + const detail = userFacingDispatchErrorMessage(cause); + return new OrchestrationV2DispatchCommandError({ + commandId: command.commandId, + commandType: command.type, - Effect.gen(function* () { - // Pending uploads are claimed into the thread's attachment store - // at intake; a failed dispatch releases the claimed copies while - // the pending upload stays behind as the client's retry source. - const claimed = - command.type === "message.dispatch" && - command.attachments.some(attachmentIsPendingUpload) - ? yield* claimPendingAttachments({ - threadId: command.threadId, - attachments: command.attachments, + message: detail ?? "Failed to dispatch orchestration V2 command", + ...(detail === undefined ? {} : { detail }), + cause, + }); + }), + ), - }) - : null; - const effectiveCommand = - claimed === null || command.type !== "message.dispatch" - ? command - : { ...command, attachments: claimed.attachments }; - return yield* startup - .enqueueCommand( - threadManagement.dispatch( - ThreadManagementService.withCreationProvenance(effectiveCommand, { - createdBy: "user", - creationSource: "creationSource" in command ? command.creationSource : "web", - }), - ), - ) - .pipe( - Effect.tapError(() => - claimed === null - ? Effect.void - : releaseClaimedAttachments(claimed.claimedPaths), - ), - ); - }).pipe( - Effect.tap(() => recordClientCommandAnalytics(command)), - Effect.map((result) => ({ sequence: result.sequence })), - Effect.mapError((cause) => { - const detail = userFacingDispatchErrorMessage(cause); - return new OrchestrationV2DispatchCommandError({ - commandId: command.commandId, - commandType: command.type, - message: detail ?? "Failed to dispatch orchestration V2 command", - ...(detail === undefined ? {} : { detail }), - cause, - }); - }), - ), { "rpc.aggregate": "orchestrationV2", "orchestration_v2.command_id": command.commandId, @@ -1454,104 +1432,79 @@ [ORCHESTRATION_V2_WS_METHODS.launchThread]: (input) => observeRpcEffect( ORCHESTRATION_V2_WS_METHODS.launchThread, + startup + .enqueueCommand( + ThreadMessageIntake.launchThread({ - Effect.gen(function* () { - const pendingUploads = - input.initialMessage?.attachments.some(attachmentIsPendingUpload) ?? false; - if (pendingUploads && input.threadId === undefined) { - return yield* new OrchestrationV2ThreadLaunchError({ commandId: input.commandId, + ...(input.threadId === undefined ? {} : { threadId: input.threadId }), + ...(input.reuseExistingThread === undefined + ? {} + : { reuseExistingThread: input.reuseExistingThread }), projectId: input.projectId, + title: input.title, + ...(input.generateTitle === undefined + ? {} + : { generateTitle: input.generateTitle }), + modelSelection: input.modelSelection, + runtimeMode: input.runtimeMode, + interactionMode: input.interactionMode, + workspaceStrategy: input.workspaceStrategy, + ...(input.initialMessage === undefined + ? {} + : { + initialMessage: { + ...(input.initialMessage.messageId === undefined + ? {} + : { messageId: input.initialMessage.messageId }), + text: input.initialMessage.text, + attachments: input.initialMessage.attachments, + }, + }), + createdBy: "user", + creationSource: input.creationSource ?? "web", + }).pipe(Effect.provide(intakeContext)), + ) + .pipe( + Effect.tap(() => + analytics + .record("client.thread.started", originProps) + .pipe( + Effect.andThen( - message: "Uploaded attachments need a thread id at launch.", - }); - } - const claimed = - pendingUploads && input.threadId !== undefined && input.initialMessage !== undefined - ? yield* claimPendingAttachments({ - threadId: input.threadId, - attachments: input.initialMessage.attachments, - }).pipe( - Effect.mapError( - (cause) => - new OrchestrationV2ThreadLaunchError({ - commandId: input.commandId, - projectId: input.projectId, - message: cause.message, - cause, - }), - ), - ) - : null; - const initialMessage = - input.initialMessage === undefined - ? undefined - : claimed === null - ? input.initialMessage - : { ...input.initialMessage, attachments: claimed.attachments }; - return yield* startup - .enqueueCommand( - threadLaunch.launch({ - commandId: input.commandId, - ...(input.threadId === undefined ? {} : { threadId: input.threadId }), - ...(input.reuseExistingThread === undefined - ? {} - : { reuseExistingThread: input.reuseExistingThread }), - projectId: input.projectId, - title: input.title, - ...(input.generateTitle === undefined - ? {} - : { generateTitle: input.generateTitle }), - modelSelection: input.modelSelection, - runtimeMode: input.runtimeMode, - interactionMode: input.interactionMode, - workspaceStrategy: input.workspaceStrategy, - ...(initialMessage === undefined - ? {} - : { - initialMessage: { - ...(initialMessage.messageId === undefined - ? {} - : { messageId: initialMessage.messageId }), - text: initialMessage.text, - attachments: initialMessage.attachments, - }, - }), - createdBy: "user", - creationSource: input.creationSource ?? "web", - }), - ) - .pipe( - Effect.tapError(() => - claimed === null - ? Effect.void - : releaseClaimedAttachments(claimed.claimedPaths), - ), - Effect.tap(() => - analytics - .record("client.thread.started", originProps) - .pipe( - Effect.andThen( - input.initialMessage === undefined - ? Effect.void - : analytics.record("client.turn.requested", originProps), - ), - Effect.ignore, + input.initialMessage === undefined + ? Effect.void + : analytics.record("client.turn.requested", originProps), ), - ), - Effect.map((result) => ({ - ...result, - projection: projectThreadProjectionForWire(result.projection), - })), - Effect.mapError( - (cause) => - new OrchestrationV2ThreadLaunchError({ - commandId: input.commandId, - projectId: input.projectId, - message: "Failed to launch thread", - cause, - }), - ), - ); - }), + Effect.ignore, + ), + ), + Effect.map((result) => ({ + ...result, + projection: projectThreadProjectionForWire(result.projection), + })), + Effect.catchTags({ + AttachmentClaimError: (cause) => + new OrchestrationV2ThreadLaunchError({ + commandId: input.commandId, + projectId: input.projectId, + message: cause.message, + cause, + }), + ThreadLaunchError: (cause) => + new OrchestrationV2ThreadLaunchError({ + commandId: input.commandId, + projectId: input.projectId, + message: "Failed to launch thread", + cause, + }), + ServerRuntimeStartupError: (cause) => + new OrchestrationV2ThreadLaunchError({ + commandId: input.commandId, + projectId: input.projectId, + message: "Failed to launch thread", + cause, + }), + }), + ), { "rpc.aggregate": "orchestration", "orchestration_v2.command_id": input.commandId, ```