1import { remapComposerContextAttachments } from "@t3tools/shared/composerContextReferences";
2import { appendUserInputAttachmentPaths } from "../provider/userInputAttachments.ts";
3import type { ChatAttachment, OrchestrationV2Command } from "@t3tools/contracts";
4import * as Effect from "effect/Effect";
5import * as Schema from "effect/Schema";
6import { resolveAttachmentPath } from "../attachmentStore.ts";
7import * as ServerConfig from "../config.ts";
8import * as Orchestrator from "./Orchestrator.ts";
9
10import * as AttachmentClaims from "./AttachmentClaims.ts";
11import * as ThreadLaunch from "./ThreadLaunchService.ts";
12import * as ThreadManagement from "./ThreadManagementService.ts";
13
14// These failures precede commitCommand; generic dispatch errors can follow commits and remain ambiguous.
15function dispatchWasNotAccepted(
16 error: Orchestrator.OrchestratorV2Error | ThreadManagement.ThreadManagementError,
17) {
18 switch (error._tag) {
19 case "OrchestratorCommandRejectedError":
20 case "OrchestratorThreadAlreadyExistsError":
21 case "OrchestratorProjectionError":
22 case "OrchestratorProviderAdapterError":
23 case "OrchestratorCommandPreviouslyRejectedError":
24 case "OrchestratorCommandIdConflictError":
25 case "OrchestratorSubagentThreadReadOnlyError":
26 case "OrchestratorThreadAboveModeLimitError":
27 case "OrchestratorProviderSelectionBusyError":
28 case "OrchestratorRuntimeModeAboveCeilingError":
29 case "ThreadManagementProviderInstanceDisabledError":
30 return true;
31 default:
32 return false;
33 }
34}
35const isOrchestratorError = Schema.is(Orchestrator.OrchestratorV2Error);
36const isThreadManagementError = Schema.is(ThreadManagement.ThreadManagementError);
37
38const releaseUnusedClaims = Effect.fn("ThreadMessageIntake.releaseUnusedClaims")(function* (
39 claimedPaths: ReadonlyArray<string>,
40 accepted: ReadonlyArray<ChatAttachment>,
41) {
42 if (claimedPaths.length === 0) return;
43 const config = yield* ServerConfig.ServerConfig;
44 const retained = new Set(
45 accepted.flatMap((attachment) => {
46 const path = resolveAttachmentPath({
47 attachmentsDir: config.attachmentsDir,
48 attachment,
49 });
50 return path ? [path, `${path}.txt`] : [];
51 }),
52 );
53 yield* AttachmentClaims.releaseClaimedAttachments(
54 claimedPaths.filter((path) => !retained.has(path)),
55 );
56});
57
58/**
59 * A guest message goes out with the files the guest attached, never ones a client names. They
60 * are already the thread's attachments. A message no longer waiting gets none; the orchestrator
61 * then refuses or replays the send.
62 */
63const withGuestAttachments = Effect.fn("ThreadMessageIntake.withGuestAttachments")(function* (
64 command: Extract<OrchestrationV2Command, { readonly type: "message.dispatch" }>,
65) {
66 if (command.guestMessageId === undefined) return command;
67 const threads = yield* ThreadManagement.ThreadManagementService;
68 // Only the thread record is needed; the window keeps the read small.
69 const { projection } = yield* threads
70 .getThreadSnapshotWindow(command.threadId, { rowLimit: 1 })
71 .pipe(
72 Effect.mapError(
73 (cause) =>
74 new AttachmentClaims.AttachmentClaimError({
75 message: "Failed to read the guest message.",
76 cause,
77 }),
78 ),
79 );
80 const message = projection.thread.guestShare?.messages.find(
81 (candidate) => candidate.id === command.guestMessageId && candidate.status === "pending",
82 );
83 return { ...command, attachments: message?.attachments ?? [] };
84});
85
86export const dispatchCommand = Effect.fn("ThreadMessageIntake.dispatchCommand")(function* (
87 incoming: OrchestrationV2Command,
88) {
89 const command =
90 incoming.type === "message.dispatch" ? yield* withGuestAttachments(incoming) : incoming;
91 const threads = yield* ThreadManagement.ThreadManagementService;
92 if (command.type === "runtime-request.respond" && command.attachmentsByQuestionId) {
93 const config = yield* ServerConfig.ServerConfig;
94 const incomingByQuestionId = command.attachmentsByQuestionId;
95 yield* AttachmentClaims.validateAttachmentLimits(Object.values(incomingByQuestionId).flat());
96 const claimedPaths: string[] = [];
97 const prepared = yield* Effect.gen(function* () {
98 const attachmentsByQuestionId: import("@t3tools/contracts").UserInputAttachments = {};
99 for (const [questionId, attachments] of Object.entries(incomingByQuestionId)) {
100 const claimed = yield* AttachmentClaims.claimPendingAttachments({
101 threadId: command.threadId,
102 attachments,
103 });
104 claimedPaths.push(...claimed.claimedPaths);
105 Object.defineProperty(attachmentsByQuestionId, questionId, {
106 value: claimed.attachments,
107 enumerable: true,
108 });
109 }
110 const answers = yield* appendUserInputAttachmentPaths({
111 answers: command.answers ?? {},
112 attachmentsByQuestionId,
113 attachmentsDir: config.attachmentsDir,
114 }).pipe(
115 Effect.mapError(
116 (cause) => new AttachmentClaims.AttachmentClaimError({ message: cause.issue }),
117 ),
118 );
119 return { answers, attachmentsByQuestionId };
120 }).pipe(Effect.onError(() => AttachmentClaims.releaseClaimedAttachments(claimedPaths)));
121 return yield* threads
122 .dispatch({
123 ...command,
124 answers: prepared.answers,
125 attachmentsByQuestionId: prepared.attachmentsByQuestionId,
126 })
127 .pipe(
128 Effect.tap((result) => {
129 // Replayed receipts refer to the first answer; release this attempt's unused copies only with a resolved item.
130 // Missing questionAnswer means no accepted attachments; missing resolution leaves delivery ambiguous.
131 const answeredItems = result.storedEvents.flatMap(({ event }) =>
132 event.type === "turn-item.updated" &&
133 event.payload.type === "user_input_request" &&
134 event.payload.requestId === command.requestId
135 ? [event.payload]
136 : [],
137 );
138 return answeredItems.length > 0
139 ? releaseUnusedClaims(
140 claimedPaths,
141 answeredItems.flatMap((item) =>
142 item.questionAnswer === undefined
143 ? []
144 : Object.values(item.questionAnswer.attachmentsByQuestionId).flat(),
145 ),
146 )
147 : Effect.void;
148 }),
149 Effect.tapError((error) =>
150 dispatchWasNotAccepted(error)
151 ? AttachmentClaims.releaseClaimedAttachments(claimedPaths)
152 : Effect.void,
153 ),
154 );
155 }
156 if (
157 command.type !== "message.dispatch" &&
158 (command.type !== "queued-run.edit" || command.attachments === undefined)
159 )
160 return yield* threads.dispatch(command);
161 const claimed = yield* AttachmentClaims.claimPendingAttachments({
162 threadId: command.threadId,
163 attachments: command.attachments ?? [],
164 });
165 return yield* threads
166 .dispatch({
167 ...command,
168 attachments: claimed.attachments,
169 ...(command.context
170 ? {
171 context: remapComposerContextAttachments(
172 command.context,
173 command.attachments ?? [],
174 claimed.attachments,
175 ),
176 }
177 : {}),
178 })
179 .pipe(
180 Effect.tap((result) =>
181 releaseUnusedClaims(
182 claimed.claimedPaths,
183 result.storedEvents.flatMap(({ event }) =>
184 event.type === "message.updated" ? event.payload.attachments : [],
185 ),
186 ),
187 ),
188 Effect.tapError((error) =>
189 dispatchWasNotAccepted(error)
190 ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths)
191 : Effect.void,
192 ),
193 );
194});
195
196export const sendToThread = Effect.fn("ThreadMessageIntake.sendToThread")(function* (
197 input: ThreadManagement.ThreadManagementSendInput,
198) {
199 const threads = yield* ThreadManagement.ThreadManagementService;
200 const claimed = yield* AttachmentClaims.claimPendingAttachments(input);
201 return yield* threads.sendToThread({ ...input, attachments: claimed.attachments }).pipe(
202 Effect.tap((result) => releaseUnusedClaims(claimed.claimedPaths, result.message.attachments)),
203 Effect.tapError((error) =>
204 dispatchWasNotAccepted(error)
205 ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths)
206 : Effect.void,
207 ),
208 );
209});
210
211export const launchThread = Effect.fn("ThreadMessageIntake.launchThread")(function* (
212 input: ThreadLaunch.ThreadLaunchInput,
213) {
214 const launches = yield* ThreadLaunch.ThreadLaunchService;
215 yield* AttachmentClaims.validateAttachmentLimits(input.initialMessage?.attachments ?? []);
216 if (!input.initialMessage?.attachments.some(AttachmentClaims.attachmentIsPendingUpload)) {
217 return yield* launches.launch(input);
218 }
219 if (input.threadId === undefined) {
220 return yield* new AttachmentClaims.AttachmentClaimError({
221 message: "Uploaded attachments need a thread id at launch.",
222 });
223 }
224 const claimed = yield* AttachmentClaims.claimPendingAttachments({
225 threadId: input.threadId,
226 attachments: input.initialMessage.attachments,
227 });
228 return yield* launches
229 .launch({
230 ...input,
231 initialMessage: {
232 ...input.initialMessage,
233 attachments: claimed.attachments,
234 ...(input.initialMessage.context
235 ? {
236 context: remapComposerContextAttachments(
237 input.initialMessage.context,
238 input.initialMessage.attachments,
239 claimed.attachments,
240 ),
241 }
242 : {}),
243 },
244 })
245 .pipe(
246 Effect.tap((result) =>
247 releaseUnusedClaims(
248 claimed.claimedPaths,
249 result.projection.messages.flatMap((message) => message.attachments),
250 ),
251 ),
252 Effect.tapError((error) => {
253 if (error._tag === "ThreadManagementProviderInstanceDisabledError") {
254 return AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths);
255 }
256 // Project/receipt reads precede dispatch; create-thread errors can wrap post-message reads and prove nothing alone.
257 const notAccepted =
258 error.operation === "resolve-project" ||
259 error.operation === "read-receipt" ||
260 ((error.operation === "create-thread" || error.operation === "dispatch-message") &&
261 (isOrchestratorError(error.cause) || isThreadManagementError(error.cause)) &&
262 // Projection errors under create-thread can occur after the message commit.
263 (error.operation !== "create-thread" ||
264 error.cause._tag !== "OrchestratorProjectionError") &&
265 dispatchWasNotAccepted(error.cause));
266 return notAccepted
267 ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths)
268 : Effect.void;
269 }),
270 );
271});