1import { remapComposerContextAttachments } from "@t3tools/shared/composerContextReferences";
2import {
3 type ThreadLinkedPullRequest,
4 CommandId,
5 CheckpointId,
6 CheckpointScopeId,
7 ORCHESTRATION_V2_WS_METHODS,
8 OrchestrationV2CheckpointUnavailableError,
9 WS_METHODS,
10 type ChatAttachment,
11 MessageId,
12 type ModelSelection,
13 type NodeId,
14 type OrchestrationV2Command,
15 type OrchestrationV2CreationSource,
16 type PlanId,
17 type ProjectId,
18 type ProjectIconOverride,
19 type ProjectScript,
20 type ProviderApprovalDecision,
21 type ProviderInteractionMode,
22 type ProviderUserInputAnswers,
23 type RunId,
24 type RuntimeMode,
25 type RuntimeRequestId,
26 type ThreadId,
27 type ThreadEnvMode,
28 type UploadChatAttachment,
29 type GuestMessageId,
30} from "@t3tools/contracts";
31import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
32import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests";
33import * as SecureRandom from "effect/crypto/SecureRandom";
34import * as Cause from "effect/Cause";
35import * as Effect from "effect/Effect";
36import * as Exit from "effect/Exit";
37import * as Option from "effect/Option";
38import * as Stream from "effect/Stream";
39import * as SubscriptionRef from "effect/SubscriptionRef";
40
41import * as EnvironmentSupervisor from "../connection/supervisor.ts";
42import {
43 EnvironmentRpcUnavailableError,
44 getInitialServerConfig,
45 isRpcClientError,
46 request,
47} from "../rpc/client.ts";
48
49interface CommandMetadata {
50 readonly commandId?: CommandId;
51 readonly createdAt?: string;
52 readonly creationSource?: OrchestrationV2CreationSource;
53}
54
55export interface CreateProjectInput extends CommandMetadata {
56 readonly projectId: ProjectId;
57 readonly title: string;
58 readonly workspaceRoot: string;
59 readonly createWorkspaceRootIfMissing?: boolean;
60 readonly defaultModelSelection?: ModelSelection | null;
61 readonly scripts?: ReadonlyArray<ProjectScript>;
62}
63
64export interface UpdateProjectInput extends CommandMetadata {
65 readonly projectId: ProjectId;
66 readonly title?: string;
67 readonly workspaceRoot?: string;
68 readonly defaultModelSelection?: ModelSelection | null;
69 readonly autoPull?: boolean;
70 readonly projectIcon?: ProjectIconOverride | null;
71 readonly faviconPath?: string | null;
72 readonly defaultThreadEnvMode?: ThreadEnvMode | null;
73 readonly scripts?: ReadonlyArray<ProjectScript>;
74 readonly pinned?: boolean;
75 readonly archived?: boolean;
76}
77
78export interface DeleteProjectInput extends CommandMetadata {
79 readonly projectId: ProjectId;
80 readonly force?: boolean;
81}
82
83export interface CreateThreadInput extends CommandMetadata {
84 readonly threadId: ThreadId;
85 readonly projectId: ProjectId;
86 readonly title: string;
87 readonly modelSelection: ModelSelection;
88 readonly runtimeMode: RuntimeMode;
89 readonly interactionMode: ProviderInteractionMode;
90 readonly branch: string | null;
91 readonly worktreePath: string | null;
92}
93
94export interface ThreadCommandInput extends CommandMetadata {
95 readonly threadId: ThreadId;
96}
97
98export interface DeleteThreadInput extends ThreadCommandInput {
99 /** The server deletes only a thread with no runs and no messages. */
100 readonly onlyIfUnused?: boolean;
101}
102export type ArchiveThreadInput = ThreadCommandInput;
103export type UnarchiveThreadInput = ThreadCommandInput;
104export type SettleThreadInput = ThreadCommandInput;
105
106export interface UnsettleThreadInput extends ThreadCommandInput {
107 readonly reason: "user";
108}
109
110export interface PinThreadInput extends ThreadCommandInput {
111 readonly orderKey?: string;
112}
113export type UnpinThreadInput = ThreadCommandInput;
114
115export interface ReorderPinnedThreadInput extends ThreadCommandInput {
116 readonly orderKey: string;
117}
118
119export interface ReorderActiveThreadInput extends ThreadCommandInput {
120 readonly orderKey: string;
121}
122
123export interface SnoozeThreadInput extends ThreadCommandInput {
124 readonly snoozedUntil: string;
125}
126
127export interface UnsnoozeThreadInput extends ThreadCommandInput {
128 readonly reason: "user";
129}
130
131export interface VisitThreadInput extends ThreadCommandInput {
132 readonly visitedAt: string;
133 /** Counts toward search frecency; `searchQuery` is the palette query that picked it. */
134 readonly opened?: boolean;
135 readonly searchQuery?: string;
136}
137
138export type MarkThreadUnreadInput = ThreadCommandInput;
139
140export interface UpdateThreadMetadataInput extends ThreadCommandInput {
141 readonly limitRecovery?: import("@t3tools/contracts").OrchestrationV2LimitRecoveryUpdate | null;
142 readonly title?: string;
143 readonly modelSelection?: ModelSelection;
144 readonly branch?: string | null;
145 readonly worktreePath?: string | null;
146 readonly regenerateTitle?: boolean;
147 readonly linkedPullRequest?: ThreadLinkedPullRequest | null;
148 readonly expectedEmpty?: boolean;
149}
150
151export interface SetThreadRuntimeModeInput extends ThreadCommandInput {
152 readonly runtimeMode: RuntimeMode;
153}
154
155export interface SetThreadInteractionModeInput extends ThreadCommandInput {
156 readonly interactionMode: ProviderInteractionMode;
157}
158
159interface StartThreadBootstrap {
160 readonly createThread?: {
161 readonly projectId: ProjectId;
162 readonly title: string;
163 readonly modelSelection: ModelSelection;
164 readonly runtimeMode: RuntimeMode;
165 readonly interactionMode: ProviderInteractionMode;
166 readonly branch: string | null;
167 readonly worktreePath: string | null;
168 readonly createdAt: string;
169 };
170 readonly prepareWorktree?: {
171 readonly requireWorktree?: boolean;
172 readonly projectCwd: string;
173 readonly baseBranch: string;
174 readonly branch?: string;
175 readonly startFromOrigin?: boolean;
176 };
177 readonly runSetupScript?: boolean;
178}
179
180export interface StartThreadTurnInput extends ThreadCommandInput {
181 readonly manualContinuationOfRunId?: RunId;
182 readonly message: {
183 readonly messageId: MessageId;
184 readonly role: "user";
185 readonly text: string;
186 readonly attachments: ReadonlyArray<ChatAttachment | UploadChatAttachment>;
187 readonly context?: import("@t3tools/contracts").OrchestrationMessageContext;
188 readonly idleCompaction?: import("@t3tools/contracts").OrchestrationV2IdleCompaction;
189 };
190 readonly modelSelection?: ModelSelection;
191 readonly titleSeed?: string;
192 readonly runtimeMode: RuntimeMode;
193 readonly interactionMode: ProviderInteractionMode;
194 readonly bootstrap?: StartThreadBootstrap;
195 readonly sourceProposedPlan?: { readonly threadId: ThreadId; readonly planId: PlanId };
196 readonly dispatchMode?: "auto" | "queue" | "steer" | "restart" | "start";
197 /** Start a new provider session; needs the server's `freshContext` capability. */
198 readonly freshContext?: boolean;
199 /** Defaults to true. Callers with their own retry queue opt out so a dead socket fails fast. */
200 readonly resendAfterSocketLoss?: boolean;
201}
202
203export interface InterruptThreadTurnInput extends ThreadCommandInput {
204 readonly runId?: RunId;
205 readonly subagentId?: NodeId;
206 readonly turnId?: string;
207}
208
209export interface RespondToThreadApprovalInput extends ThreadCommandInput {
210 readonly requestId: RuntimeRequestId;
211 readonly decision: ProviderApprovalDecision;
212 readonly feedback?: string;
213}
214
215export interface RespondToThreadUserInputInput extends ThreadCommandInput {
216 readonly requestId: RuntimeRequestId;
217 readonly answers: ProviderUserInputAnswers;
218 readonly attachmentsByQuestionId?: import("@t3tools/contracts").UserInputAttachments;
219}
220
221export interface DismissThreadUserInputInput extends ThreadCommandInput {
222 readonly requestId: RuntimeRequestId;
223}
224
225export interface RevertThreadCheckpointInput extends ThreadCommandInput {
226 readonly restoreFiles?: boolean;
227 readonly checkpointId?: string;
228 readonly scopeId?: string;
229 readonly turnCount?: number;
230}
231
232export type StopThreadSessionInput = ThreadCommandInput;
233
234export interface ForkThreadFromRunInput extends CommandMetadata {
235 readonly sourceThreadId: ThreadId;
236 readonly targetThreadId: ThreadId;
237 readonly runId: RunId;
238 readonly title?: string;
239}
240
241export interface ForkSideChatInput extends CommandMetadata {
242 readonly sourceThreadId: ThreadId;
243 readonly targetThreadId: ThreadId;
244 readonly title?: string;
245}
246
247export interface MergeThreadBackInput extends CommandMetadata {
248 readonly sourceThreadId: ThreadId;
249 readonly targetThreadId: ThreadId;
250 readonly runId: RunId;
251}
252
253export interface ResumeThreadQueueInput extends ThreadCommandInput {
254 readonly runId?: RunId;
255}
256
257export interface PromoteQueuedRunInput extends ThreadCommandInput {
258 readonly queuedRunId: RunId;
259 readonly targetRunId: RunId;
260}
261
262export interface CancelQueuedRunInput extends ThreadCommandInput {
263 readonly runId: RunId;
264}
265
266export interface RetryWorkspacePreparationInput extends ThreadCommandInput {
267 readonly runId: RunId;
268}
269
270export interface EditQueuedRunInput extends ThreadCommandInput {
271 readonly runId: RunId;
272 readonly text: string;
273 /**
274 * Full replacement attachment list for the queued message. Omitted =
275 * text-only edit that leaves attachments untouched.
276 */
277 readonly edit?: {
278 readonly messageId: MessageId;
279 readonly attachments: ReadonlyArray<ChatAttachment | UploadChatAttachment>;
280 readonly context?: import("@t3tools/contracts").OrchestrationMessageContext;
281 };
282}
283
284const allocateCommandId = Effect.fn("EnvironmentCommands.allocateCommandId")(function* (
285 input: CommandMetadata,
286) {
287 if (input.commandId !== undefined) return input.commandId;
288 const random = yield* SecureRandom.SecureRandom;
289 return CommandId.make(yield* random.uuidV4());
290});
291
292const dispatch = (command: OrchestrationV2Command) =>
293 request(ORCHESTRATION_V2_WS_METHODS.dispatchCommand, command);
294
295// The socket can close after the server accepted the frame but before it answered.
296// The server replays an accepted commandId from its receipt, so the same command
297// waits out reconnect backoff until the supervisor stops trying. The resend cap
298// stops a frame that kills the socket from looping.
299const MAX_LOST_TRANSPORT_RESENDS = 2;
300
301// A live socket can still carry a command the host never answers; the heartbeat
302// alone would leave the send pending forever.
303const MESSAGE_DISPATCH_TIMEOUT = "60 seconds";
304
305const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(function* (
306 command: Extract<OrchestrationV2Command, { type: "message.dispatch" }>,
307 resendAfterSocketLoss: boolean,
308) {
309 const supervisor = yield* EnvironmentSupervisor.EnvironmentSupervisor;
310 const attempt = dispatch(command).pipe(
311 Effect.timeoutOrElse({
312 duration: MESSAGE_DISPATCH_TIMEOUT,
313 orElse: () =>
314 Effect.fail(
315 new EnvironmentRpcUnavailableError({
316 environmentId: supervisor.target.environmentId,
317 message: `${supervisor.target.label} did not confirm the message within 60 seconds.`,
318 }),
319 ),
320 }),
321 );
322 if (!resendAfterSocketLoss) return yield* attempt;
323 for (let resends = 0; ; resends += 1) {
324 const attemptSession = yield* SubscriptionRef.get(supervisor.session);
325 const exit = yield* Effect.exit(attempt);
326 if (Exit.isSuccess(exit)) return exit.value;
327 // Closing the session interrupts the request, which can beat the protocol error.
328 const lostTransport =
329 Cause.hasInterruptsOnly(exit.cause) ||
330 exit.cause.reasons.some((reason) => reason._tag === "Fail" && isRpcClientError(reason.error));
331 if (!lostTransport || resends >= MAX_LOST_TRANSPORT_RESENDS) return yield* exit;
332 const replacement = yield* Effect.raceFirst(
333 SubscriptionRef.changes(supervisor.session).pipe(
334 Stream.filter(
335 (session) =>
336 Option.isSome(session) &&
337 !(Option.isSome(attemptSession) && session.value === attemptSession.value),
338 ),
339 Stream.runHead,
340 ),
341 SubscriptionRef.changes(supervisor.state).pipe(
342 Stream.filter((state) => state.phase === "available" || state.phase === "blocked"),
343 Stream.runHead,
344 Effect.as(Option.none()),
345 ),
346 );
347 if (Option.isNone(replacement)) return yield* exit;
348 }
349});
350
351const getProjection = (threadId: ThreadId) =>
352 request(ORCHESTRATION_V2_WS_METHODS.getThreadProjection, { threadId });
353
354const persistAttachments = Effect.fn("EnvironmentCommands.persistAttachments")(function* (
355 threadId: ThreadId,
356 messageId: MessageId,
357 attachments: ReadonlyArray<ChatAttachment | UploadChatAttachment>,
358) {
359 const stored = attachments.filter(
360 (attachment): attachment is ChatAttachment => !("dataUrl" in attachment),
361 );
362 const uploads = attachments.filter(
363 (attachment): attachment is UploadChatAttachment => "dataUrl" in attachment,
364 );
365 if (uploads.length === 0) return stored;
366 const result = yield* request(WS_METHODS.assetsPersistChatAttachments, {
367 threadId,
368 messageId,
369 attachments: uploads,
370 });
371 if (stored.length === 0) return result.attachments;
372 const byUpload = new Map(
373 uploads.map((attachment, index) => [attachment, result.attachments[index]]),
374 );
375 return attachments.flatMap((attachment) => {
376 if (!("dataUrl" in attachment)) return [attachment];
377 const persisted = byUpload.get(attachment);
378 return persisted === undefined ? [] : [persisted];
379 });
380});
381
382const mutateProject = Effect.fn("EnvironmentCommands.mutateProject")(function* (
383 mutation:
384 | {
385 readonly type: "project.create";
386 readonly commandId: CommandId;
387 readonly projectId: ProjectId;
388 readonly title: string;
389 readonly workspaceRoot: string;
390 readonly createWorkspaceRootIfMissing?: boolean;
391 readonly defaultModelSelection?: ModelSelection | null;
392 readonly scripts?: ReadonlyArray<ProjectScript>;
393 }
394 | {
395 readonly type: "project.update";
396 readonly commandId: CommandId;
397 readonly projectId: ProjectId;
398 readonly title?: string;
399 readonly workspaceRoot?: string;
400 readonly defaultModelSelection?: ModelSelection | null;
401 readonly scripts?: ReadonlyArray<ProjectScript>;
402 }
403 | {
404 readonly type: "project.delete";
405 readonly commandId: CommandId;
406 readonly projectId: ProjectId;
407 readonly force?: boolean;
408 },
409) {
410 return yield* request(WS_METHODS.projectsMutate, mutation);
411});
412
413export const createProject = Effect.fn("EnvironmentCommands.createProject")(function* (
414 input: CreateProjectInput,
415) {
416 return yield* mutateProject({
417 type: "project.create",
418 commandId: yield* allocateCommandId(input),
419 projectId: input.projectId,
420 title: input.title,
421 workspaceRoot: input.workspaceRoot,
422 ...(input.createWorkspaceRootIfMissing === undefined
423 ? {}
424 : { createWorkspaceRootIfMissing: input.createWorkspaceRootIfMissing }),
425 ...(input.defaultModelSelection === undefined
426 ? {}
427 : { defaultModelSelection: input.defaultModelSelection }),
428 ...(input.scripts === undefined ? {} : { scripts: input.scripts }),
429 });
430});
431
432export const updateProject = Effect.fn("EnvironmentCommands.updateProject")(function* (
433 input: UpdateProjectInput,
434) {
435 return yield* mutateProject({
436 type: "project.update",
437 commandId: yield* allocateCommandId(input),
438 projectId: input.projectId,
439 ...(input.title === undefined ? {} : { title: input.title }),
440 ...(input.workspaceRoot === undefined ? {} : { workspaceRoot: input.workspaceRoot }),
441 ...(input.defaultModelSelection === undefined
442 ? {}
443 : { defaultModelSelection: input.defaultModelSelection }),
444 ...(input.autoPull === undefined ? {} : { autoPull: input.autoPull }),
445 ...(input.projectIcon === undefined ? {} : { projectIcon: input.projectIcon }),
446 ...(input.faviconPath === undefined ? {} : { faviconPath: input.faviconPath }),
447 ...(input.defaultThreadEnvMode === undefined
448 ? {}
449 : { defaultThreadEnvMode: input.defaultThreadEnvMode }),
450 ...(input.scripts === undefined ? {} : { scripts: input.scripts }),
451 ...(input.pinned === undefined ? {} : { pinned: input.pinned }),
452 ...(input.archived === undefined ? {} : { archived: input.archived }),
453 });
454});
455
456export const deleteProject = Effect.fn("EnvironmentCommands.deleteProject")(function* (
457 input: DeleteProjectInput,
458) {
459 return yield* mutateProject({
460 type: "project.delete",
461 commandId: yield* allocateCommandId(input),
462 projectId: input.projectId,
463 ...(input.force === undefined ? {} : { force: input.force }),
464 });
465});
466
467export const createThread = Effect.fn("EnvironmentCommands.createThread")(function* (
468 input: CreateThreadInput,
469) {
470 return yield* dispatch({
471 type: "thread.create",
472 commandId: yield* allocateCommandId(input),
473 createdBy: "user",
474 creationSource: input.creationSource ?? "web",
475 threadId: input.threadId,
476 projectId: input.projectId,
477 title: input.title,
478 modelSelection: input.modelSelection,
479 runtimeMode: input.runtimeMode,
480 interactionMode: input.interactionMode,
481 branch: input.branch,
482 worktreePath: input.worktreePath,
483 });
484});
485
486function simpleThreadCommand(
487 type:
488 | "thread.archive"
489 | "thread.unarchive"
490 | "thread.settle"
491 | "thread.pin"
492 | "thread.unpin"
493 | "thread.guest-share.close",
494 input: ThreadCommandInput,
495) {
496 return allocateCommandId(input).pipe(
497 Effect.flatMap((commandId) => dispatch({ type, commandId, threadId: input.threadId })),
498 );
499}
500
501export const deleteThread = Effect.fn("EnvironmentCommands.deleteThread")(function* (
502 input: DeleteThreadInput,
503) {
504 return yield* dispatch({
505 type: "thread.delete",
506 commandId: yield* allocateCommandId(input),
507 threadId: input.threadId,
508 ...(input.onlyIfUnused === undefined ? {} : { onlyIfUnused: input.onlyIfUnused }),
509 });
510});
511
512export type CloseGuestShareInput = ThreadCommandInput;
513export const closeGuestShare = Effect.fn("EnvironmentCommands.closeGuestShare")(function* (
514 input: CloseGuestShareInput,
515) {
516 return yield* simpleThreadCommand("thread.guest-share.close", input);
517});
518
519export interface ResolveGuestMessageInput extends ThreadCommandInput {
520 readonly guestMessageId: GuestMessageId;
521 readonly decision: "send" | "dismiss";
522 readonly creationSource?: "web" | "mobile";
523}
524/** Send runs the guest's text as a normal turn; the server supplies the text, not the client. */
525export const resolveGuestMessage = Effect.fn("EnvironmentCommands.resolveGuestMessage")(function* (
526 input: ResolveGuestMessageInput,
527) {
528 const commandId = yield* allocateCommandId(input);
529 if (input.decision === "dismiss") {
530 return yield* dispatch({
531 type: "thread.guest-message.dismiss",
532 commandId,
533 threadId: input.threadId,
534 guestMessageId: input.guestMessageId,
535 });
536 }
537 return yield* dispatch({
538 type: "message.dispatch",
539 commandId,
540 threadId: input.threadId,
541 messageId: MessageId.make(commandId),
542 guestMessageId: input.guestMessageId,
543 text: "",
544 attachments: [],
545 dispatchMode: { type: "queue_after_active" },
546 createdBy: "user",
547 creationSource: input.creationSource ?? "web",
548 });
549});
550
551export const archiveThread = Effect.fn("EnvironmentCommands.archiveThread")(function* (
552 input: ArchiveThreadInput,
553) {
554 return yield* simpleThreadCommand("thread.archive", input);
555});
556
557export const unarchiveThread = Effect.fn("EnvironmentCommands.unarchiveThread")(function* (
558 input: UnarchiveThreadInput,
559) {
560 return yield* simpleThreadCommand("thread.unarchive", input);
561});
562
563export const settleThread = Effect.fn("EnvironmentCommands.settleThread")(function* (
564 input: SettleThreadInput,
565) {
566 return yield* simpleThreadCommand("thread.settle", input);
567});
568
569export const pinThread = Effect.fn("EnvironmentCommands.pinThread")(function* (
570 input: PinThreadInput,
571) {
572 const commandId = yield* allocateCommandId(input);
573 return yield* dispatch({
574 type: "thread.pin",
575 commandId,
576 threadId: input.threadId,
577 ...(input.orderKey === undefined ? {} : { orderKey: input.orderKey }),
578 });
579});
580
581export interface SetThreadAutoSettleInput extends ThreadCommandInput {
582 readonly enabled: boolean;
583}
584export const setThreadAutoSettle = Effect.fn("EnvironmentCommands.setThreadAutoSettle")(function* (
585 input: SetThreadAutoSettleInput,
586) {
587 return yield* dispatch({
588 type: "thread.auto-settle.set",
589 commandId: yield* allocateCommandId(input),
590 threadId: input.threadId,
591 enabled: input.enabled,
592 });
593});
594
595export const reorderPinnedThread = Effect.fn("EnvironmentCommands.reorderPinnedThread")(function* (
596 input: ReorderPinnedThreadInput,
597) {
598 const commandId = yield* allocateCommandId(input);
599 return yield* dispatch({
600 type: "thread.pin.reorder",
601 commandId,
602 threadId: input.threadId,
603 orderKey: input.orderKey,
604 });
605});
606
607export const reorderActiveThread = Effect.fn("EnvironmentCommands.reorderActiveThread")(function* (
608 input: ReorderActiveThreadInput,
609) {
610 return yield* dispatch({
611 type: "thread.active.reorder",
612 commandId: yield* allocateCommandId(input),
613 threadId: input.threadId,
614 orderKey: input.orderKey,
615 });
616});
617
618export const unpinThread = Effect.fn("EnvironmentCommands.unpinThread")(function* (
619 input: UnpinThreadInput,
620) {
621 return yield* simpleThreadCommand("thread.unpin", input);
622});
623
624export const unsettleThread = Effect.fn("EnvironmentCommands.unsettleThread")(function* (
625 input: UnsettleThreadInput,
626) {
627 return yield* dispatch({
628 type: "thread.unsettle",
629 commandId: yield* allocateCommandId(input),
630 threadId: input.threadId,
631 reason: input.reason,
632 });
633});
634
635export const snoozeThread = Effect.fn("EnvironmentCommands.snoozeThread")(function* (
636 input: SnoozeThreadInput,
637) {
638 return yield* dispatch({
639 type: "thread.snooze",
640 commandId: yield* allocateCommandId(input),
641 threadId: input.threadId,
642 snoozedUntil: input.snoozedUntil,
643 });
644});
645
646export const unsnoozeThread = Effect.fn("EnvironmentCommands.unsnoozeThread")(function* (
647 input: UnsnoozeThreadInput,
648) {
649 return yield* dispatch({
650 type: "thread.unsnooze",
651 commandId: yield* allocateCommandId(input),
652 threadId: input.threadId,
653 reason: input.reason,
654 });
655});
656
657export const visitThread = Effect.fn("EnvironmentCommands.visitThread")(function* (
658 input: VisitThreadInput,
659) {
660 return yield* dispatch({
661 type: "thread.visit",
662 commandId: yield* allocateCommandId(input),
663 threadId: input.threadId,
664 visitedAt: input.visitedAt,
665 ...(input.opened ? { opened: true } : {}),
666 ...(input.searchQuery ? { searchQuery: input.searchQuery } : {}),
667 });
668});
669
670export const markThreadUnread = Effect.fn("EnvironmentCommands.markThreadUnread")(function* (
671 input: MarkThreadUnreadInput,
672) {
673 return yield* dispatch({
674 type: "thread.mark-unread",
675 commandId: yield* allocateCommandId(input),
676 threadId: input.threadId,
677 });
678});
679
680export const updateThreadMetadata = Effect.fn("EnvironmentCommands.updateThreadMetadata")(
681 function* (input: UpdateThreadMetadataInput) {
682 const commandId = yield* allocateCommandId(input);
683 let result = null;
684 if (
685 input.title !== undefined ||
686 input.branch !== undefined ||
687 input.worktreePath !== undefined ||
688 input.regenerateTitle !== undefined ||
689 input.linkedPullRequest !== undefined ||
690 input.limitRecovery !== undefined
691 ) {
692 result = yield* dispatch({
693 type: "thread.metadata.update",
694 ...(input.limitRecovery === undefined ? {} : { limitRecovery: input.limitRecovery }),
695 commandId,
696 threadId: input.threadId,
697 ...(input.title === undefined ? {} : { title: input.title }),
698 ...(input.branch === undefined ? {} : { branch: input.branch }),
699 ...(input.worktreePath === undefined ? {} : { worktreePath: input.worktreePath }),
700 ...(input.expectedEmpty === true ? { expectedEmpty: true } : {}),
701 ...(input.regenerateTitle === undefined ? {} : { regenerateTitle: input.regenerateTitle }),
702 ...(input.linkedPullRequest === undefined
703 ? {}
704 : { linkedPullRequest: input.linkedPullRequest }),
705 });
706 }
707 if (input.modelSelection !== undefined) {
708 result = yield* dispatch({
709 type: "thread.model-selection.set",
710 commandId: result === null ? commandId : CommandId.make(`${commandId}:model-selection`),
711 threadId: input.threadId,
712 modelSelection: input.modelSelection,
713 });
714 }
715 return result ?? { sequence: 0 };
716 },
717);
718
719export const setThreadRuntimeMode = Effect.fn("EnvironmentCommands.setThreadRuntimeMode")(
720 function* (input: SetThreadRuntimeModeInput) {
721 return yield* dispatch({
722 type: "thread.runtime-mode.set",
723 commandId: yield* allocateCommandId(input),
724 threadId: input.threadId,
725 runtimeMode: input.runtimeMode,
726 });
727 },
728);
729
730export const setThreadInteractionMode = Effect.fn("EnvironmentCommands.setThreadInteractionMode")(
731 function* (input: SetThreadInteractionModeInput) {
732 return yield* dispatch({
733 type: "thread.interaction-mode.set",
734 commandId: yield* allocateCommandId(input),
735 threadId: input.threadId,
736 interactionMode: input.interactionMode,
737 });
738 },
739);
740
741export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")(function* (
742 input: StartThreadTurnInput,
743) {
744 const commandId = yield* allocateCommandId(input);
745 const attachments = yield* persistAttachments(
746 input.threadId,
747 input.message.messageId,
748 input.message.attachments,
749 );
750 const context = remapComposerContextAttachments(
751 input.message.context,
752 input.message.attachments,
753 attachments,
754 );
755 const bootstrap = input.bootstrap?.createThread;
756 const prepareWorktree = input.bootstrap?.prepareWorktree;
757 if (bootstrap !== undefined || prepareWorktree !== undefined) {
758 const existingProjection =
759 bootstrap === undefined ? yield* getProjection(input.threadId) : null;
760 const thread = bootstrap ?? existingProjection!.thread;
761 const workspaceStrategy =
762 prepareWorktree !== undefined
763 ? {
764 type: "worktree" as const,
765 baseRef: prepareWorktree.baseBranch,
766 ...(prepareWorktree.branch === undefined ? {} : { branch: prepareWorktree.branch }),
767 ...(prepareWorktree.startFromOrigin === undefined
768 ? {}
769 : { startFromOrigin: prepareWorktree.startFromOrigin }),
770 }
771 : bootstrap?.worktreePath
772 ? {
773 type: "existing_worktree" as const,
774 worktreePath: bootstrap.worktreePath,
775 ...(bootstrap.branch === null ? {} : { branch: bootstrap.branch }),
776 }
777 : {
778 type: "root" as const,
779 ...(bootstrap?.branch === null || bootstrap?.branch === undefined
780 ? {}
781 : { branch: bootstrap.branch }),
782 };
783 return yield* request(ORCHESTRATION_V2_WS_METHODS.launchThread, {
784 commandId,
785 creationSource: input.creationSource ?? "web",
786 threadId: input.threadId,
787 ...(bootstrap === undefined ? { reuseExistingThread: true } : {}),
788 projectId: thread.projectId,
789 title: input.titleSeed ?? thread.title,
790 generateTitle: input.titleSeed !== undefined,
791 modelSelection: input.modelSelection ?? thread.modelSelection,
792 runtimeMode: input.runtimeMode,
793 interactionMode: input.interactionMode,
794 workspaceStrategy,
795 initialMessage: {
796 messageId: input.message.messageId,
797 text: input.message.text,
798 ...(context ? { context } : {}),
799 attachments,
800 },
801 });
802 }
803
804 const requestedMode = input.dispatchMode ?? "auto";
805 const resendAfterSocketLoss = input.resendAfterSocketLoss ?? true;
806 if (requestedMode === "start") {
807 return yield* dispatchMessage(
808 {
809 type: "message.dispatch",
810 commandId,
811 createdBy: "user",
812 creationSource: input.creationSource ?? "web",
813 threadId: input.threadId,
814 messageId: input.message.messageId,
815 ...(input.manualContinuationOfRunId === undefined
816 ? {}
817 : { manualContinuationOfRunId: input.manualContinuationOfRunId }),
818 text: input.message.text,
819 ...(context ? { context } : {}),
820 ...(input.message.idleCompaction === undefined
821 ? {}
822 : { idleCompaction: input.message.idleCompaction }),
823 attachments,
824 ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }),
825 ...(input.sourceProposedPlan === undefined
826 ? {}
827 : { sourcePlanRef: input.sourceProposedPlan }),
828 ...(input.freshContext ? { freshContext: true } : {}),
829 dispatchMode: { type: "start_immediately" },
830 },
831 resendAfterSocketLoss,
832 );
833 }
834
835 const dispatchMode =
836 requestedMode === "queue"
837 ? ({ type: "queue_after_active" } as const)
838 : ({ type: "start_immediately" } as const);
839 return yield* dispatchMessage(
840 {
841 type: "message.dispatch",
842 commandId,
843 createdBy: "user",
844 creationSource: input.creationSource ?? "web",
845 threadId: input.threadId,
846 messageId: input.message.messageId,
847 text: input.message.text,
848 ...(context ? { context } : {}),
849 ...(input.message.idleCompaction === undefined
850 ? {}
851 : { idleCompaction: input.message.idleCompaction }),
852 attachments,
853 ...(input.titleSeed === undefined ? {} : { titleSeed: input.titleSeed }),
854 ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }),
855 ...(input.sourceProposedPlan === undefined
856 ? {}
857 : { sourcePlanRef: input.sourceProposedPlan }),
858 ...(input.freshContext ? { freshContext: true } : {}),
859 ...(requestedMode !== "queue" ? { deliveryIntent: requestedMode } : {}),
860 dispatchMode,
861 },
862 resendAfterSocketLoss,
863 );
864});
865
866export const interruptThreadTurn = Effect.fn("EnvironmentCommands.interruptThreadTurn")(function* (
867 input: InterruptThreadTurnInput,
868) {
869 let threadId = input.threadId;
870 if (input.subagentId !== undefined) {
871 const projection = yield* getProjection(input.threadId);
872 const subagent = projection.subagents.find((agent) => agent.id === input.subagentId);
873 if (subagent?.origin === "app_owned" && subagent.childThreadId !== null) {
874 // The parent ends the task even when its child lost track of it. Older
875 // servers reject this for app-owned tasks, so the child stop below still runs.
876 if (subagent.runId !== null) {
877 yield* dispatch({
878 type: "subagent.stop",
879 commandId: yield* allocateCommandId({}),
880 threadId: input.threadId,
881 runId: subagent.runId,
882 subagentId: subagent.id,
883 }).pipe(Effect.ignore);
884 }
885 threadId = subagent.childThreadId;
886 } else {
887 if (subagent?.runId == null) return { sequence: 0 };
888 return yield* dispatch({
889 type: "subagent.stop",
890 commandId: yield* allocateCommandId(input),
891 threadId: input.threadId,
892 runId: subagent.runId,
893 subagentId: subagent.id,
894 });
895 }
896 }
897 // A run named by the caller belongs to the parent, not the rerouted child.
898 let runId =
899 threadId === input.threadId ? (input.runId ?? (input.turnId as RunId | undefined)) : undefined;
900 if (runId === undefined) {
901 const config = yield* getInitialServerConfig();
902 if (config.environment.capabilities.threadStop === true) {
903 // A paginated projection can omit commands still running from earlier turns.
904 return yield* dispatch({
905 type: "thread.stop",
906 commandId: yield* allocateCommandId(input),
907 threadId,
908 });
909 }
910 }
911 if (runId === undefined) {
912 const projection = yield* getProjection(threadId);
913 runId = projection.runs.findLast(
914 (run) =>
915 run.status === "preparing" ||
916 run.status === "starting" ||
917 run.status === "running" ||
918 run.status === "waiting",
919 )?.id;
920 if (runId === undefined) {
921 const latestRun = projection.runs.at(-1);
922 if (
923 derivePendingBackgroundWork({
924 latestRun,
925 providerThreads: projection.providerThreads,
926 turnItems: projection.turnItems,
927 activeProviderThreadId: projection.thread.activeProviderThreadId,
928 runs: projection.runs,
929 }).length > 0
930 ) {
931 runId = latestRun?.id;
932 }
933 }
934 if (runId === undefined) {
935 let result = { sequence: 0 };
936 for (const link of visibleThreadPullRequests(projection.thread.pullRequests ?? [])) {
937 if (link.watch === undefined) continue;
938 result = yield* dispatch({
939 type: "thread.pull-request.watch",
940 commandId: yield* allocateCommandId(result.sequence === 0 ? input : {}),
941 threadId,
942 host: link.host,
943 repository: link.repository,
944 number: link.number,
945 watching: false,
946 });
947 }
948 return result;
949 }
950 }
951 if (runId === undefined) return { sequence: 0 };
952 return yield* dispatch({
953 type: "run.interrupt",
954 commandId: yield* allocateCommandId(input),
955 threadId,
956 runId,
957 holdQueue: true,
958 });
959});
960
961export const respondToThreadApproval = Effect.fn("EnvironmentCommands.respondToThreadApproval")(
962 function* (input: RespondToThreadApprovalInput) {
963 return yield* dispatch({
964 type: "runtime-request.respond",
965 commandId: yield* allocateCommandId(input),
966 threadId: input.threadId,
967 requestId: input.requestId,
968 decision: input.decision,
969 ...(input.feedback === undefined ? {} : { feedback: input.feedback }),
970 });
971 },
972);
973
974export const respondToThreadUserInput = Effect.fn("EnvironmentCommands.respondToThreadUserInput")(
975 function* (input: RespondToThreadUserInputInput) {
976 return yield* dispatch({
977 type: "runtime-request.respond",
978 commandId: yield* allocateCommandId(input),
979 threadId: input.threadId,
980 requestId: input.requestId,
981 answers: input.answers,
982 ...(input.attachmentsByQuestionId === undefined
983 ? {}
984 : { attachmentsByQuestionId: input.attachmentsByQuestionId }),
985 });
986 },
987);
988
989export const dismissThreadUserInput = Effect.fn("EnvironmentCommands.dismissThreadUserInput")(
990 function* (input: DismissThreadUserInputInput) {
991 return yield* dispatch({
992 type: "thread.user-input.dismiss",
993 commandId: yield* allocateCommandId(input),
994 threadId: input.threadId,
995 requestId: input.requestId,
996 });
997 },
998);
999
1000export const revertThreadCheckpoint = Effect.fn("EnvironmentCommands.revertThreadCheckpoint")(
1001 function* (input: RevertThreadCheckpointInput) {
1002 if (input.checkpointId !== undefined && input.scopeId !== undefined) {
1003 return yield* dispatch({
1004 type: "checkpoint.rollback",
1005 ...(input.restoreFiles === undefined ? {} : { restoreFiles: input.restoreFiles }),
1006 commandId: yield* allocateCommandId(input),
1007 threadId: input.threadId,
1008 scopeId: CheckpointScopeId.make(input.scopeId),
1009 checkpointId: CheckpointId.make(input.checkpointId),
1010 });
1011 }
1012 const projection = yield* getProjection(input.threadId);
1013 const checkpoint =
1014 projection.checkpoints.find(
1015 (candidate) => candidate.id === input.checkpointId && candidate.scopeId === input.scopeId,
1016 ) ??
1017 projection.checkpoints.findLast((candidate) =>
1018 input.turnCount === 0
1019 ? candidate.ordinalWithinScope === 0 && candidate.appRunOrdinal === null
1020 : candidate.appRunOrdinal === input.turnCount,
1021 );
1022 if (
1023 checkpoint === undefined ||
1024 // Missing snapshots support conversation rewind only.
1025 (checkpoint.status !== "ready" &&
1026 !(input.restoreFiles === false && checkpoint.status === "missing"))
1027 ) {
1028 const target =
1029 input.checkpointId === undefined
1030 ? `run ordinal ${input.turnCount ?? "unknown"}`
1031 : `checkpoint ${input.checkpointId}`;
1032 return yield* new OrchestrationV2CheckpointUnavailableError({
1033 threadId: input.threadId,
1034 target,
1035 });
1036 }
1037 return yield* dispatch({
1038 type: "checkpoint.rollback",
1039 ...(input.restoreFiles === undefined ? {} : { restoreFiles: input.restoreFiles }),
1040 commandId: yield* allocateCommandId(input),
1041 threadId: input.threadId,
1042 scopeId: checkpoint.scopeId,
1043 checkpointId: checkpoint.id,
1044 });
1045 },
1046);
1047
1048export const stopThreadSession = Effect.fn("EnvironmentCommands.stopThreadSession")(function* (
1049 input: StopThreadSessionInput,
1050) {
1051 const projection = yield* getProjection(input.threadId);
1052 const commandId = yield* allocateCommandId(input);
1053 let result = { sequence: 0 };
1054 for (const session of projection.providerSessions) {
1055 result = yield* dispatch({
1056 type: "provider-session.detach",
1057 commandId: CommandId.make(`${commandId}:detach:${session.id}`),
1058 threadId: input.threadId,
1059 providerSessionId: session.id,
1060 reason: "client-requested",
1061 });
1062 }
1063 return result;
1064});
1065
1066export const forkThreadFromRun = Effect.fn("EnvironmentCommands.forkThreadFromRun")(function* (
1067 input: ForkThreadFromRunInput,
1068) {
1069 return yield* dispatch({
1070 type: "thread.fork",
1071 commandId: yield* allocateCommandId(input),
1072 createdBy: "user",
1073 creationSource: input.creationSource ?? "web",
1074 sourceThreadId: input.sourceThreadId,
1075 targetThreadId: input.targetThreadId,
1076 sourcePoint: { type: "run", runId: input.runId },
1077 ...(input.title === undefined ? {} : { title: input.title }),
1078 });
1079});
1080
1081/** Forks the thread's latest stable point into a hidden side chat; works while it runs. */
1082export const forkSideChat = Effect.fn("EnvironmentCommands.forkSideChat")(function* (
1083 input: ForkSideChatInput,
1084) {
1085 return yield* dispatch({
1086 type: "thread.fork",
1087 commandId: yield* allocateCommandId(input),
1088 createdBy: "user",
1089 creationSource: input.creationSource ?? "web",
1090 sourceThreadId: input.sourceThreadId,
1091 targetThreadId: input.targetThreadId,
1092 sourcePoint: { type: "latest_stable" },
1093 sideChat: true,
1094 ...(input.title === undefined ? {} : { title: input.title }),
1095 });
1096});
1097
1098export const mergeThreadBack = Effect.fn("EnvironmentCommands.mergeThreadBack")(function* (
1099 input: MergeThreadBackInput,
1100) {
1101 return yield* dispatch({
1102 type: "thread.merge_back",
1103 commandId: yield* allocateCommandId(input),
1104 createdBy: "user",
1105 creationSource: input.creationSource ?? "web",
1106 sourceThreadId: input.sourceThreadId,
1107 targetThreadId: input.targetThreadId,
1108 sourcePoint: { type: "run", runId: input.runId },
1109 });
1110});
1111
1112export const resumeThreadQueue = Effect.fn("EnvironmentCommands.resumeThreadQueue")(function* (
1113 input: ResumeThreadQueueInput,
1114) {
1115 return yield* dispatch({
1116 ...(input.runId === undefined
1117 ? { type: "queue.resume" as const }
1118 : { type: "queued-run.start" as const, runId: input.runId }),
1119 commandId: yield* allocateCommandId(input),
1120 threadId: input.threadId,
1121 });
1122});
1123
1124export const promoteQueuedRun = Effect.fn("EnvironmentCommands.promoteQueuedRun")(function* (
1125 input: PromoteQueuedRunInput,
1126) {
1127 return yield* dispatch({
1128 type: "queued-message.promote-to-steer",
1129 commandId: yield* allocateCommandId(input),
1130 threadId: input.threadId,
1131 queuedRunId: input.queuedRunId,
1132 targetRunId: input.targetRunId,
1133 });
1134});
1135
1136export const cancelQueuedRun = Effect.fn("EnvironmentCommands.cancelQueuedRun")(function* (
1137 input: CancelQueuedRunInput,
1138) {
1139 return yield* dispatch({
1140 type: "queued-run.cancel",
1141 commandId: yield* allocateCommandId(input),
1142 threadId: input.threadId,
1143 runId: input.runId,
1144 });
1145});
1146
1147export const retryWorkspacePreparation = Effect.fn("EnvironmentCommands.retryWorkspacePreparation")(
1148 function* (input: RetryWorkspacePreparationInput) {
1149 return yield* dispatch({
1150 type: "prepared-run.retry",
1151 commandId: yield* allocateCommandId(input),
1152 threadId: input.threadId,
1153 runId: input.runId,
1154 });
1155 },
1156);
1157
1158export const editQueuedRun = Effect.fn("EnvironmentCommands.editQueuedRun")(function* (
1159 input: EditQueuedRunInput,
1160) {
1161 const attachments =
1162 input.edit === undefined
1163 ? undefined
1164 : yield* persistAttachments(input.threadId, input.edit.messageId, input.edit.attachments);
1165 return yield* dispatch({
1166 type: "queued-run.edit",
1167 commandId: yield* allocateCommandId(input),
1168 threadId: input.threadId,
1169 runId: input.runId,
1170 text: input.text,
1171 ...(attachments === undefined ? {} : { attachments }),
1172 ...(input.edit?.context && attachments
1173 ? {
1174 context: remapComposerContextAttachments(
1175 input.edit.context,
1176 input.edit.attachments,
1177 attachments,
1178 ),
1179 }
1180 : {}),
1181 });
1182});
1183
1184export type LinkThreadPullRequestInput = Omit<
1185 Extract<OrchestrationV2Command, { type: "thread.pull-request.link" }>,
1186 "type" | "commandId"
1187> &
1188 CommandMetadata;
1189export type UnlinkThreadPullRequestInput = Omit<
1190 Extract<OrchestrationV2Command, { type: "thread.pull-request.unlink" }>,
1191 "type" | "commandId"
1192> &
1193 CommandMetadata;
1194export const linkThreadPullRequest = Effect.fn("EnvironmentCommands.linkThreadPullRequest")(
1195 function* (input: LinkThreadPullRequestInput) {
1196 return yield* dispatch({
1197 ...input,
1198 type: "thread.pull-request.link",
1199 commandId: yield* allocateCommandId(input),
1200 });
1201 },
1202);
1203export type WatchThreadPullRequestInput = Omit<
1204 Extract<OrchestrationV2Command, { type: "thread.pull-request.watch" }>,
1205 "type" | "commandId"
1206> &
1207 CommandMetadata;
1208export const watchThreadPullRequest = Effect.fn("EnvironmentCommands.watchThreadPullRequest")(
1209 function* (input: WatchThreadPullRequestInput) {
1210 return yield* dispatch({
1211 ...input,
1212 type: "thread.pull-request.watch",
1213 commandId: yield* allocateCommandId(input),
1214 });
1215 },
1216);
1217export const unlinkThreadPullRequest = Effect.fn("EnvironmentCommands.unlinkThreadPullRequest")(
1218 function* (input: UnlinkThreadPullRequestInput) {
1219 return yield* dispatch({
1220 ...input,
1221 type: "thread.pull-request.unlink",
1222 commandId: yield* allocateCommandId(input),
1223 });
1224 },
1225);