Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
28 changes: 23 additions & 5 deletions apps/web/src/components/ChatView.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ import { useThreadLifecycle } from "./chat/useThreadLifecycle";
import { useRightPanelActions } from "./chat/useRightPanelActions";
import { type TerminalLaunchContext, useChatTerminalActions } from "./chat/useChatTerminalActions";
import { useComposerSend } from "./chat/useComposerSend";
import { useComposerResume } from "./chat/useComposerResume";
import { ThreadFind, ThreadFindCanvas, type ThreadFindControls } from "./chat/ThreadFindProvider";
import { THREAD_FIND_BAR_RESERVED_HEIGHT } from "./chat/ThreadFindBar";
import { usageLimitRecoveryBannerItem } from "./chat/UsageLimitRecoveryBanner";
Expand Down Expand Up @@ -2923,7 +2924,28 @@ export default function ChatView(props: ChatViewProps) {
setPendingServerThreadStartFromOriginByThreadId,
scheduleComposerFocus,
});
const { onSend, onResume, onImplementPlanInNewThread, isResuming } = useComposerSend({
const { onResume, isResuming } = useComposerResume({
environmentId,
routeThreadKey,
currentRouteThreadKeyRef,
activeThread,
resumableRunId,
hasHeldQueuedRuns,
isSendBusy,
isConnecting,
isRevertingCheckpoint,
threadDetailLoading,
activeEnvironmentUnavailable,
runtimeMode,
interactionMode,
localCheckoutBranchMismatch,
composerRef,
persistThreadSettingsForNextTurn,
clearUsageLimitsFor,
scrollToEnd,
setThreadError,
});
const { onSend, onImplementPlanInNewThread } = useComposerSend({
environmentId,
routeThreadKey,
currentRouteThreadKeyRef,
Expand All @@ -2939,12 +2961,9 @@ export default function ChatView(props: ChatViewProps) {
isServerThread,
isLocalDraftThread,
isGitRepo,
hasHeldQueuedRuns,
resumableRunId,
phase,
runtimeMode,
defaultRuntimeMode,
interactionMode,
sendEnvMode,
startFromOrigin,
localCheckoutBranchMismatch,
Expand Down Expand Up @@ -3000,7 +3019,6 @@ export default function ChatView(props: ChatViewProps) {
setMultipleModelSelections,
setThreadError,
setKeepFullHistory,
scrollToEnd,
anchorNewTurn,
acknowledgeActiveThreadWoke,
handleInteractionModeChange,
Expand Down
167 changes: 167 additions & 0 deletions apps/web/src/components/chat/useComposerResume.ts
Original file line number Diff line number Diff line change
@@ -0,0 +1,167 @@
import {
type EnvironmentId,
type ProviderInteractionMode,
type RunId,
type RuntimeMode,
} from "@t3tools/contracts";
import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell";
import {
isAtomCommandInterrupted,
squashAtomCommandFailure,
type AtomCommandResult,
} from "@t3tools/client-runtime/state/runtime";
import { useState } from "react";
import { newMessageId } from "~/lib/utils";
import type { ComposerHandleRef } from "../../composerHandleContext";
import { threadEnvironment } from "../../state/threads";
import { useOrchestrationCommand } from "../../state/use-orchestration-command";
import type { makeThreadErrorSetter } from "../ChatView.logic";
import { draftSendState } from "./chatViewSend";
import type { useTimelineFollow } from "./useTimelineFollow";
import type { useUsageLimitsPanel } from "./useUsageLimitsPanel";

interface ComposerResumeInput {
readonly environmentId: EnvironmentId;
readonly routeThreadKey: string;
readonly currentRouteThreadKeyRef: { readonly current: string | null };
readonly activeThread: EnvironmentThreadShell | undefined;
readonly resumableRunId: RunId | null;
readonly hasHeldQueuedRuns: boolean;
readonly isSendBusy: boolean;
readonly isConnecting: boolean;
readonly isRevertingCheckpoint: boolean;
readonly threadDetailLoading: boolean;
readonly activeEnvironmentUnavailable: boolean;
readonly runtimeMode: RuntimeMode;
readonly interactionMode: ProviderInteractionMode;
readonly localCheckoutBranchMismatch: { readonly currentBranch: string } | null;
readonly composerRef: ComposerHandleRef;
readonly persistThreadSettingsForNextTurn: (input: {
threadId: EnvironmentThreadShell["id"];
createdAt: string;
branch?: string;
runtimeMode: RuntimeMode;
interactionMode: ProviderInteractionMode;
}) => Promise<AtomCommandResult<void, unknown>>;
readonly clearUsageLimitsFor: ReturnType<typeof useUsageLimitsPanel>["clearFor"];
readonly scrollToEnd: ReturnType<typeof useTimelineFollow>["scrollToEnd"];
readonly setThreadError: ReturnType<typeof makeThreadErrorSetter>;
}

/**
* Resumes an interrupted run ("Continue where you left off.") and/or releases the turns held
* in its queue. Shares the per-route send lock with the composer's send paths.
*/
export function useComposerResume(input: ComposerResumeInput) {
const {
environmentId,
routeThreadKey,
currentRouteThreadKeyRef,
activeThread,
resumableRunId,
hasHeldQueuedRuns,
isSendBusy,
isConnecting,
isRevertingCheckpoint,
threadDetailLoading,
activeEnvironmentUnavailable,
runtimeMode,
interactionMode,
localCheckoutBranchMismatch,
composerRef,
persistThreadSettingsForNextTurn,
clearUsageLimitsFor,
scrollToEnd,
setThreadError,
} = input;
const startThreadTurn = useOrchestrationCommand(threadEnvironment.startTurn, {
reportFailure: false,
});
const resumeThreadQueue = useOrchestrationCommand(threadEnvironment.resumeThreadQueue, {
reportFailure: false,
});
const [resumingThreadKeys, setResumingThreadKeys] = useState<ReadonlySet<string>>(
() => new Set(),
);
const isResuming = resumingThreadKeys.has(routeThreadKey);
const { sendInFlight: sendInFlightRef } = draftSendState(routeThreadKey);

const onResume = async () => {
if (
!activeThread ||
(resumableRunId === null && !hasHeldQueuedRuns) ||
isSendBusy ||
isResuming ||
isConnecting ||
isRevertingCheckpoint ||
threadDetailLoading ||
activeEnvironmentUnavailable ||
sendInFlightRef.current
) {
return;
}
const threadId = activeThread.id;
sendInFlightRef.current = true;
setResumingThreadKeys((current) => new Set(current).add(routeThreadKey));
setThreadError(threadId, null);
// Commands settle into a result, so this always runs.
const resume = async () => {
if (resumableRunId === null) {
return resumeThreadQueue({ environmentId, input: { threadId } });
}
const createdAt = new Date().toISOString();
const settingsResult = await persistThreadSettingsForNextTurn({
threadId,
createdAt,
...(localCheckoutBranchMismatch
? { branch: localCheckoutBranchMismatch.currentBranch }
: {}),
runtimeMode,
interactionMode,
});
if (settingsResult._tag === "Failure") return settingsResult;
const modelSelection = composerRef.current?.getSendContext().selectedModelSelection;
const turnResult = await startThreadTurn({
environmentId,
input: {
threadId,
manualContinuationOfRunId: resumableRunId,
...(modelSelection ? { modelSelection } : {}),
message: {
messageId: newMessageId(),
role: "user",
text: "Continue where you left off.",
attachments: [],
},
runtimeMode,
interactionMode,
dispatchMode: "start",
},
});
if (turnResult._tag === "Failure" || !hasHeldQueuedRuns) return turnResult;
clearUsageLimitsFor(routeThreadKey);
return resumeThreadQueue({ environmentId, input: { threadId } });
};
const result = await resume();
if (result._tag === "Failure") {
if (!isAtomCommandInterrupted(result)) {
const error = squashAtomCommandFailure(result);
setThreadError(
threadId,
error instanceof Error ? error.message : "Could not resume thread.",
);
}
} else {
clearUsageLimitsFor(routeThreadKey);
if (currentRouteThreadKeyRef.current === routeThreadKey) scrollToEnd();
}
sendInFlightRef.current = false;
setResumingThreadKeys((current) => {
const next = new Set(current);
next.delete(routeThreadKey);
return next;
});
};

return { onResume, isResuming };
}
97 changes: 2 additions & 95 deletions apps/web/src/components/chat/useComposerSend.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,7 +25,6 @@ import {
type ModelSelection,
type PreviewAnnotationPayload,
type ThreadId,
type RunId,
ProviderInteractionMode,
RuntimeMode,
} from "@t3tools/contracts";
Expand All @@ -44,7 +43,7 @@ import {
} from "@t3tools/client-runtime/environment";
import { createModelSelection } from "@t3tools/shared/model";
import { truncate } from "@t3tools/shared/String";
import { type SetStateAction, useLayoutEffect, useRef, useState } from "react";
import { type SetStateAction, useLayoutEffect, useRef } from "react";
import { flushSync } from "react-dom";
import { useNavigate } from "@tanstack/react-router";
import {
Expand Down Expand Up @@ -162,12 +161,9 @@ interface ComposerSendInput {
readonly isServerThread: boolean;
readonly isLocalDraftThread: boolean;
readonly isGitRepo: boolean;
readonly hasHeldQueuedRuns: boolean;
readonly resumableRunId: RunId | null;
readonly phase: SessionPhase;
readonly runtimeMode: RuntimeMode;
readonly defaultRuntimeMode: RuntimeMode;
readonly interactionMode: ProviderInteractionMode;
readonly sendEnvMode: DraftThreadEnvMode;
readonly startFromOrigin: boolean;
readonly localCheckoutBranchMismatch: {
Expand Down Expand Up @@ -230,7 +226,6 @@ interface ComposerSendInput {
) => void;
readonly setThreadError: ReturnType<typeof makeThreadErrorSetter>;
readonly setKeepFullHistory: (threadKey: string, keep: boolean) => void;
readonly scrollToEnd: TimelineFollow["scrollToEnd"];
readonly anchorNewTurn: TimelineFollow["anchorNewTurn"];
readonly acknowledgeActiveThreadWoke: () => void;
readonly handleInteractionModeChange: (mode: ProviderInteractionMode) => void;
Expand Down Expand Up @@ -267,12 +262,9 @@ export function useComposerSend(input: ComposerSendInput) {
isServerThread,
isLocalDraftThread,
isGitRepo,
hasHeldQueuedRuns,
resumableRunId,
phase,
runtimeMode,
defaultRuntimeMode,
interactionMode,
sendEnvMode,
startFromOrigin,
localCheckoutBranchMismatch,
Expand Down Expand Up @@ -328,7 +320,6 @@ export function useComposerSend(input: ComposerSendInput) {
setMultipleModelSelections,
setThreadError,
setKeepFullHistory,
scrollToEnd,
anchorNewTurn,
acknowledgeActiveThreadWoke,
handleInteractionModeChange,
Expand All @@ -342,9 +333,6 @@ export function useComposerSend(input: ComposerSendInput) {
const startThreadTurn = useOrchestrationCommand(threadEnvironment.startTurn, {
reportFailure: false,
});
const resumeThreadQueue = useOrchestrationCommand(threadEnvironment.resumeThreadQueue, {
reportFailure: false,
});
const uploadThreadFeedback = useOrchestrationCommand(threadEnvironment.uploadFeedback, {
reportFailure: false,
});
Expand All @@ -365,10 +353,6 @@ export function useComposerSend(input: ComposerSendInput) {
const readMultipleModelSelections = () => appAtomRegistry.get(fanoutSelectionsAtom);
const { sendInFlight: sendInFlightRef, uncertainSubmissions: uncertainMultipleSubmissionsRef } =
draftSendState(routeThreadKey);
const [resumingThreadKeys, setResumingThreadKeys] = useState<ReadonlySet<string>>(
() => new Set(),
);
const isResuming = resumingThreadKeys.has(routeThreadKey);
const composerSendGenerationRef = useRef(0);
const environmentUnavailableSendToastSlotRef = useRef(0);
// A send awaits uploads before it declines, so it reads the approvals pending by then.
Expand All @@ -379,83 +363,6 @@ export function useComposerSend(input: ComposerSendInput) {
respondingRequestIdsRef.current = respondingRequestIds;
}, [pendingApprovals, respondingRequestIds]);

const onResume = async () => {
if (
!activeThread ||
(resumableRunId === null && !hasHeldQueuedRuns) ||
isSendBusy ||
isResuming ||
isConnecting ||
isRevertingCheckpoint ||
threadDetailLoading ||
activeEnvironmentUnavailable ||
sendInFlightRef.current
) {
return;
}
const threadId = activeThread.id;
sendInFlightRef.current = true;
setResumingThreadKeys((current) => new Set(current).add(routeThreadKey));
setThreadError(threadId, null);
// Commands settle into a result, so this always runs.
const resume = async () => {
if (resumableRunId === null) {
return resumeThreadQueue({ environmentId, input: { threadId } });
}
const createdAt = new Date().toISOString();
const settingsResult = await persistThreadSettingsForNextTurn({
threadId,
createdAt,
...(localCheckoutBranchMismatch
? { branch: localCheckoutBranchMismatch.currentBranch }
: {}),
runtimeMode,
interactionMode,
});
if (settingsResult._tag === "Failure") return settingsResult;
const modelSelection = composerRef.current?.getSendContext().selectedModelSelection;
const turnResult = await startThreadTurn({
environmentId,
input: {
threadId,
manualContinuationOfRunId: resumableRunId,
...(modelSelection ? { modelSelection } : {}),
message: {
messageId: newMessageId(),
role: "user",
text: "Continue where you left off.",
attachments: [],
},
runtimeMode,
interactionMode,
dispatchMode: "start",
},
});
if (turnResult._tag === "Failure" || !hasHeldQueuedRuns) return turnResult;
clearUsageLimitsFor(routeThreadKey);
return resumeThreadQueue({ environmentId, input: { threadId } });
};
const result = await resume();
if (result._tag === "Failure") {
if (!isAtomCommandInterrupted(result)) {
const error = squashAtomCommandFailure(result);
setThreadError(
threadId,
error instanceof Error ? error.message : "Could not resume thread.",
);
}
} else {
clearUsageLimitsFor(routeThreadKey);
if (currentRouteThreadKeyRef.current === routeThreadKey) scrollToEnd();
}
sendInFlightRef.current = false;
setResumingThreadKeys((current) => {
const next = new Set(current);
next.delete(routeThreadKey);
return next;
});
};

const onSend = async (
e?: { preventDefault: () => void },
dispatchMode: ComposerDispatchMode = "auto",
Expand Down Expand Up @@ -1912,5 +1819,5 @@ export function useComposerSend(input: ComposerSendInput) {
finish();
};

return { onSend, onResume, onImplementPlanInNewThread, isResuming };
return { onSend, onImplementPlanInNewThread };
}
Loading