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 dispatcher failures occur in receipt validation or planning, before
15// commitCommand. Generic dispatch errors can follow a commit and remain uncertain.
16// A disabled-instance refusal comes from ThreadManagementService.dispatch's
17// own enablement check, which also runs before any commit is attempted.
18function dispatchWasNotAccepted(
19 error: Orchestrator.OrchestratorV2Error | ThreadManagement.ThreadManagementError,
20) {
21 switch (error._tag) {
22 case "OrchestratorCommandRejectedError":
23 case "OrchestratorProjectionError":
24 case "OrchestratorProviderAdapterError":
25 case "OrchestratorCommandPreviouslyRejectedError":
26 case "OrchestratorCommandIdConflictError":
27 case "OrchestratorSubagentThreadReadOnlyError":
28 case "OrchestratorThreadAboveModeLimitError":
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.map((attachment) =>
46 resolveAttachmentPath({
47 attachmentsDir: config.attachmentsDir,
48 attachment,
49 }),
50 ),
51 );
52 yield* AttachmentClaims.releaseClaimedAttachments(
53 claimedPaths.filter((path) => !retained.has(path)),
54 );
55});
56
57export const dispatchCommand = Effect.fn("ThreadMessageIntake.dispatchCommand")(function* (
58 command: OrchestrationV2Command,
59) {
60 const threads = yield* ThreadManagement.ThreadManagementService;
61 if (command.type === "runtime-request.respond" && command.attachmentsByQuestionId) {
62 const config = yield* ServerConfig.ServerConfig;
63 const incomingByQuestionId = command.attachmentsByQuestionId;
64 yield* AttachmentClaims.validateAttachmentLimits(Object.values(incomingByQuestionId).flat());
65 // Claims accumulate across questions, so all of preparation shares one
66 // rollback boundary: any failure before dispatch removes every new copy.
67 const claimedPaths: string[] = [];
68 const prepared = yield* Effect.gen(function* () {
69 const attachmentsByQuestionId: import("@t3tools/contracts").UserInputAttachments = {};
70 for (const [questionId, attachments] of Object.entries(incomingByQuestionId)) {
71 const claimed = yield* AttachmentClaims.claimPendingAttachments({
72 threadId: command.threadId,
73 attachments,
74 });
75 claimedPaths.push(...claimed.claimedPaths);
76 Object.defineProperty(attachmentsByQuestionId, questionId, {
77 value: claimed.attachments,
78 enumerable: true,
79 });
80 }
81 const answers = yield* appendUserInputAttachmentPaths({
82 answers: command.answers ?? {},
83 attachmentsByQuestionId,
84 attachmentsDir: config.attachmentsDir,
85 }).pipe(
86 Effect.mapError(
87 (cause) => new AttachmentClaims.AttachmentClaimError({ message: cause.issue }),
88 ),
89 );
90 return { answers, attachmentsByQuestionId };
91 }).pipe(Effect.onError(() => AttachmentClaims.releaseClaimedAttachments(claimedPaths)));
92 return yield* threads
93 .dispatch({
94 ...command,
95 answers: prepared.answers,
96 attachmentsByQuestionId: prepared.attachmentsByQuestionId,
97 })
98 .pipe(
99 Effect.tap((result) => {
100 // A replayed receipt reports the first attempt's answer, so this
101 // attempt's copies go unreferenced and are released. The resolved
102 // turn item proves the respond was applied; questionAnswer is only
103 // recorded when the accepted command carried attachments, so its
104 // absence means the accepted answer referenced no copies. With no
105 // resolved item at all the outcome is ambiguous and everything stays.
106 const answeredItems = result.storedEvents.flatMap(({ event }) =>
107 event.type === "turn-item.updated" &&
108 event.payload.type === "user_input_request" &&
109 event.payload.requestId === command.requestId
110 ? [event.payload]
111 : [],
112 );
113 return answeredItems.length > 0
114 ? releaseUnusedClaims(
115 claimedPaths,
116 answeredItems.flatMap((item) =>
117 item.questionAnswer === undefined
118 ? []
119 : Object.values(item.questionAnswer.attachmentsByQuestionId).flat(),
120 ),
121 )
122 : Effect.void;
123 }),
124 Effect.tapError((error) =>
125 dispatchWasNotAccepted(error)
126 ? AttachmentClaims.releaseClaimedAttachments(claimedPaths)
127 : Effect.void,
128 ),
129 );
130 }
131 if (
132 command.type !== "message.dispatch" &&
133 (command.type !== "queued-run.edit" || command.attachments === undefined)
134 )
135 return yield* threads.dispatch(command);
136 const claimed = yield* AttachmentClaims.claimPendingAttachments({
137 threadId: command.threadId,
138 attachments: command.attachments ?? [],
139 });
140 return yield* threads
141 .dispatch({
142 ...command,
143 attachments: claimed.attachments,
144 ...(command.context
145 ? {
146 context: remapComposerContextAttachments(
147 command.context,
148 command.attachments ?? [],
149 claimed.attachments,
150 ),
151 }
152 : {}),
153 })
154 .pipe(
155 Effect.tap((result) =>
156 releaseUnusedClaims(
157 claimed.claimedPaths,
158 result.storedEvents.flatMap(({ event }) =>
159 event.type === "message.updated" ? event.payload.attachments : [],
160 ),
161 ),
162 ),
163 Effect.tapError((error) =>
164 dispatchWasNotAccepted(error)
165 ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths)
166 : Effect.void,
167 ),
168 );
169});
170
171export const sendToThread = Effect.fn("ThreadMessageIntake.sendToThread")(function* (
172 input: ThreadManagement.ThreadManagementSendInput,
173) {
174 const threads = yield* ThreadManagement.ThreadManagementService;
175 const claimed = yield* AttachmentClaims.claimPendingAttachments(input);
176 return yield* threads.sendToThread({ ...input, attachments: claimed.attachments }).pipe(
177 Effect.tap((result) => releaseUnusedClaims(claimed.claimedPaths, result.message.attachments)),
178 Effect.tapError((error) =>
179 dispatchWasNotAccepted(error)
180 ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths)
181 : Effect.void,
182 ),
183 );
184});
185
186export const launchThread = Effect.fn("ThreadMessageIntake.launchThread")(function* (
187 input: ThreadLaunch.ThreadLaunchInput,
188) {
189 const launches = yield* ThreadLaunch.ThreadLaunchService;
190 yield* AttachmentClaims.validateAttachmentLimits(input.initialMessage?.attachments ?? []);
191 if (!input.initialMessage?.attachments.some(AttachmentClaims.attachmentIsPendingUpload)) {
192 return yield* launches.launch(input);
193 }
194 if (input.threadId === undefined) {
195 return yield* new AttachmentClaims.AttachmentClaimError({
196 message: "Uploaded attachments need a thread id at launch.",
197 });
198 }
199 const claimed = yield* AttachmentClaims.claimPendingAttachments({
200 threadId: input.threadId,
201 attachments: input.initialMessage.attachments,
202 });
203 return yield* launches
204 .launch({
205 ...input,
206 initialMessage: {
207 ...input.initialMessage,
208 attachments: claimed.attachments,
209 ...(input.initialMessage.context
210 ? {
211 context: remapComposerContextAttachments(
212 input.initialMessage.context,
213 input.initialMessage.attachments,
214 claimed.attachments,
215 ),
216 }
217 : {}),
218 },
219 })
220 .pipe(
221 Effect.tap((result) =>
222 releaseUnusedClaims(
223 claimed.claimedPaths,
224 result.projection.messages.flatMap((message) => message.attachments),
225 ),
226 ),
227 Effect.tapError((error) => {
228 if (error._tag === "ThreadManagementProviderInstanceDisabledError") {
229 return AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths);
230 }
231 // Project/receipt reads precede message dispatch. The create-thread error
232 // also wraps post-message projection reads, so its tag alone is not proof.
233 // The initial message's own dispatch can still find the instance newly
234 // disabled, which wraps the same tag handled above instead of surfacing
235 // it directly, so the guard below also recognizes it.
236 const notAccepted =
237 error.operation === "resolve-project" ||
238 error.operation === "read-receipt" ||
239 ((error.operation === "create-thread" || error.operation === "dispatch-message") &&
240 (isOrchestratorError(error.cause) || isThreadManagementError(error.cause)) &&
241 // Projection errors under create-thread can occur after the message commit.
242 (error.operation !== "create-thread" ||
243 error.cause._tag !== "OrchestratorProjectionError") &&
244 dispatchWasNotAccepted(error.cause));
245 return notAccepted
246 ? AttachmentClaims.releaseClaimedAttachments(claimed.claimedPaths)
247 : Effect.void;
248 }),
249 );
250});