1import { isTransportConnectionErrorMessage } from "@t3tools/client-runtime/errors";
2import {
3 clampFileAttachmentUploadBytes,
4 fileAttachmentTooLargeMessage,
5} from "@t3tools/client-runtime/state/attachments";
6import type { EnvironmentShellStatus } from "@t3tools/client-runtime/state/shell";
7import type { ThreadFamilyRead } from "@t3tools/client-runtime/state/threads";
8import {
9 CommandId,
10 DEFAULT_PROVIDER_INTERACTION_MODE,
11 EnvironmentId,
12 IsoDateTime,
13 MessageId,
14 ModelSelection,
15 OrchestrationMessageContext,
16 ProjectId,
17 ProviderInteractionMode,
18 RuntimeMode,
19 ThreadId,
20 type ModelSelection as ModelSelectionType,
21 type ProjectId as ProjectIdType,
22 type ProviderInteractionMode as ProviderInteractionModeType,
23 type RuntimeMode as RuntimeModeType,
24 type ServerProvider,
25 type ServerProviderModel,
26} from "@t3tools/contracts";
27import * as Schema from "effect/Schema";
28
29import { DraftComposerAttachmentSchema } from "../lib/composer-image-schema";
30import { withImplicitFastModeSelection } from "../lib/providerOptions";
31import type { ComposerDispatchMode } from "@t3tools/client-runtime/state/composer-dispatch";
32import type { DraftComposerAttachment } from "../lib/composerImages";
33import { scopedThreadKey } from "../lib/scopedEntities";
34
35// Keep current writes until a compatible native baseline includes the v4 reader.
36const THREAD_OUTBOX_SCHEMA_VERSION = 3;
37const THREAD_OUTBOX_MAX_RETRY_DELAY_MS = 16_000;
38
39const QueuedThreadCreationSchema = Schema.Struct({
40 projectId: ProjectId,
41 projectTitle: Schema.optional(Schema.String),
42 projectCwd: Schema.optional(Schema.String),
43 workspaceMode: Schema.Literals(["local", "worktree"]),
44 branch: Schema.NullOr(Schema.String),
45 worktreePath: Schema.NullOr(Schema.String),
46 startFromOrigin: Schema.optional(Schema.Boolean),
47 pullRequestReference: Schema.optional(Schema.String),
48});
49
50export const QueuedThreadMessageSchema = Schema.Struct({
51 schemaVersion: Schema.Literals([1, 2, THREAD_OUTBOX_SCHEMA_VERSION, 4]),
52 environmentId: EnvironmentId,
53 threadId: ThreadId,
54 messageId: MessageId,
55 commandId: CommandId,
56 text: Schema.String,
57 context: Schema.optional(OrchestrationMessageContext),
58 attachments: Schema.Array(DraftComposerAttachmentSchema),
59 modelSelection: Schema.optional(ModelSelection),
60 dispatchMode: Schema.optional(Schema.Literals(["auto", "queue", "steer", "restart"])),
61 compactBeforeSend: Schema.optional(Schema.Boolean),
62 freshContext: Schema.optional(Schema.Boolean),
63 runtimeMode: Schema.optional(RuntimeMode),
64 interactionMode: Schema.optional(ProviderInteractionMode),
65 creation: Schema.optional(QueuedThreadCreationSchema),
66 createdAt: IsoDateTime,
67});
68
69const decodeStoredQueuedThreadMessage = Schema.decodeUnknownSync(QueuedThreadMessageSchema);
70const encodeStoredQueuedThreadMessage = Schema.encodeUnknownSync(QueuedThreadMessageSchema);
71
72export interface QueuedThreadCreation {
73 readonly projectId: ProjectIdType;
74 readonly projectTitle?: string;
75 readonly projectCwd?: string;
76 readonly workspaceMode: "local" | "worktree";
77 readonly branch: string | null;
78 readonly worktreePath: string | null;
79 readonly startFromOrigin?: boolean;
80 readonly pullRequestReference?: string;
81}
82
83export interface QueuedThreadMessage {
84 readonly environmentId: EnvironmentId;
85 readonly threadId: ThreadId;
86 readonly messageId: MessageId;
87 readonly commandId: CommandId;
88 readonly text: string;
89 readonly context?: OrchestrationMessageContext;
90 readonly attachments: ReadonlyArray<DraftComposerAttachment>;
91 readonly modelSelection?: ModelSelectionType;
92 readonly runtimeMode?: RuntimeModeType;
93 readonly interactionMode?: ProviderInteractionModeType;
94 /** Captured at enqueue because draining may occur long after the tap. */
95 readonly dispatchMode?: ComposerDispatchMode;
96 /** Sends "When you return"'s `/compact` ahead of the message, which queues behind it. */
97 readonly compactBeforeSend?: boolean;
98 /** Starts a new provider session (`/clear`). */
99 readonly freshContext?: boolean;
100 readonly creation?: QueuedThreadCreation;
101 readonly createdAt: string;
102}
103
104export interface ThreadSettingsSnapshot {
105 readonly modelSelection: ModelSelectionType;
106 readonly runtimeMode: RuntimeModeType;
107 readonly interactionMode: ProviderInteractionModeType;
108}
109
110// Providers without a Build/Plan toggle always run in build mode.
111function resolveProviderInteractionMode(
112 provider: Pick<ServerProvider, "showInteractionModeToggle"> | undefined,
113 interactionMode: ProviderInteractionModeType,
114): ProviderInteractionModeType {
115 return provider?.showInteractionModeToggle === false
116 ? DEFAULT_PROVIDER_INTERACTION_MODE
117 : interactionMode;
118}
119
120export function resolveQueuedThreadSettings(
121 message: QueuedThreadMessage,
122 thread: ThreadSettingsSnapshot,
123 providers: ReadonlyArray<
124 Pick<ServerProvider, "instanceId" | "showInteractionModeToggle"> & {
125 readonly models: ReadonlyArray<Pick<ServerProviderModel, "slug" | "capabilities">>;
126 }
127 > = [],
128): ThreadSettingsSnapshot {
129 const modelSelection = withImplicitFastModeSelection(
130 message.modelSelection ?? thread.modelSelection,
131 providers,
132 );
133 const provider = providers.find(
134 (candidate) => candidate.instanceId === modelSelection.instanceId,
135 );
136 return {
137 modelSelection,
138 runtimeMode: message.runtimeMode ?? thread.runtimeMode,
139 interactionMode: resolveProviderInteractionMode(
140 provider,
141 message.interactionMode ?? thread.interactionMode,
142 ),
143 };
144}
145
146export function modelSelectionsEqual(left: ModelSelectionType, right: ModelSelectionType): boolean {
147 return (
148 left.instanceId === right.instanceId &&
149 left.model === right.model &&
150 JSON.stringify(left.options ?? null) === JSON.stringify(right.options ?? null)
151 );
152}
153
154export function encodeQueuedThreadMessage(message: QueuedThreadMessage): unknown {
155 return encodeStoredQueuedThreadMessage({
156 schemaVersion: THREAD_OUTBOX_SCHEMA_VERSION,
157 ...message,
158 });
159}
160
161export function decodeQueuedThreadMessage(value: unknown): QueuedThreadMessage {
162 const { schemaVersion: _, ...message } = decodeStoredQueuedThreadMessage(value);
163 return message;
164}
165
166export function groupQueuedThreadMessages(
167 messages: ReadonlyArray<QueuedThreadMessage>,
168): Record<string, ReadonlyArray<QueuedThreadMessage>> {
169 const deduplicated = new Map<MessageId, QueuedThreadMessage>();
170 for (const message of messages) {
171 deduplicated.set(message.messageId, message);
172 }
173
174 const grouped: Record<string, Array<QueuedThreadMessage>> = {};
175 for (const message of deduplicated.values()) {
176 const threadKey = scopedThreadKey(message.environmentId, message.threadId);
177 (grouped[threadKey] ??= []).push(message);
178 }
179 for (const queue of Object.values(grouped)) {
180 queue.sort((left, right) => left.createdAt.localeCompare(right.createdAt));
181 }
182 return grouped;
183}
184
185export function flattenQueuedThreadMessages(
186 queues: Record<string, ReadonlyArray<QueuedThreadMessage>>,
187): ReadonlyArray<QueuedThreadMessage> {
188 return Object.values(queues).flat();
189}
190
191export function threadOutboxRetryDelayMs(attempt: number): number {
192 return Math.min(1_000 * 2 ** Math.max(0, attempt - 1), THREAD_OUTBOX_MAX_RETRY_DELAY_MS);
193}
194
195export type ThreadOutboxDeliveryAction = "wait" | "remove" | "send";
196
197export function resolveThreadOutboxDeliveryAction(input: {
198 readonly isCreation: boolean;
199 readonly threadExists: boolean;
200 readonly shellStatus: EnvironmentShellStatus;
201 readonly environmentConnected: boolean;
202 /**
203 * Set when the shell can leave idle subagent children out: absence alone is not deletion there,
204 * only a settled family read without the row is.
205 */
206 readonly familyRead?: ThreadFamilyRead;
207}): ThreadOutboxDeliveryAction {
208 if (input.isCreation) {
209 if (input.threadExists) {
210 return "remove";
211 }
212 // Wait for live shells; an unsynchronized list can hide delivered creations and cause duplicate re-issuance.
213 return input.environmentConnected && input.shellStatus === "live" ? "send" : "wait";
214 }
215 if (!input.threadExists) {
216 return input.shellStatus === "live" &&
217 (input.familyRead === undefined || input.familyRead === "settled")
218 ? "remove"
219 : "wait";
220 }
221 return input.environmentConnected ? "send" : "wait";
222}
223
224export type ThreadOutboxDispatchStep =
225 | { readonly step: "wait" }
226 | { readonly step: "remove" }
227 | { readonly step: "retry" }
228 | { readonly step: "restore"; readonly reason: string }
229 | { readonly step: "send" };
230
231export function resolveThreadOutboxDispatchStep(input: {
232 readonly deliveryAction: ThreadOutboxDeliveryAction;
233 readonly fileAttachments: ReadonlyArray<{ readonly name: string; readonly sizeBytes: number }>;
234 readonly serverConfig: { readonly maxFileUploadBytes: number | undefined } | null;
235}): ThreadOutboxDispatchStep {
236 if (input.deliveryAction !== "send") {
237 return { step: input.deliveryAction };
238 }
239 if (input.serverConfig === null) {
240 return { step: "retry" };
241 }
242 if (input.fileAttachments.length === 0) {
243 return { step: "send" };
244 }
245 const maxBytes = input.serverConfig.maxFileUploadBytes;
246 if (maxBytes === undefined) {
247 return { step: "restore", reason: "This server does not support file attachments." };
248 }
249 const effectiveMaxBytes = clampFileAttachmentUploadBytes(maxBytes);
250 const oversized = input.fileAttachments.find(
251 (attachment) => attachment.sizeBytes > effectiveMaxBytes,
252 );
253 return oversized
254 ? { step: "restore", reason: fileAttachmentTooLargeMessage(oversized.name, effectiveMaxBytes) }
255 : { step: "send" };
256}
257
258export function isQueuedThreadCreationSendable(message: QueuedThreadMessage): boolean {
259 if (!message.creation) {
260 return false;
261 }
262 const hasContent = message.text.trim().length > 0 || message.attachments.length > 0;
263 if (!hasContent || message.modelSelection === undefined) {
264 return false;
265 }
266 return message.creation.workspaceMode !== "worktree" || Boolean(message.creation.branch);
267}
268
269function errorMessage(error: unknown): string | null {
270 if (error instanceof Error) {
271 return error.message;
272 }
273 if (typeof error === "object" && error !== null && "message" in error) {
274 return typeof error.message === "string" ? error.message : null;
275 }
276 return typeof error === "string" ? error : null;
277}
278
279/** Dispatch/auth rejections invalidate payloads; other typed failures may be disconnected transports and must retry. */
280export function shouldRetryThreadOutboxDelivery(error: unknown): boolean {
281 if (typeof error === "object" && error !== null && "_tag" in error) {
282 switch (error._tag) {
283 case "OrchestrationDispatchCommandError":
284 case "EnvironmentAuthorizationError":
285 return false;
286 case "ConnectionTransientError":
287 case "RpcClientError":
288 case "EnvironmentRpcUnavailableError":
289 case "EnvironmentNotRegisteredError":
290 return true;
291 default:
292 break;
293 }
294 }
295 return isTransportConnectionErrorMessage(errorMessage(error));
296}
297
298export type ThreadOutboxCommandStage = "settings-sync" | "start-turn";
299export type ThreadOutboxFailureAction = "retry" | "restore";
300
301export function resolveThreadOutboxFailureAction(input: {
302 readonly stage: ThreadOutboxCommandStage;
303 readonly error: unknown;
304 readonly interrupted: boolean;
305}): ThreadOutboxFailureAction {
306 if (
307 input.stage === "settings-sync" ||
308 input.interrupted ||
309 shouldRetryThreadOutboxDelivery(input.error)
310 ) {
311 return "retry";
312 }
313 return "restore";
314}
315
316export type AcknowledgedThreadMessage = QueuedThreadMessage & {
317 /** The server sequence that committed the message; older servers do not report it. */
318 readonly sequence?: number;
319};