diff --git a/apps/web/src/components/ChatView.logic.test.ts b/apps/web/src/components/ChatView.logic.test.ts index 8aa4df4c09ba5..b9609519d9d08 100644 --- a/apps/web/src/components/ChatView.logic.test.ts +++ b/apps/web/src/components/ChatView.logic.test.ts @@ -61,6 +61,10 @@ import { deriveLockedProvider, hasServerAcknowledgedLocalDispatch, waitForRevertedMessage, + composerDraftContentUnchanged, + queueComposerSubmitBehind, + resolveQueuedComposerSubmit, + settleQueuedComposerSubmit, } from "./ChatView.logic"; const environmentId = EnvironmentId.make("environment-local"); @@ -1129,6 +1133,127 @@ describe("snapshotComposerSend", () => { }); }); +describe("composerDraftContentUnchanged", () => { + const image = { + id: "img-1", + type: "image" as const, + name: "shot.png", + mimeType: "image/png", + sizeBytes: 1, + previewUrl: "blob:one", + file: new File([], "shot.png"), + }; + const submitted = { + prompt: "and then the tests", + images: [image], + files: [], + nonPersistedImageIds: [], + persistedAttachments: [], + terminalContexts: [], + previewAnnotations: [], + reviewComments: [], + threadContexts: [], + environmentContexts: [], + replyTo: null, + modelSelectionByProvider: {}, + activeProvider: null, + runtimeMode: null, + interactionMode: null, + }; + + it("ignores attachment bookkeeping and model changes the user did not make to the message", () => { + expect( + composerDraftContentUnchanged(submitted, { + ...submitted, + images: [{ ...image, previewUrl: "blob:two" }], + persistedAttachments: [{ ...image, dataUrl: "data:" }], + activeProvider: ProviderInstanceId.make("codex"), + }), + ).toBe(true); + }); + + it("treats an edited prompt, attachment set, or cleared draft as a different message", () => { + expect(composerDraftContentUnchanged(submitted, { ...submitted, prompt: "and then" })).toBe( + false, + ); + expect(composerDraftContentUnchanged(submitted, { ...submitted, images: [] })).toBe(false); + expect(composerDraftContentUnchanged(submitted, { ...submitted, terminalContexts: [] })).toBe( + false, + ); + expect(composerDraftContentUnchanged(submitted, null)).toBe(false); + }); +}); + +describe("settleQueuedComposerSubmit", () => { + it("only lets the send it waits behind, on its own thread, release or cancel it", () => { + // A started on thread A (generation 1); the user moved to B, started B (generation 2), and + // pressed Enter again while B was sending. + const waiting = { routeThreadKey: "thread-b", afterSend: 2, ready: false }; + const settle = (sendRouteThreadKey: string, sendGeneration: number | null, started: boolean) => + settleQueuedComposerSubmit({ waiting, sendRouteThreadKey, sendGeneration, started }); + expect(settle("thread-a", 1, true)).toBe("unrelated"); + expect(settle("thread-b", null, false)).toBe("unrelated"); + expect(settle("thread-b", 2, false)).toBe("cancel"); + expect(settle("thread-b", 2, true)).toBe("ready"); + }); +}); + +describe("queueComposerSubmitBehind", () => { + it("waits behind the send running on its own thread, not the one started last", () => { + expect( + queueComposerSubmitBehind({ + routeThreadKey: "thread-x", + sendInFlight: true, + runningSends: new Map([ + ["thread-x", 1], + ["thread-y", 2], + ]), + queued: null, + }), + ).toEqual({ afterSend: 1, ready: false }); + }); + + it("lets a new Enter send once no send on this thread can settle the queued one", () => { + const stale = { afterSend: 2, ready: false }; + expect( + queueComposerSubmitBehind({ + routeThreadKey: "thread-x", + sendInFlight: false, + runningSends: new Map(), + queued: stale, + }), + ).toBeNull(); + expect( + queueComposerSubmitBehind({ + routeThreadKey: "thread-x", + sendInFlight: true, + runningSends: new Map([["thread-y", 2]]), + queued: stale, + }), + ).toBeNull(); + }); +}); + +describe("resolveQueuedComposerSubmit", () => { + it("sends only the unedited message on its own thread, and never as an answer", () => { + const ready = { + ready: true, + leftThread: false, + sendBlocked: false, + pendingRequest: false, + draftUnchanged: true, + }; + expect(resolveQueuedComposerSubmit(ready)).toBe("send"); + expect(resolveQueuedComposerSubmit({ ...ready, sendBlocked: true })).toBe("wait"); + expect(resolveQueuedComposerSubmit({ ...ready, pendingRequest: true })).toBe("cancel"); + expect(resolveQueuedComposerSubmit({ ...ready, ready: false, pendingRequest: true })).toBe( + "cancel", + ); + expect(resolveQueuedComposerSubmit({ ...ready, draftUnchanged: false })).toBe("cancel"); + expect(resolveQueuedComposerSubmit({ ...ready, ready: false, leftThread: true })).toBe("drop"); + }); +}); + describe("planComposerTurn", () => { const base = { messageText: "hello", diff --git a/apps/web/src/components/ChatView.logic.ts b/apps/web/src/components/ChatView.logic.ts index 92a456f12d308..a1611943a68de 100644 --- a/apps/web/src/components/ChatView.logic.ts +++ b/apps/web/src/components/ChatView.logic.ts @@ -69,6 +69,7 @@ import { import { type ComposerFileAttachment, type ComposerImageAttachment, + type ComposerThreadDraftState, type DraftId, type DraftThreadState, } from "../composerDraftStore"; @@ -1260,6 +1261,108 @@ export interface ComposerSendContents { readonly replyTo: ResponseContextRecord | null; } +/** + * Whether a draft still holds the message the user submitted. Only its content + * counts: the model picker and attachment upload bookkeeping can change without + * the user editing the message. + */ +export function composerDraftContentUnchanged( + submitted: ComposerThreadDraftState | null, + current: ComposerThreadDraftState | null, +): boolean { + if (submitted === null || current === null) return submitted === current; + const ids = (attachments: ReadonlyArray<{ readonly id: string }>) => + attachments.map((attachment) => attachment.id).join("\n"); + return ( + submitted.prompt === current.prompt && + ids(submitted.images) === ids(current.images) && + ids(submitted.files) === ids(current.files) && + submitted.terminalContexts === current.terminalContexts && + submitted.previewAnnotations === current.previewAnnotations && + submitted.reviewComments === current.reviewComments && + submitted.threadContexts === current.threadContexts && + submitted.environmentContexts === current.environmentContexts && + submitted.replyTo === current.replyTo + ); +} + +/** A draft the user pressed Enter on while their previous message was still sending. */ +export interface QueuedComposerSubmit { + readonly routeThreadKey: string; + readonly draft: ComposerThreadDraftState | null; + readonly dispatchMode: Mode; + readonly submissionIntent: Intent; + /** The send generation it waits behind. */ + readonly afterSend: number; + readonly ready: boolean; +} + +/** + * Where Enter goes while a message may still be sending: null sends now, otherwise the + * submit waits behind the send running on this thread. `runningSends` holds each thread's + * running send generation, since one composer sends on several threads as the user moves. + * A queued submit no send on this thread will settle is stale and gives way to the new Enter. + */ +export function queueComposerSubmitBehind(input: { + readonly routeThreadKey: string; + /** Whether this thread's send is still in flight, rather than finishing up. */ + readonly sendInFlight: boolean; + readonly runningSends: ReadonlyMap; + readonly queued: { readonly afterSend: number; readonly ready: boolean } | null; +}): { readonly afterSend: number; readonly ready: boolean } | null { + const { queued } = input; + if (queued?.ready) return queued; + const running = input.sendInFlight ? input.runningSends.get(input.routeThreadKey) : undefined; + return running === undefined ? null : { afterSend: running, ready: false }; +} + +/** + * What a finished send does to the submit waiting behind it. Only the send it waits behind, + * on the same thread, settles it: another thread's send finishing says nothing about this one. + */ +export function settleQueuedComposerSubmit(input: { + readonly waiting: { + readonly routeThreadKey: string; + readonly afterSend: number; + readonly ready: boolean; + } | null; + readonly sendRouteThreadKey: string; + /** The generation the finished send claimed, or null if it stopped before claiming one. */ + readonly sendGeneration: number | null; + readonly started: boolean; +}): "unrelated" | "ready" | "cancel" { + const { waiting, sendGeneration } = input; + if ( + sendGeneration === null || + waiting === null || + waiting.ready || + waiting.routeThreadKey !== input.sendRouteThreadKey || + waiting.afterSend !== sendGeneration + ) { + return "unrelated"; + } + return input.started ? "ready" : "cancel"; +} + +/** + * Whether a ready queued submit sends now. It is the user's earlier Enter, so it only sends + * as the plain message they wrote: an edit, or an approval or question that arrived meanwhile + * (which would turn it into an answer they never saw), cancels it instead. + */ +export function resolveQueuedComposerSubmit(input: { + readonly ready: boolean; + readonly leftThread: boolean; + readonly sendBlocked: boolean; + readonly pendingRequest: boolean; + readonly draftUnchanged: boolean; +}): "wait" | "send" | "drop" | "cancel" { + if (input.leftThread) return "drop"; + // Cancel as soon as a request shows up: another client may answer it before this one is ready. + if (input.pendingRequest) return "cancel"; + if (!input.ready || input.sendBlocked) return "wait"; + return input.draftUnchanged ? "send" : "cancel"; +} + /** * Copies the composer contents a send dispatches, so a failed send can put the * same draft back. Terminal contexts that lost their text are dropped from both diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts index 7257dee5ced9e..495cd84def8eb 100644 --- a/apps/web/src/components/chat/useComposerSend.ts +++ b/apps/web/src/components/chat/useComposerSend.ts @@ -14,6 +14,11 @@ import { snapshotComposerSend, withMessageContext, makeThreadErrorSetter, + composerDraftContentUnchanged, + resolveQueuedComposerSubmit, + queueComposerSubmitBehind, + settleQueuedComposerSubmit, + type QueuedComposerSubmit, } from "../ChatView.logic"; import { isUsageLimitsCommand } from "@t3tools/shared/usageLimits"; import { @@ -27,7 +32,10 @@ import { ProviderInteractionMode, RuntimeMode, } from "@t3tools/contracts"; -import { deriveThreadTitleSeed } from "@t3tools/client-runtime/operations"; +import { + deriveThreadTitleSeed, + ThreadLaunchUnansweredError, +} from "@t3tools/client-runtime/operations"; import { wasBootstrapThreadDeleted } from "@t3tools/client-runtime/errors"; import { declinePendingApprovals, @@ -44,8 +52,9 @@ import { scopeThreadRef, } from "@t3tools/client-runtime/environment"; import { createModelSelection } from "@t3tools/shared/model"; -import { type SetStateAction, useLayoutEffect, useRef } from "react"; +import { type SetStateAction, useEffect, useLayoutEffect, useRef, useState } from "react"; import { flushSync } from "react-dom"; +import * as Schema from "effect/Schema"; import { useNavigate } from "@tanstack/react-router"; import { isAtomCommandInterrupted, @@ -93,7 +102,7 @@ import { responseContextReference, } from "../../lib/composerContextRecords"; import { environmentServerConfigsAtom } from "../../state/server"; -import { threadEnvironment } from "../../state/threads"; +import { environmentThreadShells, threadEnvironment } from "../../state/threads"; import { runMobileComposerTransition } from "./draftHeroTransition"; import { useClientSettingsHydrated } from "../../hooks/useSettings"; import { useStableCallback } from "../../hooks/useStableCallback"; @@ -259,6 +268,17 @@ interface ComposerSendInput { readonly linkCheckoutPullRequest: (threadRef: ScopedThreadRef, url: string) => void; } +const isThreadLaunchUnanswered = Schema.is(ThreadLaunchUnansweredError); + +const notifyQueuedSubmitNotSent = () => + toastManager.add( + stackedThreadToast({ + type: "info", + title: "Your next message was not sent", + description: "It is still in the composer. Send it when you're ready.", + }), + ); + /** * Everything the composer's send button can start: a turn on this thread or a new one, * a multiple-model start, a queued or steering message, a plan follow-up or implementation @@ -376,7 +396,8 @@ export function useComposerSend(input: ComposerSendInput) { respondingRequestIdsRef.current = respondingRequestIds; }, [pendingApprovals, respondingRequestIds]); - const onSend = async ( + // Resolves to true once the turn started; a send that stops earlier resolves otherwise. + const sendComposerDraft = async ( e?: { preventDefault: () => void }, dispatchMode: ComposerDispatchMode = "auto", submissionIntent: ComposerSubmissionIntent = "foreground", @@ -873,8 +894,10 @@ export function useComposerSend(input: ComposerSendInput) { multipleTargets.push(resolved); } + // onSend reads the generation this claims right after calling, so no await may precede it. sendInFlightRef.current = true; const sendGeneration = ++composerSendGenerationRef.current; + const draftAtSend = useComposerDraftStore.getState().getComposerDraft(composerDraftTarget); const attachmentCapabilitiesBeforeUpload = readLiveAttachmentCapabilities(); if (attachmentCapabilitiesBeforeUpload.fileBlockReason !== null) { sendInFlightRef.current = false; @@ -1179,9 +1202,20 @@ export function useComposerSend(input: ComposerSendInput) { : null; // Keep new-thread flows from reusing this draft while attachments and worktree setup settle. if (sentDraftThreadRef) markPromotedDraftThreadByRef(sentDraftThreadRef); - promptRef.current = ""; - clearComposerDraftContent(composerDraftTarget); - composerRef.current?.resetCursorState(); + // Uploads and the dock transition above are awaited; anything typed meanwhile is the next + // message, not this one. A direct annotation writes into the draft as it sends, so it + // always clears. + if ( + directAnnotation || + composerDraftContentUnchanged( + draftAtSend, + useComposerDraftStore.getState().getComposerDraft(composerDraftTarget), + ) + ) { + promptRef.current = ""; + clearComposerDraftContent(composerDraftTarget); + composerRef.current?.resetCursorState(); + } const title = deriveThreadTitleSeed({ text: messageTextForSend, @@ -1391,15 +1425,25 @@ export function useComposerSend(input: ComposerSendInput) { } if (failure !== null) { + const failureError = squashAtomCommandFailure(failure); + // An unanswered launch whose thread already arrived keeps the draft promoted to it, and its + // composer content now lives under that thread; one that has not arrived restores the + // draft, which promotes again if the thread lands. + const launchedThreadRef = scopeThreadRef(environmentId, threadIdForSend); + const launchedThreadArrived = + isLocalDraftThread && + isThreadLaunchUnanswered(failureError) && + appAtomRegistry.get(environmentThreadShells.threadShellAtom(launchedThreadRef)) !== null; + const recoveryTarget = launchedThreadArrived ? launchedThreadRef : composerDraftTarget; if (isLocalDraftThread && draftId && draftThread) { - restoreFailedDraftThread( - draftId, - draftThread, - wasBootstrapThreadDeleted(squashAtomCommandFailure(failure)) - ? newThreadId() - : threadIdForSend, - ); - clearBackgroundDraftSubmissionByRef(scopeThreadRef(environmentId, threadIdForSend)); + if (!launchedThreadArrived) { + restoreFailedDraftThread( + draftId, + draftThread, + wasBootstrapThreadDeleted(failureError) ? newThreadId() : threadIdForSend, + ); + } + clearBackgroundDraftSubmissionByRef(launchedThreadRef); } // Only an acknowledged send replaces its optimistic row, so a failed one leaves here. setOptimisticUserMessages((existing) => { @@ -1410,7 +1454,7 @@ export function useComposerSend(input: ComposerSendInput) { optimisticBlobPreviews.release(optimisticMessage); if ( composerDraftHasUserContent( - useComposerDraftStore.getState().getComposerDraft(composerDraftTarget), + useComposerDraftStore.getState().getComposerDraft(recoveryTarget), ) ) { // A draft typed meanwhile keeps the composer, so the failed text stays reachable here. @@ -1436,7 +1480,7 @@ export function useComposerSend(input: ComposerSendInput) { ); } } else { - const retryComposerImages = restoreComposerDraft(composerDraftTarget, sendSnapshot); + const retryComposerImages = restoreComposerDraft(recoveryTarget, sendSnapshot); if (currentRouteThreadKeyRef.current === routeThreadKey) { promptRef.current = sendSnapshot.draftText; composerImagesRef.current = retryComposerImages; @@ -1453,10 +1497,9 @@ export function useComposerSend(input: ComposerSendInput) { } } if (!isAtomCommandInterrupted(failure)) { - const error = squashAtomCommandFailure(failure); setThreadError( threadIdForSend, - error instanceof Error ? error.message : "Failed to send message.", + failureError instanceof Error ? failureError.message : "Failed to send message.", ); if ( draftId && @@ -1469,7 +1512,8 @@ export function useComposerSend(input: ComposerSendInput) { submissionIntent === "background" ? "Background task failed" : "Could not start thread", - description: error instanceof Error ? error.message : "Failed to send message.", + description: + failureError instanceof Error ? failureError.message : "Failed to send message.", actionProps: { children: "Open draft", onClick: () => { @@ -1489,8 +1533,96 @@ export function useComposerSend(input: ComposerSendInput) { ); if (!sendQueuesBehindSetup) resetLocalDispatch(); } + return turnStartSucceeded; + }; + + // Enter while the previous message is still sending keeps the draft and sends it once that + // send starts its turn. A send that did not start one may still land, so it is never retried. + // Finishing sends read the ref from their own closure; the state only re-renders the effect. + type QueuedSubmit = QueuedComposerSubmit; + const queuedSubmitRef = useRef(null); + const [queuedSubmit, setQueuedSubmitState] = useState(null); + // The generation of the send running on each thread. The hook outlives navigation, so a + // submit has to wait behind its own thread's send, not whichever send started last. + const runningSendsRef = useRef(new Map()); + const setQueuedSubmit = (next: QueuedSubmit | null) => { + queuedSubmitRef.current = next; + setQueuedSubmitState(next); }; + const onSend: typeof sendComposerDraft = async ( + e, + dispatchMode = "auto", + submissionIntent = "foreground", + directAnnotation, + ) => { + // Enter on an open question answers it, which a send in flight must not hold back. + if (activePendingProgress && !directAnnotation) { + return sendComposerDraft(e, dispatchMode, submissionIntent); + } + const queued = + queuedSubmitRef.current?.routeThreadKey === routeThreadKey ? queuedSubmitRef.current : null; + const behind = directAnnotation + ? null + : queueComposerSubmitBehind({ + routeThreadKey, + sendInFlight: sendInFlightRef.current, + runningSends: runningSendsRef.current, + queued, + }); + if (behind !== null) { + e?.preventDefault(); + const draft = useComposerDraftStore.getState().getComposerDraft(composerDraftTarget); + setQueuedSubmit( + composerDraftHasUserContent(draft) + ? { + routeThreadKey, + draft, + // Written as a follow-up to the message still sending, so it waits behind that + // turn instead of steering it. + dispatchMode: dispatchMode === "auto" ? "queue" : dispatchMode, + submissionIntent, + ...behind, + } + : null, + ); + return; + } + if (queued !== null) setQueuedSubmit(null); + const generationBefore = composerSendGenerationRef.current; + const sending = sendComposerDraft(e, dispatchMode, submissionIntent, directAnnotation); + // The send claims its generation before its first await. Read it now: once this awaits, + // a send started on another thread from this same composer moves the counter on. + const claimedGeneration = composerSendGenerationRef.current; + const sendGeneration = claimedGeneration === generationBefore ? null : claimedGeneration; + const runningSends = runningSendsRef.current; + if (sendGeneration !== null) runningSends.set(routeThreadKey, sendGeneration); + let started: boolean | undefined = false; + let failure: { readonly error: unknown } | null = null; + try { + started = await sending; + } catch (error) { + failure = { error }; + } + if (sendGeneration !== null && runningSends.get(routeThreadKey) === sendGeneration) { + runningSends.delete(routeThreadKey); + } + const waiting = queuedSubmitRef.current; + const settled = settleQueuedComposerSubmit({ + waiting, + sendRouteThreadKey: routeThreadKey, + sendGeneration, + started: started === true, + }); + if (waiting !== null && settled === "ready") { + setQueuedSubmit({ ...waiting, ready: true }); + } else if (waiting !== null && settled === "cancel") { + setQueuedSubmit(null); + notifyQueuedSubmitNotSent(); + } + if (failure !== null) throw failure.error; + return started; + }; const onSendLatest = useStableCallback(onSend); const onSendWithRouting = async (...args: Parameters) => { args[0]?.preventDefault(); @@ -1505,5 +1637,40 @@ export function useComposerSend(input: ComposerSendInput) { }); if (ready) return onSendLatest(...args); }; + const onSendWithRoutingLatest = useStableCallback(onSendWithRouting); + + const queuedSubmitBlocked = + (isSendBusy && !sendQueuesBehindSetup) || isRevertingCheckpoint || threadDetailLoading; + const pendingRequest = pendingApprovals.length > 0 || Boolean(activePendingProgress); + // oxlint-disable-next-line t3code/no-unreasoned-state-sync -- fires the deferred Enter once server state (send settled, connection, pending requests) lets it; no user action marks that moment + useEffect(() => { + const next = queuedSubmit; + if (next === null) return; + const action = resolveQueuedComposerSubmit({ + ready: next.ready, + leftThread: next.routeThreadKey !== routeThreadKey, + sendBlocked: queuedSubmitBlocked || sendInFlightRef.current, + pendingRequest: pendingRequest || pendingApprovalsRef.current.length > 0, + draftUnchanged: composerDraftContentUnchanged( + next.draft, + useComposerDraftStore.getState().getComposerDraft(composerDraftTarget), + ), + }); + if (action === "wait") return; + queuedSubmitRef.current = null; + setQueuedSubmitState(null); + if (action === "cancel") notifyQueuedSubmitNotSent(); + if (action === "send") + void onSendWithRoutingLatest(undefined, next.dispatchMode, next.submissionIntent); + }, [ + composerDraftTarget, + onSendWithRoutingLatest, + pendingRequest, + queuedSubmit, + queuedSubmitBlocked, + routeThreadKey, + sendInFlightRef, + ]); + return { onSend: onSendWithRouting }; } diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index 87c8b785563b3..b28ebd97ebc27 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -458,6 +458,59 @@ describe("V2 environment commands", () => { }).pipe(Effect.provide(layerTestRandom)), ); + it.effect("fails a first-message launch the host never answers without cancelling it", () => + Effect.gen(function* () { + const supervisor = yield* makeSupervisor({ commands: [], projects: [] }); + const healthySession = Option.getOrThrow(yield* SubscriptionRef.get(supervisor.session)); + let interrupted = false; + const silentClient = { + ...healthySession.client, + [ORCHESTRATION_V2_WS_METHODS.launchThread]: () => + Effect.never.pipe(Effect.onInterrupt(() => Effect.sync(() => (interrupted = true)))), + } as unknown as WsRpcProtocolClient; + yield* SubscriptionRef.set( + supervisor.session, + Option.some({ ...healthySession, client: silentClient }), + ); + + const sendFiber = yield* startThreadTurn({ + commandId: CommandId.make("command-launch-unanswered"), + threadId: v2ThreadId, + message: { + messageId: MessageId.make("message-launch-unanswered"), + role: "user", + text: "hello", + attachments: [], + }, + runtimeMode: "full-access", + interactionMode: "default", + bootstrap: { + createThread: { + projectId: ProjectId.make("project-1"), + title: "Thread", + modelSelection: v2Projection.thread.modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: "2026-06-20T00:00:00.000Z", + }, + }, + }).pipe( + Effect.provideService(EnvironmentSupervisor.EnvironmentSupervisor, supervisor), + Effect.flip, + Effect.forkChild, + ); + yield* TestClock.adjust("119 seconds"); + expect(sendFiber.pollUnsafe()).toBeUndefined(); + yield* TestClock.adjust("1 second"); + + expect((yield* Fiber.join(sendFiber))._tag).toBe("ThreadLaunchUnansweredError"); + // The server commits a launch in several steps, so the late request keeps running. + expect(interrupted).toBe(false); + }).pipe(Effect.provide(layerTestRandom)), + ); + it.effect("holds a lost message through reconnect backoff until disconnect", () => Effect.gen(function* () { const commands: OrchestrationV2Command[] = []; diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index c6f5c8c34166e..330561d5433f2 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -34,13 +34,16 @@ import * as SecureRandom from "effect/crypto/SecureRandom"; import * as Cause from "effect/Cause"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; +import * as Fiber from "effect/Fiber"; import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; import * as Stream from "effect/Stream"; import * as SubscriptionRef from "effect/SubscriptionRef"; import * as EnvironmentSupervisor from "../connection/supervisor.ts"; import { EnvironmentRpcUnavailableError, + type EnvironmentRpcInput, getInitialServerConfig, isRpcClientError, request, @@ -348,6 +351,38 @@ const dispatchMessage = Effect.fn("EnvironmentCommands.dispatchMessage")(functio } }); +export class ThreadLaunchUnansweredError extends Schema.TaggedError()( + "ThreadLaunchUnansweredError", + { message: Schema.String }, +) {} + +// Interrupting a launch would cancel the server handler mid-launch, so the deadline bounds only +// the wait and the launch keeps running; the server may still create the thread. +const THREAD_LAUNCH_TIMEOUT = "2 minutes"; + +const launchThread = Effect.fn("EnvironmentCommands.launchThread")(function* ( + input: EnvironmentRpcInput, +) { + const answer = yield* Effect.forkDetach( + request(ORCHESTRATION_V2_WS_METHODS.launchThread, input), + { + startImmediately: true, + }, + ); + return yield* Fiber.join(answer).pipe( + Effect.timeoutOrElse({ + duration: THREAD_LAUNCH_TIMEOUT, + orElse: () => + Effect.fail( + new ThreadLaunchUnansweredError({ + message: + "The server did not confirm this message. It may still arrive; check the thread before sending it again.", + }), + ), + }), + ); +}); + const getProjection = (threadId: ThreadId) => request(ORCHESTRATION_V2_WS_METHODS.getThreadProjection, { threadId }); @@ -780,7 +815,7 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")( ? {} : { branch: bootstrap.branch }), }; - return yield* request(ORCHESTRATION_V2_WS_METHODS.launchThread, { + return yield* launchThread({ commandId, creationSource: input.creationSource ?? "web", threadId: input.threadId,