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 {
8 CommandId,
9 EnvironmentId,
10 IsoDateTime,
11 MessageId,
12 ModelSelection,
13 OrchestrationMessageContext,
14 ProjectId,
15 ProviderInteractionMode,
16 RuntimeMode,
17 ThreadId,
18 type ModelSelection as ModelSelectionType,
19 type ProjectId as ProjectIdType,
20 type ProviderInteractionMode as ProviderInteractionModeType,
21 type RuntimeMode as RuntimeModeType,
22 type ServerProvider,
23} from "@t3tools/contracts";
24import * as Schema from "effect/Schema";
25
26import { DraftComposerAttachmentSchema } from "../lib/composer-image-schema";
27import type { ComposerDispatchMode } from "@t3tools/client-runtime/state/composer-dispatch";
28import type { DraftComposerAttachment } from "../lib/composerImages";
29import { scopedThreadKey } from "../lib/scopedEntities";
30import { resolveProviderInteractionMode } from "./legacy-plan-mode";
31
32// Keep current writes until a compatible native baseline includes the v4 reader.
33const THREAD_OUTBOX_SCHEMA_VERSION = 3;
34const THREAD_OUTBOX_MAX_RETRY_DELAY_MS = 16_000;
35
36const QueuedThreadCreationSchema = Schema.Struct({
37 projectId: ProjectId,
38 // Snapshot of the project's display metadata so a pending task stays
39 // presentable in the thread list even when the project shell is not loaded.
40 projectTitle: Schema.optional(Schema.String),
41 projectCwd: Schema.optional(Schema.String),
42 workspaceMode: Schema.Literals(["local", "worktree"]),
43 branch: Schema.NullOr(Schema.String),
44 worktreePath: Schema.NullOr(Schema.String),
45 startFromOrigin: Schema.optional(Schema.Boolean),
46 pullRequestReference: Schema.optional(Schema.String),
47});
48
49export const QueuedThreadMessageSchema = Schema.Struct({
50 schemaVersion: Schema.Literals([1, 2, THREAD_OUTBOX_SCHEMA_VERSION, 4]),
51 environmentId: EnvironmentId,
52 threadId: ThreadId,
53 messageId: MessageId,
54 commandId: CommandId,
55 text: Schema.String,
56 context: Schema.optional(OrchestrationMessageContext),
57 attachments: Schema.Array(DraftComposerAttachmentSchema),
58 modelSelection: Schema.optional(ModelSelection),
59 dispatchMode: Schema.optional(Schema.Literals(["auto", "queue", "steer", "restart"])),
60 runtimeMode: Schema.optional(RuntimeMode),
61 interactionMode: Schema.optional(ProviderInteractionMode),
62 // Present when the queued item creates a brand-new thread (pending task)
63 // instead of appending a turn to an existing one.
64 creation: Schema.optional(QueuedThreadCreationSchema),
65 createdAt: IsoDateTime,
66});
67
68const decodeStoredQueuedThreadMessage = Schema.decodeUnknownSync(QueuedThreadMessageSchema);
69const encodeStoredQueuedThreadMessage = Schema.encodeUnknownSync(QueuedThreadMessageSchema);
70
71export interface QueuedThreadCreation {
72 readonly projectId: ProjectIdType;
73 readonly projectTitle?: string;
74 readonly projectCwd?: string;
75 readonly workspaceMode: "local" | "worktree";
76 readonly branch: string | null;
77 readonly worktreePath: string | null;
78 readonly startFromOrigin?: boolean;
79 readonly pullRequestReference?: string;
80}
81
82export interface QueuedThreadMessage {
83 readonly environmentId: EnvironmentId;
84 readonly threadId: ThreadId;
85 readonly messageId: MessageId;
86 readonly commandId: CommandId;
87 readonly text: string;
88 readonly context?: OrchestrationMessageContext;
89 readonly attachments: ReadonlyArray<DraftComposerAttachment>;
90 readonly modelSelection?: ModelSelectionType;
91 readonly runtimeMode?: RuntimeModeType;
92 readonly interactionMode?: ProviderInteractionModeType;
93 /**
94 * How this message should be delivered if a turn is still running when the
95 * outbox drains. Captured at enqueue time because the drain can fire long
96 * after the tap. Absent on rows written before follow-up behavior existed,
97 * which keep the previous always-queue delivery.
98 */
99 readonly dispatchMode?: ComposerDispatchMode;
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
110export function resolveQueuedThreadSettings(
111 message: QueuedThreadMessage,
112 thread: ThreadSettingsSnapshot,
113 providers: ReadonlyArray<Pick<ServerProvider, "instanceId" | "showInteractionModeToggle">> = [],
114): ThreadSettingsSnapshot {
115 const modelSelection = message.modelSelection ?? thread.modelSelection;
116 const provider = providers.find(
117 (candidate) => candidate.instanceId === modelSelection.instanceId,
118 );
119 return {
120 modelSelection,
121 runtimeMode: message.runtimeMode ?? thread.runtimeMode,
122 interactionMode: resolveProviderInteractionMode(
123 provider,
124 message.interactionMode ?? thread.interactionMode,
125 ),
126 };
127}
128
129export function modelSelectionsEqual(left: ModelSelectionType, right: ModelSelectionType): boolean {
130 return (
131 left.instanceId === right.instanceId &&
132 left.model === right.model &&
133 JSON.stringify(left.options ?? null) === JSON.stringify(right.options ?? null)
134 );
135}
136
137export function encodeQueuedThreadMessage(message: QueuedThreadMessage): unknown {
138 return encodeStoredQueuedThreadMessage({
139 schemaVersion: THREAD_OUTBOX_SCHEMA_VERSION,
140 ...message,
141 });
142}
143
144export function decodeQueuedThreadMessage(value: unknown): QueuedThreadMessage {
145 const { schemaVersion: _, ...message } = decodeStoredQueuedThreadMessage(value);
146 return message;
147}
148
149export function groupQueuedThreadMessages(
150 messages: ReadonlyArray<QueuedThreadMessage>,
151): Record<string, ReadonlyArray<QueuedThreadMessage>> {
152 const deduplicated = new Map<MessageId, QueuedThreadMessage>();
153 for (const message of messages) {
154 deduplicated.set(message.messageId, message);
155 }
156
157 const grouped: Record<string, Array<QueuedThreadMessage>> = {};
158 for (const message of deduplicated.values()) {
159 const threadKey = scopedThreadKey(message.environmentId, message.threadId);
160 (grouped[threadKey] ??= []).push(message);
161 }
162 for (const queue of Object.values(grouped)) {
163 queue.sort((left, right) => left.createdAt.localeCompare(right.createdAt));
164 }
165 return grouped;
166}
167
168export function flattenQueuedThreadMessages(
169 queues: Record<string, ReadonlyArray<QueuedThreadMessage>>,
170): ReadonlyArray<QueuedThreadMessage> {
171 return Object.values(queues).flat();
172}
173
174export function threadOutboxRetryDelayMs(attempt: number): number {
175 return Math.min(1_000 * 2 ** Math.max(0, attempt - 1), THREAD_OUTBOX_MAX_RETRY_DELAY_MS);
176}
177
178export type ThreadOutboxDeliveryAction = "wait" | "remove" | "send";
179
180export function resolveThreadOutboxDeliveryAction(input: {
181 readonly isCreation: boolean;
182 readonly threadExists: boolean;
183 readonly shellStatus: EnvironmentShellStatus;
184 readonly environmentConnected: boolean;
185 readonly threadBusy: boolean;
186}): ThreadOutboxDeliveryAction {
187 if (input.isCreation) {
188 // A pending task creates its thread on delivery. If the thread already
189 // exists the creation command went through and only cleanup remains.
190 if (input.threadExists) {
191 return "remove";
192 }
193 // Wait for the shell to be live before sending: until the thread list has
194 // synchronized, a previously delivered creation whose cleanup failed would
195 // look missing and get re-issued, duplicating the thread.
196 return input.environmentConnected && input.shellStatus === "live" ? "send" : "wait";
197 }
198 if (!input.threadExists) {
199 return input.shellStatus === "live" ? "remove" : "wait";
200 }
201 return input.environmentConnected ? "send" : "wait";
202}
203
204export type ThreadOutboxDispatchStep =
205 | { readonly step: "wait" }
206 | { readonly step: "remove" }
207 | { readonly step: "retry" }
208 | { readonly step: "restore"; readonly reason: string }
209 | { readonly step: "send" };
210
211/**
212 * Wait for provider and file capabilities before sending. Cleanup does not
213 * need config: a creation whose thread exists, or a message whose thread is
214 * gone, can still be removed while config loads.
215 */
216export function resolveThreadOutboxDispatchStep(input: {
217 readonly deliveryAction: ThreadOutboxDeliveryAction;
218 readonly fileAttachments: ReadonlyArray<{ readonly name: string; readonly sizeBytes: number }>;
219 /** Null while the environment's server config has not synced yet. */
220 readonly serverConfig: { readonly maxFileUploadBytes: number | undefined } | null;
221}): ThreadOutboxDispatchStep {
222 if (input.deliveryAction !== "send") {
223 return { step: input.deliveryAction };
224 }
225 if (input.serverConfig === null) {
226 return { step: "retry" };
227 }
228 if (input.fileAttachments.length === 0) {
229 return { step: "send" };
230 }
231 const maxBytes = input.serverConfig.maxFileUploadBytes;
232 if (maxBytes === undefined) {
233 return { step: "restore", reason: "This server does not support file attachments." };
234 }
235 const effectiveMaxBytes = clampFileAttachmentUploadBytes(maxBytes);
236 const oversized = input.fileAttachments.find(
237 (attachment) => attachment.sizeBytes > effectiveMaxBytes,
238 );
239 return oversized
240 ? { step: "restore", reason: fileAttachmentTooLargeMessage(oversized.name, effectiveMaxBytes) }
241 : { step: "send" };
242}
243
244/**
245 * A queued creation can only be dispatched once its payload would pass server
246 * validation; incomplete payloads stay pending until the user edits them.
247 */
248export function isQueuedThreadCreationSendable(message: QueuedThreadMessage): boolean {
249 if (!message.creation) {
250 return false;
251 }
252 const hasContent = message.text.trim().length > 0 || message.attachments.length > 0;
253 if (!hasContent || message.modelSelection === undefined) {
254 return false;
255 }
256 return message.creation.workspaceMode !== "worktree" || Boolean(message.creation.branch);
257}
258
259function errorMessage(error: unknown): string | null {
260 if (error instanceof Error) {
261 return error.message;
262 }
263 if (typeof error === "object" && error !== null && "message" in error) {
264 return typeof error.message === "string" ? error.message : null;
265 }
266 return typeof error === "string" ? error : null;
267}
268
269/**
270 * Only a failure the server actually decided (`OrchestrationDispatchCommandError`,
271 * or an authorization rejection) means the payload itself is bad. The other
272 * typed failures a queued send can hit are transport-shaped: a socket that
273 * dropped mid-request (`RpcClientError` wrapping a Socket read/write/close
274 * reason), or an environment that is not connected or not registered. Those
275 * are matched by tag, not by message text, because a `SocketReadError` message
276 * is just "An error occurred during Read". A wrong answer here restores the
277 * pending task into a draft and it disappears from the list.
278 */
279export function shouldRetryThreadOutboxDelivery(error: unknown): boolean {
280 if (typeof error === "object" && error !== null && "_tag" in error) {
281 switch (error._tag) {
282 case "OrchestrationDispatchCommandError":
283 case "EnvironmentAuthorizationError":
284 return false;
285 case "ConnectionTransientError":
286 case "RpcClientError":
287 case "EnvironmentRpcUnavailableError":
288 case "EnvironmentNotRegisteredError":
289 return true;
290 default:
291 break;
292 }
293 }
294 return isTransportConnectionErrorMessage(errorMessage(error));
295}
296
297export type ThreadOutboxCommandStage = "settings-sync" | "start-turn";
298export type ThreadOutboxFailureAction = "retry" | "restore";
299
300export function resolveThreadOutboxFailureAction(input: {
301 readonly stage: ThreadOutboxCommandStage;
302 readonly error: unknown;
303 readonly interrupted: boolean;
304}): ThreadOutboxFailureAction {
305 if (
306 input.stage === "settings-sync" ||
307 input.interrupted ||
308 shouldRetryThreadOutboxDelivery(input.error)
309 ) {
310 return "retry";
311 }
312 return "restore";
313}