fix: recover delayed session entry (#3631)

This commit is contained in:
Carlos Florêncio
2026-09-13 12:14:45 +01:00
committed by GitHub
parent a5fe42777b
commit 9e9d3b1fb0
39 changed files with 2684 additions and 194 deletions
@@ -1,5 +1,7 @@
"use client";
/* eslint-disable max-lines -- native transcript composition owns scrolling. */
import { useEffect, useMemo, useRef, memo, forwardRef, useImperativeHandle } from "react";
import { SessionPanelContent } from "@kandev/ui/pannel-session";
import type { Message, TaskSessionState } from "@/lib/types/http";
@@ -264,6 +266,9 @@ type NativeMessageListBodyProps = {
isLoadingMore: boolean;
isInitialLoading: boolean;
showLoadingState: boolean;
historyStatus: MessageListProps["historyStatus"];
historyError: MessageListProps["historyError"];
onRetryHistory: MessageListProps["onRetryHistory"];
retryLoadMore: () => void;
showRecovery: boolean;
sentinelRef: (node: HTMLDivElement | null) => void;
@@ -470,6 +475,9 @@ function NativeMessageListBody({
isLoadingMore,
isInitialLoading,
showLoadingState,
historyStatus,
historyError,
onRetryHistory,
retryLoadMore,
showRecovery,
sentinelRef,
@@ -495,6 +503,11 @@ function NativeMessageListBody({
messagesLoading={messagesLoading}
isInitialLoading={isInitialLoading}
messagesCount={messages.length}
sessionId={sessionId}
sessionState={sessionState}
historyStatus={historyStatus}
historyError={historyError}
onRetryHistory={onRetryHistory}
onLoadMore={retryLoadMore}
showRecovery={showRecovery}
/>
@@ -557,6 +570,9 @@ export const NativeMessageList = memo(
sessionId,
messagesLoading,
historyRefreshPending = false,
historyStatus = "ready",
historyError = null,
onRetryHistory,
isWorking,
sessionState,
worktreePath,
@@ -673,6 +689,9 @@ export const NativeMessageList = memo(
isLoadingMore={isLoadingMore}
isInitialLoading={isInitialLoading}
showLoadingState={showLoadingState}
historyStatus={historyStatus}
historyError={historyError}
onRetryHistory={onRetryHistory}
retryLoadMore={retryLoadMore}
showRecovery={showRecovery}
sentinelRef={sentinelRef}
@@ -825,6 +825,43 @@ describe("MessageListStatus", () => {
expect(screen.queryByTestId("conversation-loading-state")).not.toBeNull();
expect(screen.queryByText("Loading conversation...")).not.toBeNull();
});
it("does not show the empty invitation before history is ready", () => {
render(
<MessageListStatus
isLoadingMore={false}
hasMore={false}
showLoadingState
messagesLoading
isInitialLoading
messagesCount={0}
sessionId="sess-1"
historyStatus="loading"
onRetryHistory={vi.fn()}
/>,
);
expect(screen.queryByText("No messages yet. Start the conversation!")).toBeNull();
expect(screen.getByTestId("session-history-loading")).toBeTruthy();
});
it("shows the empty invitation after a successful empty history snapshot", () => {
render(
<MessageListStatus
isLoadingMore={false}
hasMore={false}
showLoadingState={false}
messagesLoading={false}
isInitialLoading={false}
messagesCount={0}
sessionId="sess-1"
historyStatus="ready"
onRetryHistory={vi.fn()}
/>,
);
expect(screen.getByText("No messages yet. Start the conversation!")).toBeTruthy();
});
});
describe("UnreadDivider", () => {
@@ -20,6 +20,8 @@ import {
import { isLaunchErrorSurfaceMessage } from "./types";
import { useTranslation } from "react-i18next";
import { useResponsiveBreakpoint } from "@/hooks/use-responsive-breakpoint";
import type { MessageHistoryStatus } from "@/hooks/domains/session/use-message-fetch-state";
import { SessionHistoryFeedback } from "./session-entry-feedback";
export type MessageListProps = {
items: RenderItem[];
@@ -33,6 +35,9 @@ export type MessageListProps = {
messagesLoading: boolean;
/** Latest-session history is still reconciling while cached rows remain visible. */
historyRefreshPending?: boolean;
historyStatus?: MessageHistoryStatus;
historyError?: unknown;
onRetryHistory?: () => void;
isWorking: boolean;
sessionState?: TaskSessionState;
worktreePath?: string;
@@ -470,9 +475,31 @@ export function UnreadDivider() {
);
}
function SessionHistoryStatus({
sessionId,
sessionState,
historyStatus,
historyError,
onRetryHistory,
}: {
sessionId: string | null;
sessionState?: TaskSessionState;
historyStatus: MessageHistoryStatus;
historyError: unknown;
onRetryHistory?: () => void;
}) {
if (!sessionId || sessionState === "CREATED" || historyStatus === "ready" || !onRetryHistory) {
return null;
}
return (
<SessionHistoryFeedback status={historyStatus} error={historyError} onRetry={onRetryHistory} />
);
}
/** Transcript status footer: the loading-older indicator, an explicit
* load-older button, the conversation loading spinner, and the empty-state
* message when there are no messages. */
// eslint-disable-next-line complexity -- transcript status owns independent loading, pagination, and recovery states.
export function MessageListStatus({
isLoadingMore,
hasMore,
@@ -482,6 +509,11 @@ export function MessageListStatus({
messagesCount,
onLoadMore,
showRecovery = false,
sessionId = null,
sessionState,
historyStatus = "ready",
historyError = null,
onRetryHistory,
}: {
isLoadingMore: boolean;
hasMore: boolean;
@@ -497,6 +529,11 @@ export function MessageListStatus({
onLoadMore?: () => void;
/** Shows the explicit control only after a recoverable pagination failure. */
showRecovery?: boolean;
sessionId?: string | null;
sessionState?: TaskSessionState;
historyStatus?: MessageHistoryStatus;
historyError?: unknown;
onRetryHistory?: () => void;
}) {
const { t } = useTranslation();
const { isFinePointer } = useResponsiveBreakpoint();
@@ -521,7 +558,14 @@ export function MessageListStatus({
</Button>
</div>
)}
{showLoadingState && (
<SessionHistoryStatus
sessionId={sessionId}
sessionState={sessionState}
historyStatus={historyStatus}
historyError={historyError}
onRetryHistory={onRetryHistory}
/>
{showLoadingState && historyStatus === "ready" && (
<div
className="flex items-center justify-center py-8 text-muted-foreground"
data-testid="conversation-loading-state"
@@ -530,11 +574,14 @@ export function MessageListStatus({
<span>{t("task:loadingConversation")}</span>
</div>
)}
{!messagesLoading && !isInitialLoading && messagesCount === 0 && (
<div className="flex items-center justify-center py-8 text-muted-foreground">
<span>{t("task:noMessagesYetStartTheConversation")}</span>
</div>
)}
{!messagesLoading &&
!isInitialLoading &&
messagesCount === 0 &&
historyStatus === "ready" && (
<div className="flex items-center justify-center py-8 text-muted-foreground">
<span>{t("task:noMessagesYetStartTheConversation")}</span>
</div>
)}
</>
);
}
@@ -0,0 +1,55 @@
import { cleanup, fireEvent, render, screen } from "@testing-library/react";
import { afterEach, describe, expect, it, vi } from "vitest";
const breakpoint = vi.hoisted(() => ({ isFinePointer: true }));
vi.mock("@/hooks/use-responsive-breakpoint", () => ({
useResponsiveBreakpoint: () => breakpoint,
}));
import { SessionHistoryFeedback } from "./session-entry-feedback";
afterEach(() => {
cleanup();
breakpoint.isFinePointer = true;
});
describe("SessionHistoryFeedback", () => {
it("announces loading without showing the empty conversation invitation", () => {
render(<SessionHistoryFeedback status="loading" error={null} onRetry={vi.fn()} />);
expect(screen.getByTestId("session-history-loading").getAttribute("role")).toBe("status");
expect(
screen.getByTestId("session-history-loading").querySelector('[aria-hidden="true"]'),
).toBeTruthy();
expect(screen.getByText("Loading conversation...")).toBeTruthy();
expect(screen.queryByText("No messages yet. Start the conversation!")).toBeNull();
});
it("shows a retry and collapsed technical detail after history fails", () => {
const onRetry = vi.fn();
render(
<SessionHistoryFeedback
status="unavailable"
error={new Error("WebSocket request timed out: message.list")}
onRetry={onRetry}
/>,
);
fireEvent.click(screen.getByTestId("session-history-retry"));
expect(onRetry).toHaveBeenCalledTimes(1);
const details = screen.getByTestId("session-history-details") as HTMLDetailsElement;
expect(details.open).toBe(false);
fireEvent.click(screen.getByTestId("session-history-details-summary"));
expect(details.open).toBe(true);
expect(screen.getByText("WebSocket request timed out: message.list")).toBeTruthy();
});
it("uses a touch-sized retry control for coarse pointers", () => {
breakpoint.isFinePointer = false;
render(<SessionHistoryFeedback status="unavailable" error={null} onRetry={vi.fn()} />);
expect(screen.getByTestId("session-history-retry").className).toContain("min-h-11");
});
});
@@ -0,0 +1,99 @@
"use client";
import { IconInfoCircle, IconRefresh } from "@tabler/icons-react";
import { Button } from "@kandev/ui/button";
import { useTranslation } from "react-i18next";
import { GridSpinner } from "@/components/grid-spinner";
import { useResponsiveBreakpoint } from "@/hooks/use-responsive-breakpoint";
import type { MessageHistoryStatus } from "@/hooks/domains/session/use-message-fetch-state";
import { cn } from "@/lib/utils";
function errorDetail(error: unknown, fallback: string): string {
if (error instanceof Error && error.message) return error.message;
if (typeof error === "string" && error) return error;
return fallback;
}
export function SessionHistoryFeedback({
status,
error,
onRetry,
}: {
status: MessageHistoryStatus;
error: unknown;
onRetry: () => void;
}) {
const { t } = useTranslation();
const { isFinePointer } = useResponsiveBreakpoint();
if (status === "ready") return null;
if (status === "loading" || status === "retrying") {
return (
<div
className="flex items-center justify-center gap-2 py-8 text-muted-foreground"
data-testid={status === "retrying" ? "session-history-retrying" : "session-history-loading"}
role="status"
aria-live="polite"
>
<span aria-hidden="true">
<GridSpinner className="text-primary" />
</span>
<span>
{status === "retrying" ? t("task:sessionHistoryRetrying") : t("task:loadingConversation")}
</span>
</div>
);
}
return (
<div
className="mb-3 min-w-0 rounded-md border border-border bg-muted/20 px-3 py-2 text-sm"
data-testid="session-history-unavailable"
role="status"
aria-live="polite"
>
<div className="flex min-w-0 items-start gap-2">
<IconInfoCircle
className="mt-0.5 size-4 shrink-0 text-muted-foreground"
aria-hidden="true"
/>
<div className="min-w-0 flex-1">
<div className="font-medium">{t("task:sessionHistoryUnavailable")}</div>
<div
className={cn(
"mt-2 flex gap-2",
isFinePointer ? "flex-wrap items-center" : "flex-col items-stretch",
)}
>
<Button
type="button"
variant="outline"
size="sm"
className={cn("cursor-pointer px-2 text-xs", !isFinePointer && "min-h-11 w-full")}
data-testid="session-history-retry"
onClick={onRetry}
>
<IconRefresh className="size-3" aria-hidden="true" />
{t("task:retry")}
</Button>
<details className="min-w-0 text-xs" data-testid="session-history-details">
<summary
className={cn(
"block cursor-pointer select-none rounded-sm underline underline-offset-2 focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring",
isFinePointer ? "min-h-7 py-1" : "min-h-11 py-3",
)}
data-testid="session-history-details-summary"
>
{t("task:details")}
</summary>
<p className="mt-2 break-words text-muted-foreground">
{errorDetail(error, t("task:sessionHistoryUnknownError"))}
</p>
</details>
</div>
</div>
</div>
</div>
);
}
@@ -454,6 +454,9 @@ function useSessionData(
historyRefreshPending,
historyInitialized,
hasMore: hasOlderMessages,
historyStatus,
historyError,
retryHistory,
} = useSessionMessages(resolvedSessionId);
const turns = useAppStore((state) =>
resolvedSessionId ? state.turns.bySession[resolvedSessionId] : undefined,
@@ -490,6 +493,9 @@ function useSessionData(
messagesLoading,
isInitialMessagesLoading,
historyRefreshPending,
historyStatus,
historyError,
retryHistory,
...processed,
sessionModel,
activeModel,
@@ -175,6 +175,31 @@ describe("EnsureSessionErrorBanner", () => {
expect(screen.getByText(RESUME_FAILURE_DETAIL)).toBeTruthy();
});
it("renders status failures as compact retryable feedback with collapsed details", () => {
const onRetry = vi.fn();
render(
<SessionRecoveryFeedback
error={null}
notice={null}
onRetry={onRetry}
recoveryFailure={{
outcome: "status_unavailable",
kind: "timeout",
statusError: "WebSocket request timed out: task.session.status",
}}
/>,
);
expect(screen.getByText("Session status is unavailable.")).toBeTruthy();
expect(screen.queryByText("Couldn't start a session")).toBeNull();
const details = screen.getByTestId("session-status-details") as HTMLDetailsElement;
expect(details.open).toBe(false);
fireEvent.click(screen.getByTestId("session-status-details-summary"));
expect(screen.getByText("WebSocket request timed out: task.session.status")).toBeTruthy();
fireEvent.click(screen.getByTestId("session-status-retry"));
expect(onRetry).toHaveBeenCalledTimes(1);
});
it("renders page-wide feedback when automatic recovery and workspace restoration fail", () => {
render(
<SessionRecoveryFeedback
@@ -8,6 +8,7 @@ import { useTranslation } from "react-i18next";
import { t } from "@/lib/i18n";
import { cn } from "@/lib/utils";
import type { SessionRecoveryFailure } from "@/hooks/domains/session/use-session-resumption";
import { useResponsiveBreakpoint } from "@/hooks/use-responsive-breakpoint";
// EnsureSessionErrorInfo wraps a parsed ensure error so UI can offer a targeted action for the missing-agent-profile case.
export type EnsureSessionErrorInfo = {
@@ -56,6 +57,21 @@ type RecoveryAction = {
disabled?: boolean;
};
export function getSessionRecoveryRetry(source: {
recoveryFailure: SessionRecoveryFailure | null;
resumeSession: () => Promise<boolean>;
retrySessionStatus: () => Promise<void>;
}): () => void {
if (source.recoveryFailure?.outcome === "status_unavailable") {
return () => {
void source.retrySessionStatus();
};
}
return () => {
void source.resumeSession();
};
}
type BannerProps = {
error: Error | null;
onRetry: () => void;
@@ -70,6 +86,7 @@ type BannerProps = {
function RecoveryFailureDetails({ failure }: { failure: SessionRecoveryFailure }) {
const { t } = useTranslation();
if (failure.outcome === "status_unavailable") return null;
return (
<details className="mt-2 min-w-0 text-xs" data-testid="session-recovery-details">
<summary
@@ -94,6 +111,53 @@ function RecoveryFailureDetails({ failure }: { failure: SessionRecoveryFailure }
);
}
function SessionStatusUnavailableNotice({
failure,
onRetry,
retryDisabled,
}: {
failure: Extract<SessionRecoveryFailure, { outcome: "status_unavailable" }>;
onRetry: () => void;
retryDisabled?: boolean;
}) {
const { t } = useTranslation();
const { isFinePointer } = useResponsiveBreakpoint();
return (
<div className="px-3 pt-2" data-testid="session-status-unavailable">
<Alert>
<IconInfoCircle />
<AlertTitle>{t("task:sessionStatusUnavailable")}</AlertTitle>
<AlertDescription>
<span>{t("task:sessionStatusUnavailableDetail")}</span>
<details className="mt-2 min-w-0 text-xs" data-testid="session-status-details">
<summary
className={cn(
"block max-w-full cursor-pointer select-none rounded-sm py-1 underline underline-offset-2 focus-visible:outline-none focus-visible:ring-2 focus-visible:ring-ring",
isFinePointer ? "min-h-7" : "min-h-11 py-3",
)}
data-testid="session-status-details-summary"
>
{t("task:details")}
</summary>
<p className="mt-2 break-words text-muted-foreground">{failure.statusError}</p>
</details>
<Button
variant="outline"
size="sm"
className={cn("mt-2 cursor-pointer px-2 text-xs", !isFinePointer && "min-h-11")}
onClick={onRetry}
disabled={retryDisabled}
data-testid="session-status-retry"
>
<IconRefresh className="size-3" aria-hidden="true" />
{t("task:retry")}
</Button>
</AlertDescription>
</Alert>
</div>
);
}
/** Slim banner for the task page, rendered above the layout. */
export function EnsureSessionErrorBanner({
error,
@@ -215,8 +279,16 @@ export function SessionRecoveryFeedback({
}) {
const readOnlyRecovery =
recoveryFailure?.outcome === "workspace_read_only" ? recoveryFailure : null;
const statusFailure = recoveryFailure?.outcome === "status_unavailable" ? recoveryFailure : null;
return (
<>
{statusFailure ? (
<SessionStatusUnavailableNotice
failure={statusFailure}
onRetry={onRetry}
retryDisabled={retryDisabled}
/>
) : null}
<EnsureSessionErrorBanner
error={error ? new Error(error) : null}
onRetry={onRetry}
@@ -13,7 +13,11 @@ import { useTaskStatusSummary } from "@/hooks/domains/task/use-task-status-summa
import type { UseEnsureTaskSessionResult } from "@/hooks/domains/session/use-ensure-task-session";
import type { AgentProfileOption } from "@/lib/state/slices";
import type { TaskSession } from "@/lib/types/http";
import { EnsureSessionErrorEmptyState, SessionRecoveryFeedback } from "./ensure-session-error";
import {
EnsureSessionErrorEmptyState,
getSessionRecoveryRetry,
SessionRecoveryFeedback,
} from "./ensure-session-error";
import { PassthroughToolbar } from "./passthrough-toolbar";
import { PreviewSessionTabMenu } from "./preview-session-tab-menu";
import { SessionTabDialogs } from "./session-tab-menu";
@@ -278,7 +282,7 @@ function PreviewSessionRecoverySurface({
error={resumption.error}
notice={resumption.notice}
recoveryFailure={resumption.recoveryFailure}
onRetry={() => void resumption.resumeSession()}
onRetry={getSessionRecoveryRetry(resumption)}
retryDisabled={
resumption.resumptionState === "checking" || resumption.resumptionState === "resuming"
}
@@ -678,7 +682,7 @@ function PreviewNoSessionsState({
error={resumption.error}
notice={resumption.notice}
recoveryFailure={resumption.recoveryFailure}
onRetry={() => void resumption.resumeSession()}
onRetry={getSessionRecoveryRetry(resumption)}
retryDisabled={
resumption.resumptionState === "checking" || resumption.resumptionState === "resuming"
}
@@ -1017,6 +1017,9 @@ export const TaskChatPanel = memo(function TaskChatPanel({
messagesLoading,
historyRefreshPending,
isInitialMessagesLoading,
historyStatus,
historyError,
retryHistory,
groupedItems,
allMessages,
footerActionMessages,
@@ -1194,6 +1197,9 @@ export const TaskChatPanel = memo(function TaskChatPanel({
sessionId={resolvedSessionId}
messagesLoading={messagesLoading}
historyRefreshPending={historyRefreshPending}
historyStatus={historyStatus}
historyError={historyError}
onRetryHistory={retryHistory}
isWorking={isWorking}
sessionState={session?.state}
worktreePath={getSessionWorkspacePath(session)}
+2 -1
View File
@@ -12,6 +12,7 @@ import { useAppStore } from "@/components/state-provider";
import type { UseEnsureTaskSessionResult } from "@/hooks/domains/session/use-ensure-task-session";
import {
EnsureSessionErrorBanner,
getSessionRecoveryRetry,
SessionRecoveryFeedback,
} from "@/components/task/ensure-session-error";
import { TaskMoveErrorBanner } from "@/components/task/task-move-error-banner";
@@ -271,7 +272,7 @@ function TaskPageRecoveryFeedback({
error={resumption.error}
notice={resumption.notice}
recoveryFailure={resumption.recoveryFailure}
onRetry={() => void resumption.resumeSession()}
onRetry={getSessionRecoveryRetry(resumption)}
retryDisabled={
resumption.resumptionState === "checking" || resumption.resumptionState === "resuming"
}
@@ -0,0 +1,113 @@
import type { Page } from "@playwright/test";
import { injectLatency } from "./causal-waits";
type WireFrame = {
id?: unknown;
type?: unknown;
action?: unknown;
};
type DelayRule = {
remaining: number;
delayMs: number;
reason: string;
};
export type SessionEntryRecoveryProxy = {
delayNextResponses: (action: string, count: number, delayMs: number, reason: string) => void;
requestCount: (action: string) => number;
delayedResponseCount: (action: string) => number;
};
function parseFrame(value: string): WireFrame | null {
try {
const parsed = JSON.parse(value) as unknown;
return typeof parsed === "object" && parsed !== null ? (parsed as WireFrame) : null;
} catch {
return null;
}
}
function isResponseFrame(frame: WireFrame | null): boolean {
return frame?.type === "response" || frame?.type === "error";
}
function responseAction(
frame: WireFrame | null,
requestActions: Map<string, string>,
): string | undefined {
if (typeof frame?.action === "string") return frame.action;
if (typeof frame?.id === "string") return requestActions.get(frame.id);
return undefined;
}
/**
* Delay selected gateway responses while forwarding every other frame. Rules
* correlate replies by request id, so the test never relies on action-only or
* payload timing and does not inspect message contents.
*/
export async function routeSessionEntryRecovery(page: Page): Promise<SessionEntryRecoveryProxy> {
const requestActions = new Map<string, string>();
const requestCounts = new Map<string, number>();
const delayedCounts = new Map<string, number>();
const rules = new Map<string, DelayRule>();
await page.routeWebSocket(/\/ws$/, (ws) => {
const server = ws.connectToServer();
ws.onMessage((message) => {
if (typeof message === "string") {
for (const part of message.split("\n")) {
const frame = parseFrame(part.trim());
if (
frame?.type === "request" &&
typeof frame.id === "string" &&
typeof frame.action === "string"
) {
requestActions.set(frame.id, frame.action);
requestCounts.set(frame.action, (requestCounts.get(frame.action) ?? 0) + 1);
}
}
}
server.send(message);
});
server.onMessage((message) => {
if (typeof message !== "string") {
ws.send(message);
return;
}
for (const part of message.split("\n")) {
const trimmed = part.trim();
if (!trimmed) continue;
const frame = parseFrame(trimmed);
const action = responseAction(frame, requestActions);
if (typeof frame?.id === "string") requestActions.delete(frame.id);
const rule = isResponseFrame(frame) && action ? rules.get(action) : null;
if (rule && rule.remaining > 0 && action) {
rule.remaining -= 1;
delayedCounts.set(action, (delayedCounts.get(action) ?? 0) + 1);
void (async () => {
await injectLatency(rule.delayMs, rule.reason);
ws.send(trimmed);
})();
continue;
}
ws.send(trimmed);
}
});
});
return {
delayNextResponses: (action, count, delayMs, reason) => {
if (count < 1) throw new Error("delayNextResponses requires a positive response count");
if (delayMs < 0) throw new Error("delayNextResponses requires a non-negative delay");
rules.set(action, { remaining: count, delayMs, reason });
},
requestCount: (action) => requestCounts.get(action) ?? 0,
delayedResponseCount: (action) => delayedCounts.get(action) ?? 0,
};
}
@@ -0,0 +1,58 @@
import { expect, test } from "../../fixtures/test-base";
import type { SeedData } from "../../fixtures/test-base";
import type { ApiClient } from "../../helpers/api-client";
import { assertNoDocumentHorizontalOverflow } from "../../helpers/layout-assertions";
import { openTaskSession } from "../../helpers/session";
import { routeSessionEntryRecovery } from "../../helpers/session-entry-recovery";
async function createEntryTask(apiClient: ApiClient, seedData: SeedData, title: string) {
return apiClient.createTaskWithAgent(seedData.workspaceId, title, seedData.agentProfileId, {
description: "/e2e:simple-message",
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
});
}
test.describe("mobile session entry recovery", () => {
test.describe.configure({ retries: 1 });
test("keeps cached access while history recovers with phone-sized controls", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(150_000);
const proxy = await routeSessionEntryRecovery(testPage);
const task = await createEntryTask(
apiClient,
seedData,
`Mobile history recovery ${Date.now()}`,
);
proxy.delayNextResponses(
"message.list",
2,
11_000,
"force both bounded history attempts to expose the mobile unavailable state",
);
const session = await openTaskSession(testPage, task.id);
const chat = session.activeChat();
const historyNotice = chat.getByTestId("session-history-unavailable");
await expect(historyNotice).toBeVisible({ timeout: 45_000 });
const retry = historyNotice.getByTestId("session-history-retry");
const retryBox = await retry.boundingBox();
expect(retryBox?.height).toBeGreaterThanOrEqual(44);
const detailsSummary = historyNotice.getByTestId("session-history-details-summary");
const detailsBox = await detailsSummary.boundingBox();
expect(detailsBox?.height).toBeGreaterThanOrEqual(44);
await assertNoDocumentHorizontalOverflow(testPage, "mobile session history recovery");
await retry.click();
await expect(historyNotice).toHaveCount(0);
await expect(chat).toContainText("simple mock response", { timeout: 30_000 });
expect(proxy.delayedResponseCount("message.list")).toBe(2);
});
});
@@ -0,0 +1,145 @@
import { expect, test } from "../../fixtures/test-base";
import type { SeedData } from "../../fixtures/test-base";
import type { ApiClient } from "../../helpers/api-client";
import { openTaskSession } from "../../helpers/session";
import { routeSessionEntryRecovery } from "../../helpers/session-entry-recovery";
async function createEntryTask(apiClient: ApiClient, seedData: SeedData, title: string) {
return apiClient.createTaskWithAgent(seedData.workspaceId, title, seedData.agentProfileId, {
description: "/e2e:simple-message",
workflow_id: seedData.workflowId,
workflow_step_id: seedData.startStepId,
repository_ids: [seedData.repositoryId],
});
}
test.describe("session entry recovery", () => {
test.describe.configure({ retries: 1 });
test("completes entry when status and subscription acknowledgements take seven seconds", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(90_000);
const proxy = await routeSessionEntryRecovery(testPage);
const task = await createEntryTask(apiClient, seedData, `Delayed entry ${Date.now()}`);
proxy.delayNextResponses(
"task.session.status",
1,
7_000,
"expose a delayed status acknowledgement without crossing the entry deadline",
);
proxy.delayNextResponses(
"session.subscribe",
1,
7_000,
"expose a delayed subscription acknowledgement without a reload",
);
const session = await openTaskSession(testPage, task.id);
await expect(session.activeChat()).toContainText("simple mock response", { timeout: 45_000 });
await expect(testPage.getByTestId("ensure-session-error-banner")).toHaveCount(0);
expect(proxy.delayedResponseCount("task.session.status")).toBe(1);
expect(proxy.delayedResponseCount("session.subscribe")).toBe(1);
});
test("retries a timed out subscription automatically while the page remains open", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(120_000);
const proxy = await routeSessionEntryRecovery(testPage);
const task = await createEntryTask(apiClient, seedData, `Retry entry ${Date.now()}`);
proxy.delayNextResponses(
"session.subscribe",
1,
11_000,
"force the first subscription acknowledgement past the ten-second deadline",
);
const session = await openTaskSession(testPage, task.id);
await expect
.poll(() => proxy.requestCount("session.subscribe"), {
timeout: 30_000,
message: "Waiting for the automatic second session.subscribe request",
})
.toBeGreaterThan(1);
await expect(session.activeChat()).toContainText("simple mock response", { timeout: 60_000 });
await expect(testPage.getByTestId("ensure-session-error-banner")).toHaveCount(0);
expect(proxy.delayedResponseCount("session.subscribe")).toBe(1);
});
test("shows one recoverable history notice and restores it with Retry", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(150_000);
const proxy = await routeSessionEntryRecovery(testPage);
const task = await createEntryTask(apiClient, seedData, `History recovery ${Date.now()}`);
proxy.delayNextResponses(
"message.list",
2,
11_000,
"force both bounded history attempts to expose the unavailable state",
);
const session = await openTaskSession(testPage, task.id);
const historyNotice = session.activeChat().getByTestId("session-history-unavailable");
await expect(historyNotice).toBeVisible({ timeout: 45_000 });
await expect(
session.activeChat().getByText("No messages yet. Start the conversation!", { exact: true }),
).toHaveCount(0);
const details = session.activeChat().getByTestId("session-history-details");
await expect(details).toBeVisible();
await expect(details).not.toHaveAttribute("open", "");
await details.getByTestId("session-history-details-summary").click();
await expect(details).toHaveAttribute("open", "");
await expect(details).toContainText("WebSocket request timed out: message.list");
await historyNotice.getByTestId("session-history-retry").click();
await expect(historyNotice).toHaveCount(0);
await expect(session.activeChat()).toContainText("simple mock response", { timeout: 30_000 });
expect(proxy.delayedResponseCount("message.list")).toBe(2);
});
test("labels exhausted status checks accurately and retries only the status read", async ({
testPage,
apiClient,
seedData,
}) => {
test.setTimeout(150_000);
const proxy = await routeSessionEntryRecovery(testPage);
const task = await createEntryTask(apiClient, seedData, `Status recovery ${Date.now()}`);
proxy.delayNextResponses(
"task.session.status",
2,
11_000,
"force both bounded status attempts to expose the compact status notice",
);
await openTaskSession(testPage, task.id);
const statusNotice = testPage.getByTestId("session-status-unavailable");
await expect(statusNotice).toBeVisible({ timeout: 45_000 });
await expect(statusNotice).not.toContainText("Couldn't start a session");
const details = statusNotice.getByTestId("session-status-details");
await expect(details).not.toHaveAttribute("open", "");
await details.getByTestId("session-status-details-summary").click();
await expect(details).toContainText("WebSocket request timed out: task.session.status");
const launchRequestCountBeforeRetry = proxy.requestCount("session.launch");
await statusNotice.getByTestId("session-status-retry").click();
await expect(statusNotice).toHaveCount(0);
expect(proxy.requestCount("task.session.status")).toBeGreaterThanOrEqual(3);
expect(proxy.requestCount("session.launch")).toBe(launchRequestCountBeforeRetry);
expect(proxy.delayedResponseCount("task.session.status")).toBe(2);
});
});
@@ -1,10 +1,14 @@
import { useMemo, useRef, useState } from "react";
import type { useAppStoreApi } from "@/components/state-provider";
export type MessageHistoryStatus = "loading" | "retrying" | "ready" | "unavailable";
export function useMessageFetchState(store: ReturnType<typeof useAppStoreApi>) {
const [isLoading, setIsLoading] = useState(false);
const [isWaitingForInitialMessages, setIsWaitingForInitialMessages] = useState(false);
const [isCachedHistoryRefreshPending, setIsCachedHistoryRefreshPending] = useState(false);
const [historyStatus, setHistoryStatus] = useState<MessageHistoryStatus>("loading");
const [historyError, setHistoryError] = useState<unknown>(null);
const cachedRefreshGenerationRef = useRef(0);
const initialFetchStartRef = useRef<number | null>(null);
const lastFetchedSessionIdRef = useRef<string | null>(null);
@@ -15,6 +19,8 @@ export function useMessageFetchState(store: ReturnType<typeof useAppStoreApi>) {
setIsWaitingForInitialMessages,
initialFetchStartRef,
lastFetchedSessionIdRef,
setHistoryStatus,
setHistoryError,
}),
[store],
);
@@ -23,6 +29,10 @@ export function useMessageFetchState(store: ReturnType<typeof useAppStoreApi>) {
isWaitingForInitialMessages,
isCachedHistoryRefreshPending,
setIsCachedHistoryRefreshPending,
historyStatus,
setHistoryStatus,
historyError,
setHistoryError,
setIsWaitingForInitialMessages,
initialFetchStartRef,
lastFetchedSessionIdRef,
@@ -27,6 +27,8 @@ function makeParams(
} as never,
setIsLoading: vi.fn(),
setIsWaitingForInitialMessages: vi.fn(),
setHistoryStatus: vi.fn(),
setHistoryError: vi.fn(),
initialFetchStartRef: { current: null },
lastFetchedSessionIdRef: { current: null },
fetchAndStoreMessages,
@@ -80,7 +82,7 @@ describe("doFetchMessages", () => {
expect(setMessagesLoading).toHaveBeenLastCalledWith(SESSION_ID, false);
});
it("clears hook gates when an active fetch becomes stale before settling", async () => {
it("releases store and local loading bookkeeping when a stale fetch settles", async () => {
const result = deferred<Message[]>();
const setMessagesLoading = vi.fn();
const params = makeParams(vi.fn().mockReturnValue(result.promise), setMessagesLoading);
@@ -92,7 +94,52 @@ describe("doFetchMessages", () => {
await fetch;
expect(params.setIsLoading).toHaveBeenLastCalledWith(false);
expect(params.setIsWaitingForInitialMessages).toHaveBeenLastCalledWith(false);
expect(params.setIsWaitingForInitialMessages).toHaveBeenLastCalledWith(true);
expect(setMessagesLoading).toHaveBeenLastCalledWith(SESSION_ID, false);
});
it("does not finalize a newer hook generation from a stale fetch", async () => {
const result = deferred<Message[]>();
const setMessagesLoading = vi.fn();
const params = makeParams(vi.fn().mockReturnValue(result.promise), setMessagesLoading);
const isActive = { value: true };
const fetch = doFetchMessages({
...params,
isActive: () => isActive.value,
canFinalizeLoading: () => isActive.value,
} as never);
isActive.value = false;
result.resolve([]);
await fetch;
expect(params.setIsLoading).toHaveBeenLastCalledWith(true);
expect(setMessagesLoading).toHaveBeenLastCalledWith(SESSION_ID, false);
});
it("preserves cached messages when the history request fails", async () => {
const setMessages = vi.fn();
const setMessagesLoading = vi.fn();
const store = {
getState: () => ({ setMessages, setMessagesLoading }),
} as unknown as Parameters<typeof doFetchMessages>[0]["store"];
const lastFetchedSessionIdRef = { current: null as string | null };
const initialFetchStartRef = { current: null as number | null };
await doFetchMessages({
taskSessionId: SESSION_ID,
store,
setIsLoading: vi.fn(),
setIsWaitingForInitialMessages: vi.fn(),
setHistoryStatus: vi.fn(),
setHistoryError: vi.fn(),
initialFetchStartRef,
lastFetchedSessionIdRef,
fetchAndStoreMessages: vi.fn().mockRejectedValue(new Error("history unavailable")),
onError: vi.fn(),
});
expect(setMessages).not.toHaveBeenCalled();
expect(lastFetchedSessionIdRef.current).toBeNull();
});
});
@@ -2,6 +2,7 @@ import { useEffect, type MutableRefObject } from "react";
import type { useAppStoreApi } from "@/components/state-provider";
import type { Message } from "@/lib/types/http";
import type { MessageHistoryStatus } from "./use-message-fetch-state";
type SessionMessageStore = ReturnType<typeof useAppStoreApi>;
@@ -103,17 +104,23 @@ type DoFetchMessagesParams = {
store: SessionMessageStore;
setIsLoading: (value: boolean) => void;
setIsWaitingForInitialMessages: (value: boolean) => void;
setHistoryStatus: (value: MessageHistoryStatus) => void;
setHistoryError: (value: unknown) => void;
initialFetchStartRef: MutableRefObject<number | null>;
lastFetchedSessionIdRef: MutableRefObject<string | null>;
// eslint-disable-next-line max-params -- hydration guards and retry feedback share one fetch contract.
fetchAndStoreMessages: (
sessionId: string,
store: SessionMessageStore,
isActive?: () => boolean,
hydrationRef?: SessionHydrationRef,
hydrationKey?: string,
onRetry?: () => void,
) => Promise<Message[]>;
onError?: (error: unknown) => void;
isActive?: () => boolean;
/** Allows the owning hook generation to fence local loading finalization. */
canFinalizeLoading?: () => boolean;
hydrationRef?: SessionHydrationRef;
hydrationKey?: string;
};
@@ -123,44 +130,51 @@ export async function doFetchMessages({
store,
setIsLoading,
setIsWaitingForInitialMessages,
setHistoryStatus,
setHistoryError,
initialFetchStartRef,
lastFetchedSessionIdRef,
fetchAndStoreMessages,
onError,
isActive,
canFinalizeLoading,
hydrationRef,
hydrationKey,
}: DoFetchMessagesParams): Promise<void> {
if (isInactive(isActive)) return;
beginSessionFetch(taskSessionId);
setIsLoading(true);
const isInitialHistoryFetch = lastFetchedSessionIdRef.current !== taskSessionId;
if (isInitialHistoryFetch) {
setHistoryStatus("loading");
setHistoryError(null);
}
store.getState().setMessagesLoading(taskSessionId, true);
if (initialFetchStartRef.current === null) {
initialFetchStartRef.current = Date.now();
setIsWaitingForInitialMessages(true);
}
try {
const fetched = await fetchAndStoreMessages(
taskSessionId,
store,
isActive,
hydrationRef,
hydrationKey,
await fetchAndStoreMessages(taskSessionId, store, isActive, hydrationRef, hydrationKey, () =>
setHistoryStatus("retrying"),
);
if (isInactive(isActive)) return;
lastFetchedSessionIdRef.current = taskSessionId;
if (fetched.length > 0) setIsWaitingForInitialMessages(false);
setHistoryStatus("ready");
setHistoryError(null);
setIsWaitingForInitialMessages(false);
} catch (error) {
if (isInactive(isActive)) return;
if (onError) onError(error);
else console.error("Failed to fetch messages:", error);
store.getState().setMessages(taskSessionId, []);
lastFetchedSessionIdRef.current = taskSessionId;
setHistoryStatus("unavailable");
setHistoryError(error);
} finally {
const active = !isInactive(isActive);
if (endSessionFetch(taskSessionId)) {
store.getState().setMessagesLoading(taskSessionId, false);
setIsLoading(false);
if (!canFinalizeLoading || canFinalizeLoading()) setIsLoading(false);
}
setIsWaitingForInitialMessages(false);
if (active) setIsWaitingForInitialMessages(false);
}
}
@@ -1,4 +1,6 @@
import { act, renderHook } from "@testing-library/react";
/* eslint-disable max-lines -- session history race regressions share one lifecycle test harness. */
import { act, renderHook, waitFor } from "@testing-library/react";
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import type { Message } from "@/lib/types/http";
@@ -60,6 +62,7 @@ vi.mock("@/components/state-provider", () => ({
}));
import { taskId, sessionId } from "@/lib/types/ids";
import { WebSocketRequestTimeoutError } from "@/lib/ws/client";
beforeEach(() => {
vi.clearAllMocks();
@@ -90,6 +93,7 @@ beforeEach(() => {
});
afterEach(() => {
vi.useRealTimers();
vi.restoreAllMocks();
});
import {
@@ -289,10 +293,45 @@ describe("stale concurrent fetch guard", () => {
function deferred<T>() {
let resolve!: (value: T | PromiseLike<T>) => void;
const promise = new Promise<T>((resolvePromise) => {
let reject!: (reason?: unknown) => void;
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
resolve = resolvePromise;
reject = rejectPromise;
});
return { promise, resolve };
return { promise, resolve, reject };
}
function configureSession(
id: string,
state: TaskSessionState,
messages: Message[],
historyInitialized = true,
) {
const messagesBySession = mockState.messages.bySession as Record<string, Message[]>;
const metaBySession = mockState.messages.metaBySession as Record<
string,
{
historyInitialized: boolean;
hasMore: boolean;
oldestCursor: string | null;
isLoading: boolean;
isLoadingMore: boolean;
}
>;
const sessions = mockState.taskSessions.items as Record<string, { state: TaskSessionState }>;
const turnsBySession = mockState.turns.bySession as Record<string, unknown[]>;
const activeTurns = mockState.turns.activeBySession as Record<string, string | null>;
messagesBySession[id] = messages;
metaBySession[id] = {
historyInitialized,
hasMore: false,
oldestCursor: null,
isLoading: false,
isLoadingMore: false,
};
sessions[id] = { state };
turnsBySession[id] = [];
activeTurns[id] = null;
}
describe("cached session entry history readiness", () => {
@@ -308,7 +347,7 @@ describe("cached session entry history readiness", () => {
mockWebSocketClient.request.mockReturnValue(response.promise);
mockState.messages.bySession["sess-1"] = [makeMessage({ id: "cached" })];
const { result, unmount } = renderHook(() => useSessionMessages("sess-1"));
const { result, rerender, unmount } = renderHook(() => useSessionMessages("sess-1"));
expect(result.current.historyRefreshPending).toBe(true);
@@ -318,6 +357,15 @@ describe("cached session entry history readiness", () => {
});
expect(result.current.historyRefreshPending).toBe(true);
// A live row can arrive while the cached refresh is still in flight.
// That changes the message count and reruns the entry effect before the
// refresh response settles.
mockState.messages.bySession["sess-1"] = [
makeMessage({ id: "cached" }),
makeMessage({ id: "live" }),
];
rerender();
await act(async () => {
response.resolve({ messages: [], has_more: false });
await response.promise;
@@ -442,6 +490,301 @@ describe("session subscription hydration ordering", () => {
});
});
describe("history retry state", () => {
it("retries a timed out history request and marks the snapshot ready", async () => {
vi.useFakeTimers();
mockWebSocketClient.getSessionSubscriptionReadiness.mockReturnValue(Promise.resolve());
mockWebSocketClient.subscribeSessionWithReady.mockReturnValue({
ready: Promise.resolve(),
unsubscribe: vi.fn(),
});
mockWebSocketClient.request
.mockRejectedValueOnce(new WebSocketRequestTimeoutError("message.list"))
.mockResolvedValueOnce({ messages: [], has_more: false });
const { result, unmount } = renderHook(() => useSessionMessages("sess-1"));
await act(async () => {
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
});
await act(async () => {
await vi.advanceTimersByTimeAsync(1000);
});
expect(mockWebSocketClient.request).toHaveBeenCalledTimes(2);
expect(result.current.historyStatus).toBe("ready");
unmount();
});
it("keeps a failed history snapshot unavailable without retrying non-timeout errors", async () => {
mockWebSocketClient.getSessionSubscriptionReadiness.mockReturnValue(Promise.resolve());
mockWebSocketClient.subscribeSessionWithReady.mockReturnValue({
ready: Promise.resolve(),
unsubscribe: vi.fn(),
});
mockWebSocketClient.request.mockRejectedValue(new Error("permission denied"));
vi.spyOn(console, "error").mockImplementation(() => {});
const { result, unmount } = renderHook(() => useSessionMessages("sess-1"));
await act(async () => {
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
});
expect(mockWebSocketClient.request).toHaveBeenCalledTimes(1);
expect(result.current.historyStatus).toBe("unavailable");
expect(result.current.historyError).toEqual(new Error("permission denied"));
unmount();
});
});
// eslint-disable-next-line max-lines-per-function -- deferred session transitions are easiest to audit as complete scenarios.
describe("history feedback generation", () => {
it("ignores a terminal fetch rejection from session A after switching to session B", async () => {
const aReadiness = deferred<void>();
const bReadiness = deferred<void>();
const aInitial = deferred<{ messages: Message[]; has_more: boolean }>();
const aTerminal = deferred<{ messages: Message[]; has_more: boolean }>();
const bInitial = deferred<{ messages: Message[]; has_more: boolean }>();
const readinessBySession = new Map([
["sess-a", aReadiness.promise],
["sess-b", bReadiness.promise],
]);
const aUser = makeMessage({ id: "a-user", session_id: sessionId("sess-a") });
const bUser = makeMessage({ id: "b-user", session_id: sessionId("sess-b") });
configureSession("sess-a", "RUNNING", [aUser]);
configureSession("sess-b", "RUNNING", [bUser]);
mockState.mergeMessages.mockImplementation((id, messages, metadata) => {
(mockState.messages.bySession as Record<string, Message[]>)[id] = messages;
(
mockState.messages.metaBySession as Record<
string,
(typeof mockState.messages.metaBySession)["sess-1"]
>
)[id] = {
...mockState.messages.metaBySession["sess-1"],
...metadata,
};
});
mockWebSocketClient.getSessionSubscriptionReadiness.mockImplementation(
(id: string) => readinessBySession.get(id) ?? Promise.resolve(),
);
mockWebSocketClient.subscribeSessionWithReady.mockImplementation((id: string) => ({
ready: readinessBySession.get(id) ?? Promise.resolve(),
unsubscribe: vi.fn(),
}));
mockWebSocketClient.request
.mockImplementationOnce(() => aInitial.promise)
.mockImplementationOnce(() => aTerminal.promise)
.mockImplementationOnce(() => bInitial.promise);
const { result, rerender, unmount } = renderHook(
({ activeSessionId }: { activeSessionId: string }) => useSessionMessages(activeSessionId),
{ initialProps: { activeSessionId: "sess-a" } },
);
await act(async () => {
aReadiness.resolve();
await aReadiness.promise;
});
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(1));
await act(async () => {
aInitial.resolve({ messages: [aUser], has_more: false });
await aInitial.promise;
});
(mockState.taskSessions.items as Record<string, { state: TaskSessionState }>)["sess-a"].state =
"WAITING_FOR_INPUT";
rerender({ activeSessionId: "sess-a" });
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(2));
rerender({ activeSessionId: "sess-b" });
await act(async () => {
bReadiness.resolve();
await bReadiness.promise;
});
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(3));
await act(async () => {
bInitial.resolve({ messages: [bUser], has_more: false });
await bInitial.promise;
});
await waitFor(() => expect(result.current.historyStatus).toBe("ready"));
await act(async () => {
aTerminal.reject(new Error("session A history failed"));
await aTerminal.promise.catch(() => undefined);
});
expect(result.current.historyStatus).toBe("ready");
expect(result.current.historyError).toBeNull();
unmount();
});
it("ignores an obsolete manual retry after leaving and re-entering session A", async () => {
const aReadiness = deferred<void>();
const bReadiness = deferred<void>();
const aReentryReadiness = deferred<void>();
const aInitial = deferred<{ messages: Message[]; has_more: boolean }>();
const aRetry = deferred<{ messages: Message[]; has_more: boolean }>();
const bInitial = deferred<{ messages: Message[]; has_more: boolean }>();
const aReentry = deferred<{ messages: Message[]; has_more: boolean }>();
const readinessBySession = new Map<string, Promise<void>>([["sess-a", aReadiness.promise]]);
const aUser = makeMessage({ id: "a-user", session_id: sessionId("sess-a") });
const bUser = makeMessage({ id: "b-user", session_id: sessionId("sess-b") });
const aNewUser = makeMessage({ id: "a-new-user", session_id: sessionId("sess-a") });
configureSession("sess-a", "RUNNING", [aUser]);
configureSession("sess-b", "RUNNING", [bUser]);
mockState.mergeMessages.mockImplementation((id, messages, metadata) => {
(mockState.messages.bySession as Record<string, Message[]>)[id] = messages;
(
mockState.messages.metaBySession as Record<
string,
(typeof mockState.messages.metaBySession)["sess-1"]
>
)[id] = {
...mockState.messages.metaBySession["sess-1"],
...metadata,
};
});
mockWebSocketClient.getSessionSubscriptionReadiness.mockImplementation(
(id: string) => readinessBySession.get(id) ?? Promise.resolve(),
);
mockWebSocketClient.subscribeSessionWithReady.mockImplementation((id: string) => ({
ready: readinessBySession.get(id) ?? Promise.resolve(),
unsubscribe: vi.fn(),
}));
mockWebSocketClient.request
.mockImplementationOnce(() => aInitial.promise)
.mockImplementationOnce(() => aRetry.promise)
.mockImplementationOnce(() => bInitial.promise)
.mockImplementationOnce(() => aReentry.promise);
const { result, rerender, unmount } = renderHook(
({ activeSessionId }: { activeSessionId: string }) => useSessionMessages(activeSessionId),
{ initialProps: { activeSessionId: "sess-a" } },
);
await act(async () => {
aReadiness.resolve();
await aReadiness.promise;
});
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(1));
await act(async () => {
aInitial.resolve({ messages: [aUser], has_more: false });
await aInitial.promise;
});
await act(async () => {
result.current.retryHistory();
await Promise.resolve();
});
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(2));
readinessBySession.set("sess-b", bReadiness.promise);
rerender({ activeSessionId: "sess-b" });
await act(async () => {
bReadiness.resolve();
await bReadiness.promise;
});
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(3));
await act(async () => {
bInitial.resolve({ messages: [bUser], has_more: false });
await bInitial.promise;
});
readinessBySession.set("sess-a", aReentryReadiness.promise);
rerender({ activeSessionId: "sess-a" });
await act(async () => {
aReentryReadiness.resolve();
await aReentryReadiness.promise;
});
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(4));
await act(async () => {
aReentry.resolve({ messages: [aNewUser], has_more: false });
await aReentry.promise;
});
await waitFor(() => expect(result.current.historyStatus).toBe("ready"));
await act(async () => {
aRetry.reject(new Error("obsolete retry failed"));
await aRetry.promise.catch(() => undefined);
});
expect(result.current.historyStatus).toBe("ready");
expect(result.current.historyError).toBeNull();
expect((mockState.messages.bySession as Record<string, Message[]>)["sess-a"]).toEqual([
aNewUser,
]);
unmount();
});
it("forces a fresh message snapshot when Retry follows a failed refresh", async () => {
const readiness = Promise.resolve();
const initial = deferred<{ messages: Message[]; has_more: boolean }>();
const failedRefresh = deferred<{ messages: Message[]; has_more: boolean }>();
const retried = deferred<{ messages: Message[]; has_more: boolean }>();
const oldUser = makeMessage({ id: "old-user", session_id: sessionId("sess-1") });
const newUser = makeMessage({ id: "new-user", session_id: sessionId("sess-1") });
configureSession("sess-1", "RUNNING", [oldUser]);
mockState.mergeMessages.mockImplementation((id, messages, metadata) => {
(mockState.messages.bySession as Record<string, Message[]>)[id] = messages;
(
mockState.messages.metaBySession as Record<
string,
(typeof mockState.messages.metaBySession)["sess-1"]
>
)[id] = {
...mockState.messages.metaBySession["sess-1"],
...metadata,
};
});
mockWebSocketClient.getSessionSubscriptionReadiness.mockReturnValue(readiness);
mockWebSocketClient.subscribeSessionWithReady.mockReturnValue({
ready: readiness,
unsubscribe: vi.fn(),
});
mockWebSocketClient.request
.mockImplementationOnce(() => initial.promise)
.mockImplementationOnce(() => failedRefresh.promise)
.mockImplementationOnce(() => retried.promise);
const { result, rerender, unmount } = renderHook(() => useSessionMessages("sess-1"));
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(1));
await act(async () => {
initial.resolve({ messages: [oldUser], has_more: false });
await initial.promise;
});
await act(async () => {
await new Promise((resolve) => window.setTimeout(resolve, 0));
});
(mockState.taskSessions.items as Record<string, { state: TaskSessionState }>)["sess-1"].state =
"WAITING_FOR_INPUT";
rerender();
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(2));
await act(async () => {
failedRefresh.reject(new Error("refresh failed"));
await failedRefresh.promise.catch(() => undefined);
});
await waitFor(() => expect(result.current.historyStatus).toBe("unavailable"));
await act(async () => {
result.current.retryHistory();
await Promise.resolve();
});
await waitFor(() => expect(mockWebSocketClient.request).toHaveBeenCalledTimes(3));
await act(async () => {
retried.resolve({ messages: [newUser], has_more: false });
await retried.promise;
});
await waitFor(() => expect(result.current.historyStatus).toBe("ready"));
expect(mockState.messages.bySession["sess-1"]).toEqual([newUser]);
unmount();
});
});
describe("turn loading for sessions without hydrated turns", () => {
it("fetches and merges turns when the session has none in the store", async () => {
const readiness = deferred<void>();
@@ -1,7 +1,12 @@
/* eslint-disable max-lines -- session hydration and lifecycle hooks share one transcript contract. */
import { useEffect, useRef, type MutableRefObject } from "react";
import { useCallback, useEffect, useLayoutEffect, useRef, type MutableRefObject } from "react";
import { getWebSocketClient } from "@/lib/ws/connection";
import {
isWebSocketRequestTimeoutError,
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
SESSION_ENTRY_RETRY_DELAY_MS,
} from "@/lib/ws/client";
import { useForegroundRefresh } from "@/hooks/use-foreground-refresh";
import { useAppStore, useAppStoreApi } from "@/components/state-provider";
import type { TaskSessionState, Message } from "@/lib/types/http";
@@ -18,6 +23,7 @@ import {
type SessionHydrationRef,
} from "./use-session-message-fetch";
import { useMessageFetchState } from "./use-message-fetch-state";
import type { MessageHistoryStatus } from "./use-message-fetch-state";
import { reconcileLatestMessageWindow } from "./message-window-reconciliation";
import { t } from "@/lib/i18n";
@@ -28,6 +34,7 @@ export { ensureSessionTurnsLoaded } from "./use-session-turns-hydration";
const INITIAL_FETCH_LIMIT = 100;
const RUNNING_BACKFILL_INITIAL_DELAY_MS = 1200;
const RUNNING_BACKFILL_INTERVAL_MS = 5000;
const MAX_HISTORY_FETCH_ATTEMPTS = 2;
// Monotonic guard against stale concurrent fetches. Each fetch claims an
// increasing sequence number before awaiting; a completion only merges if no
@@ -143,6 +150,9 @@ interface UseSessionMessagesReturn {
historyInitialized: boolean;
hasMore: boolean;
oldestCursor: string | null;
historyStatus: MessageHistoryStatus;
historyError: unknown;
retryHistory: () => void;
}
type MessageListResponse = { messages: Message[]; has_more?: boolean; cursor?: string };
@@ -150,6 +160,7 @@ type InFlightMessageRequest = {
readiness: Promise<void>;
promise: Promise<MessageListResponse>;
cachedAtRequest: Message[];
settled: boolean;
};
const EMPTY_MESSAGES: Message[] = [];
@@ -195,18 +206,23 @@ function requestSessionMessages(
cachedAtRequest: Message[],
): InFlightMessageRequest {
const existing = inFlightMessageRequests.get(sessionId);
if (existing?.readiness === readiness) return existing;
if (existing?.readiness === readiness && !existing.settled) return existing;
const requestParams = {
session_id: sessionId,
limit: INITIAL_FETCH_LIMIT,
sort: "desc" as const,
};
const promise = client.request<MessageListResponse>("message.list", requestParams, 10000);
const entry = { readiness, promise, cachedAtRequest: [...cachedAtRequest] };
const promise = client.request<MessageListResponse>(
"message.list",
requestParams,
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
);
const entry = { readiness, promise, cachedAtRequest: [...cachedAtRequest], settled: false };
inFlightMessageRequests.set(sessionId, entry);
void promise.then(
() => {
entry.settled = true;
window.setTimeout(() => {
if (inFlightMessageRequests.get(sessionId) === entry) {
inFlightMessageRequests.delete(sessionId);
@@ -214,6 +230,7 @@ function requestSessionMessages(
}, 0);
},
() => {
entry.settled = true;
window.setTimeout(() => {
if (inFlightMessageRequests.get(sessionId) === entry) {
inFlightMessageRequests.delete(sessionId);
@@ -225,7 +242,7 @@ function requestSessionMessages(
}
/** Fetch latest messages via WS and merge with any that arrived via live notifications. */
async function fetchAndStoreMessages(
async function fetchAndStoreMessagesAttempt(
sessionId: string,
store: ReturnType<typeof useAppStoreApi>,
isActive?: () => boolean,
@@ -291,6 +308,50 @@ async function fetchAndStoreMessages(
// and message content from the return, so `merged` is equivalent.
return merged;
}
function waitForHistoryRetry(isActive?: () => boolean): Promise<boolean> {
return new Promise((resolve) => {
const timer = window.setTimeout(
() => resolve(!isActive || isActive()),
SESSION_ENTRY_RETRY_DELAY_MS,
);
if (isActive && !isActive()) {
window.clearTimeout(timer);
resolve(false);
}
});
}
/** Fetch history with one bounded retry for a timed-out message.list request. */
// eslint-disable-next-line max-params -- session hydration needs its active and generation guards.
async function fetchAndStoreMessages(
sessionId: string,
store: ReturnType<typeof useAppStoreApi>,
isActive?: () => boolean,
hydrationRef?: SessionHydrationRef,
hydrationKey?: string,
onRetry?: () => void,
): Promise<Message[]> {
for (let attempt = 0; attempt < MAX_HISTORY_FETCH_ATTEMPTS; attempt += 1) {
try {
return await fetchAndStoreMessagesAttempt(
sessionId,
store,
isActive,
hydrationRef,
hydrationKey,
);
} catch (error) {
const retryable =
isWebSocketRequestTimeoutError(error) &&
error.action === "message.list" &&
attempt + 1 < MAX_HISTORY_FETCH_ATTEMPTS;
if (!retryable || !(await waitForHistoryRetry(isActive))) throw error;
if (!isActive || isActive()) onRetry?.();
}
}
return [];
}
/**
* When the initial fetch window contains no user/agent message rows (common
* when the latest turn produced hundreds of tool calls), the chat would render
@@ -304,10 +365,13 @@ function useTerminalStateFetch(
taskSessionId: string | null,
taskSessionState: TaskSessionState | null,
hasAgentMessage: boolean,
sessionFetchGenerationRef: MutableRefObject<number>,
refs: {
store: ReturnType<typeof useAppStoreApi>;
setIsLoading: (v: boolean) => void;
setIsWaitingForInitialMessages: (v: boolean) => void;
setHistoryStatus: (v: MessageHistoryStatus) => void;
setHistoryError: (v: unknown) => void;
initialFetchStartRef: MutableRefObject<number | null>;
lastFetchedSessionIdRef: MutableRefObject<string | null>;
},
@@ -315,20 +379,36 @@ function useTerminalStateFetch(
const lastFetchStateKeyRef = useRef<string | null>(null);
const connectionStatus = useAppStore((state) => state.connection.status);
useEffect(() => {
if (!taskSessionId || connectionStatus !== "connected") return;
if (!taskSessionState || hasAgentMessage) return;
let active = true;
const generation = sessionFetchGenerationRef.current;
const isActive = () => active && sessionFetchGenerationRef.current === generation;
const deactivate = () => {
active = false;
};
if (!taskSessionId || connectionStatus !== "connected") return deactivate;
if (!taskSessionState || hasAgentMessage) return deactivate;
const terminalStates = new Set<TaskSessionState>(["WAITING_FOR_INPUT", "COMPLETED", "FAILED"]);
if (!terminalStates.has(taskSessionState)) return;
if (!terminalStates.has(taskSessionState)) return deactivate;
const key = `${taskSessionId}:${taskSessionState}`;
if (lastFetchStateKeyRef.current === key) return;
if (lastFetchStateKeyRef.current === key) return deactivate;
lastFetchStateKeyRef.current = key;
void doFetchMessages({
taskSessionId,
...refs,
fetchAndStoreMessages,
isActive,
canFinalizeLoading: isActive,
onError: (error) => console.error("Failed to fetch messages after state change:", error),
});
}, [taskSessionId, taskSessionState, hasAgentMessage, connectionStatus, refs]);
return deactivate;
}, [
taskSessionId,
taskSessionState,
hasAgentMessage,
connectionStatus,
refs,
sessionFetchGenerationRef,
]);
}
// Silent WS disconnects (NAT timeout, laptop sleep, suspended tab) leave
@@ -338,10 +418,14 @@ function useTerminalStateFetch(
export function useVisibilityBackfill(
taskSessionId: string | null,
store: ReturnType<typeof useAppStoreApi>,
sessionFetchGenerationRef?: MutableRefObject<number>,
) {
useForegroundRefresh(
() => {
if (!taskSessionId) return;
const generation = sessionFetchGenerationRef?.current;
const isActive = () =>
generation === undefined || sessionFetchGenerationRef?.current === generation;
const visibilityState = document.visibilityState;
const state = store.getState();
const existingCount = state.messages.bySession[taskSessionId]?.length ?? 0;
@@ -354,7 +438,7 @@ export function useVisibilityBackfill(
existingCount,
newestBefore,
});
fetchAndStoreMessages(taskSessionId, store)
fetchAndStoreMessages(taskSessionId, store, isActive)
.then(() => {
const afterCount = store.getState().messages.bySession[taskSessionId]?.length ?? 0;
const newestAfter =
@@ -379,8 +463,10 @@ type SessionSubscriptionParams = {
taskSessionId: string | null;
connectionStatus: string;
store: ReturnType<typeof useAppStoreApi>;
fetchRefs: ReturnType<typeof useMessageFetchState>["refs"];
hydrationRef: SessionHydrationRef;
hydrationKey: string;
sessionFetchGenerationRef: MutableRefObject<number>;
};
function useSessionSubscription({
@@ -388,8 +474,10 @@ function useSessionSubscription({
connectionStatus,
isSessionStartingOrUnknown,
store,
fetchRefs,
hydrationRef,
hydrationKey,
sessionFetchGenerationRef,
}: SessionSubscriptionParams & { isSessionStartingOrUnknown: boolean }) {
useEffect(() => {
debug("subscription: effect ran", {
@@ -412,6 +500,7 @@ function useSessionSubscription({
debug("subscription: subscribing", { sessionId: taskSessionId });
const subscription = client.subscribeSessionWithReady(taskSessionId);
let active = true;
const generation = sessionFetchGenerationRef.current;
// Re-fetch messages after the server acknowledges the subscription to
// close the gap between SSR (which may have run before the agent
@@ -419,13 +508,17 @@ function useSessionSubscription({
void subscription.ready
.then(() => {
if (!active) return;
return fetchAndStoreMessages(
const isCurrentGeneration = () =>
active && sessionFetchGenerationRef.current === generation;
return doFetchMessages({
taskSessionId,
store,
() => active,
...fetchRefs,
fetchAndStoreMessages,
isActive: isCurrentGeneration,
canFinalizeLoading: isCurrentGeneration,
hydrationRef,
hydrationKey,
);
});
})
.catch(() => {});
@@ -438,9 +531,11 @@ function useSessionSubscription({
taskSessionId,
connectionStatus,
store,
fetchRefs,
isSessionStartingOrUnknown,
hydrationRef,
hydrationKey,
sessionFetchGenerationRef,
]);
}
@@ -458,6 +553,7 @@ function useResyncOnTurnSettle(
taskSessionState: TaskSessionState | null,
connectionStatus: string,
store: ReturnType<typeof useAppStoreApi>,
sessionFetchGenerationRef: MutableRefObject<number>,
) {
const prevRef = useRef<{ sessionId: string | null; state: TaskSessionState | null }>({
sessionId: null,
@@ -474,24 +570,35 @@ function useResyncOnTurnSettle(
prev: prevState,
next: taskSessionState,
});
fetchAndStoreMessages(taskSessionId, store).catch(() => {});
}, [taskSessionId, taskSessionState, connectionStatus, store]);
const generation = sessionFetchGenerationRef.current;
fetchAndStoreMessages(
taskSessionId,
store,
() => sessionFetchGenerationRef.current === generation,
).catch(() => {});
}, [taskSessionId, taskSessionState, connectionStatus, store, sessionFetchGenerationRef]);
}
function useRunningMessageBackfill(
taskSessionId: string | null,
shouldBackfill: boolean,
store: ReturnType<typeof useAppStoreApi>,
sessionFetchGenerationRef: MutableRefObject<number>,
) {
useEffect(() => {
if (!taskSessionId || !shouldBackfill) return;
const generation = sessionFetchGenerationRef.current;
let inFlight = false;
const sync = () => {
if (inFlight) return;
inFlight = true;
debug("running backfill", { sessionId: taskSessionId });
fetchAndStoreMessages(taskSessionId, store)
fetchAndStoreMessages(
taskSessionId,
store,
() => sessionFetchGenerationRef.current === generation,
)
.catch((err) => {
debug("running backfill failed", { sessionId: taskSessionId, err });
})
@@ -505,7 +612,7 @@ function useRunningMessageBackfill(
window.clearTimeout(initial);
window.clearInterval(interval);
};
}, [taskSessionId, shouldBackfill, store]);
}, [taskSessionId, shouldBackfill, store, sessionFetchGenerationRef]);
}
function useSessionMessageInputs(taskSessionId: string | null) {
@@ -538,8 +645,10 @@ function useSessionLifecycleSubscriptions(
activeTurnId,
messages,
store,
fetchRefs,
hydrationRef,
hydrationKey,
sessionFetchGenerationRef,
} = params;
// Bool flips exactly once when a freshly-adopted session leaves STARTING,
// so the subscription effect re-runs then (covering the backend race where
@@ -556,15 +665,23 @@ function useSessionLifecycleSubscriptions(
connectionStatus,
isSessionStartingOrUnknown,
store,
fetchRefs,
hydrationRef,
hydrationKey,
sessionFetchGenerationRef,
});
useUnknownSessionSubscriptionRetryEffect({
taskSessionId,
connectionStatus,
retryToken: unknownSessionRetryToken,
});
useResyncOnTurnSettle(taskSessionId, taskSessionState, connectionStatus, store);
useResyncOnTurnSettle(
taskSessionId,
taskSessionState,
connectionStatus,
store,
sessionFetchGenerationRef,
);
useRunningMessageBackfill(
taskSessionId,
shouldRunMessageBackfill({
@@ -574,25 +691,30 @@ function useSessionLifecycleSubscriptions(
messages,
}),
store,
sessionFetchGenerationRef,
);
}
type SessionEntryFetchParams = {
taskSessionId: string | null;
connectionStatus: string;
messagesLength: number;
historyInitialized: boolean;
store: ReturnType<typeof useAppStoreApi>;
prevSessionIdRef: MutableRefObject<string | null>;
fetchState: ReturnType<typeof useMessageFetchState>;
hydrationRef: SessionHydrationRef;
hydrationKey: string;
sessionFetchGenerationRef: MutableRefObject<number>;
};
function useInitialMessagesWait(params: SessionEntryFetchParams): void {
const { taskSessionId, messagesLength, fetchState } = params;
const { taskSessionId, messagesLength, historyInitialized, fetchState } = params;
const {
initialFetchStartRef,
lastFetchedSessionIdRef,
setIsWaitingForInitialMessages,
setIsCachedHistoryRefreshPending,
setHistoryStatus,
setHistoryError,
} = fetchState;
useEffect(() => {
if (!taskSessionId) {
@@ -600,10 +722,16 @@ function useInitialMessagesWait(params: SessionEntryFetchParams): void {
lastFetchedSessionIdRef.current = null;
setIsWaitingForInitialMessages(false);
setIsCachedHistoryRefreshPending(false);
setHistoryStatus("ready");
setHistoryError(null);
return;
}
if (messagesLength > 0) {
setIsWaitingForInitialMessages(false);
if (historyInitialized) {
setHistoryStatus("ready");
setHistoryError(null);
}
return;
}
if (initialFetchStartRef.current === null) {
@@ -617,6 +745,9 @@ function useInitialMessagesWait(params: SessionEntryFetchParams): void {
lastFetchedSessionIdRef,
setIsWaitingForInitialMessages,
setIsCachedHistoryRefreshPending,
setHistoryStatus,
setHistoryError,
historyInitialized,
]);
}
function useSessionEntryMessageFetch(params: SessionEntryFetchParams): void {
@@ -624,11 +755,13 @@ function useSessionEntryMessageFetch(params: SessionEntryFetchParams): void {
taskSessionId,
connectionStatus,
messagesLength,
historyInitialized,
store,
prevSessionIdRef,
fetchState,
hydrationRef,
hydrationKey,
sessionFetchGenerationRef,
} = params;
const {
lastFetchedSessionIdRef,
@@ -639,6 +772,9 @@ function useSessionEntryMessageFetch(params: SessionEntryFetchParams): void {
} = fetchState;
useEffect(() => {
let active = true;
const generation = sessionFetchGenerationRef.current;
const isCurrentGeneration = () => sessionFetchGenerationRef.current === generation;
const isActive = () => active && isCurrentGeneration();
const deactivate = () => {
active = false;
};
@@ -655,6 +791,7 @@ function useSessionEntryMessageFetch(params: SessionEntryFetchParams): void {
if (messagesLength > 0 && !sessionChanged && !isFreshMount) {
lastFetchedSessionIdRef.current = taskSessionId;
setIsWaitingForInitialMessages(false);
if (historyInitialized) fetchRefs.setHistoryStatus("ready");
return deactivate;
}
if (isFreshMount && messagesLength > 0) {
@@ -662,10 +799,19 @@ function useSessionEntryMessageFetch(params: SessionEntryFetchParams): void {
setIsWaitingForInitialMessages(false);
setIsCachedHistoryRefreshPending(true);
const refreshGeneration = ++cachedRefreshGenerationRef.current;
void fetchAndStoreMessages(taskSessionId, store, () => active, hydrationRef, hydrationKey)
void doFetchMessages({
taskSessionId,
...fetchRefs,
setIsWaitingForInitialMessages: () => {},
fetchAndStoreMessages,
isActive: isCurrentGeneration,
canFinalizeLoading: isCurrentGeneration,
hydrationRef,
hydrationKey,
})
.catch(() => {})
.finally(() => {
if (cachedRefreshGenerationRef.current === refreshGeneration) {
if (isCurrentGeneration() && cachedRefreshGenerationRef.current === refreshGeneration) {
setIsCachedHistoryRefreshPending(false);
}
});
@@ -676,7 +822,8 @@ function useSessionEntryMessageFetch(params: SessionEntryFetchParams): void {
taskSessionId,
...fetchRefs,
fetchAndStoreMessages,
isActive: () => active,
isActive,
canFinalizeLoading: isActive,
hydrationRef,
hydrationKey,
});
@@ -685,6 +832,7 @@ function useSessionEntryMessageFetch(params: SessionEntryFetchParams): void {
taskSessionId,
connectionStatus,
messagesLength,
historyInitialized,
store,
prevSessionIdRef,
lastFetchedSessionIdRef,
@@ -693,8 +841,98 @@ function useSessionEntryMessageFetch(params: SessionEntryFetchParams): void {
fetchRefs,
hydrationRef,
hydrationKey,
sessionFetchGenerationRef,
]);
}
function useSessionHistoryRecoveryState({
taskSessionId,
connectionStatus,
messages,
messagesLoading,
isWaitingForInitialMessages,
messagesMetaLoading,
fetchState,
hydrationRef,
hydrationKey,
}: {
taskSessionId: string | null;
connectionStatus: string;
messages: Message[];
messagesLoading: boolean;
isWaitingForInitialMessages: boolean;
messagesMetaLoading: boolean;
fetchState: ReturnType<typeof useMessageFetchState>;
hydrationRef: SessionHydrationRef;
hydrationKey: string;
}) {
const activeSessionIdRef = useRef(taskSessionId);
const historySessionIdRef = useRef(taskSessionId);
const previousConnectionStatusRef = useRef(connectionStatus);
const sessionFetchGenerationRef = useRef(0);
const { refs: fetchRefs } = fetchState;
useLayoutEffect(() => {
sessionFetchGenerationRef.current += 1;
activeSessionIdRef.current = taskSessionId;
const sessionChanged = historySessionIdRef.current !== taskSessionId;
const connectionChanged = previousConnectionStatusRef.current !== connectionStatus;
previousConnectionStatusRef.current = connectionStatus;
if (sessionChanged || connectionChanged) fetchRefs.setIsLoading(false);
if (sessionChanged) {
historySessionIdRef.current = taskSessionId;
fetchRefs.setHistoryStatus(taskSessionId ? "loading" : "ready");
fetchRefs.setHistoryError(null);
}
return () => {
sessionFetchGenerationRef.current += 1;
};
}, [connectionStatus, fetchRefs, taskSessionId]);
const retryHistory = useCallback(() => {
const retrySessionId = taskSessionId;
if (
!retrySessionId ||
connectionStatus !== "connected" ||
messagesLoading ||
isWaitingForInitialMessages ||
messagesMetaLoading
) {
return;
}
const generation = sessionFetchGenerationRef.current;
fetchState.lastFetchedSessionIdRef.current = null;
fetchState.cachedRefreshGenerationRef.current += 1;
hydrationRef.current = null;
const isCurrentGeneration = () =>
activeSessionIdRef.current === retrySessionId &&
sessionFetchGenerationRef.current === generation;
void doFetchMessages({
taskSessionId: retrySessionId,
...fetchRefs,
setIsWaitingForInitialMessages:
messages.length > 0 ? () => {} : fetchRefs.setIsWaitingForInitialMessages,
fetchAndStoreMessages,
isActive: isCurrentGeneration,
canFinalizeLoading: isCurrentGeneration,
hydrationRef,
hydrationKey,
});
}, [
connectionStatus,
fetchRefs,
hydrationKey,
hydrationRef,
isWaitingForInitialMessages,
messages.length,
messagesLoading,
messagesMetaLoading,
taskSessionId,
]);
return { retryHistory, sessionFetchGenerationRef };
}
export function useSessionMessages(taskSessionId: string | null): UseSessionMessagesReturn {
const store = useAppStoreApi();
const { messages, messagesMeta, taskSessionState, activeTurnId, connectionStatus } =
@@ -714,8 +952,21 @@ export function useSessionMessages(taskSessionId: string | null): UseSessionMess
isLoading,
isWaitingForInitialMessages,
isCachedHistoryRefreshPending,
historyStatus,
historyError,
refs: fetchRefs,
} = fetchState;
const { retryHistory, sessionFetchGenerationRef } = useSessionHistoryRecoveryState({
taskSessionId,
connectionStatus,
messages,
messagesLoading: isLoading,
isWaitingForInitialMessages,
messagesMetaLoading: messagesMeta.isLoading,
fetchState,
hydrationRef,
hydrationKey,
});
useSessionLifecycleSubscriptions({
taskSessionId,
taskSessionState,
@@ -723,23 +974,33 @@ export function useSessionMessages(taskSessionId: string | null): UseSessionMess
activeTurnId,
messages,
store,
fetchRefs,
hydrationRef,
hydrationKey,
sessionFetchGenerationRef,
});
const entryFetchParams = {
taskSessionId,
connectionStatus,
messagesLength: messages.length,
historyInitialized: messagesMeta.historyInitialized,
store,
prevSessionIdRef,
fetchState,
hydrationRef,
hydrationKey,
sessionFetchGenerationRef,
};
useInitialMessagesWait(entryFetchParams);
useSessionEntryMessageFetch(entryFetchParams);
useVisibilityBackfill(taskSessionId, store);
useTerminalStateFetch(taskSessionId, taskSessionState, hasAgentMessage, fetchRefs);
useVisibilityBackfill(taskSessionId, store, sessionFetchGenerationRef);
useTerminalStateFetch(
taskSessionId,
taskSessionState,
hasAgentMessage,
sessionFetchGenerationRef,
fetchRefs,
);
return {
isLoading: isLoading || isWaitingForInitialMessages || messagesMeta.isLoading,
@@ -754,5 +1015,8 @@ export function useSessionMessages(taskSessionId: string | null): UseSessionMess
historyInitialized: messagesMeta.historyInitialized,
hasMore: messagesMeta.hasMore,
oldestCursor: messagesMeta.oldestCursor,
historyStatus,
historyError,
retryHistory,
};
}
@@ -5,6 +5,7 @@ type RecoveryResumptionState = "resumed" | "running";
type RecoveryFeedbackSetters = {
setError: (value: string | null) => void;
setNotice?: (value: string | null) => void;
setRecoveryFailure?: (value: null) => void;
setResumptionState: (value: RecoveryResumptionState) => void;
};
@@ -20,7 +21,7 @@ export function useSessionRecoveryFeedback(
notice: string | null,
setters: RecoveryFeedbackSetters,
): void {
const { setError, setNotice, setResumptionState } = setters;
const { setError, setNotice, setRecoveryFailure, setResumptionState } = setters;
const lastObservedSessionRef = useRef<{ id: string | null; state?: string }>({
id: sessionId,
state: sessionState,
@@ -33,6 +34,16 @@ export function useSessionRecoveryFeedback(
if (!stateChanged || !isActiveSessionState(sessionState)) return;
if (error !== null) setError(null);
if (notice !== null) setNotice?.(null);
setRecoveryFailure?.(null);
setResumptionState(sessionState === "RUNNING" ? "running" : "resumed");
}, [error, notice, sessionId, sessionState, setError, setNotice, setResumptionState]);
}, [
error,
notice,
sessionId,
sessionState,
setError,
setNotice,
setRecoveryFailure,
setResumptionState,
]);
}
@@ -62,6 +62,11 @@ export type SessionRecoveryFailure =
outcome: "recovery_failed";
resumeError: string;
restoreError: string;
}
| {
outcome: "status_unavailable";
kind: "timeout" | "request";
statusError: string;
};
export type ResumeStateSetter = {
@@ -108,10 +108,14 @@ describe("useSessionResumption archive lifecycle", () => {
await Promise.resolve();
});
expect(mockRequest).toHaveBeenCalledWith("task.session.status", {
task_id: TASK_ID,
session_id: SESSION_ID,
});
expect(mockRequest).toHaveBeenCalledWith(
"task.session.status",
{
task_id: TASK_ID,
session_id: SESSION_ID,
},
10000,
);
});
it("does not start an agent after unarchive when automatic start is disabled", async () => {
@@ -1,7 +1,7 @@
/* eslint-disable max-lines -- session resumption cases share one lifecycle harness. */
import { act, renderHook, waitFor } from "@testing-library/react";
import { describe, it, expect, vi, beforeEach } from "vitest";
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
import { buildRestoreWorkspaceRequest } from "@/lib/services/session-launch-helpers";
import { sanitizeWorkspaceRestorationDetails } from "@/lib/state/slices/session-runtime/workspace-restoration";
import { createAppStore } from "@/lib/state/store";
@@ -14,24 +14,42 @@ let mockConnectionStatus = "connected";
let mockPreventAutoStart = false;
let mockSessionItems: Record<string, { started_at?: string; updated_at?: string; state?: string }> =
{};
let mockWorkspaceRestorationEnabled = false;
let mockWorkspaceStore = createAppStore();
function getMockState() {
const workspaceState = mockWorkspaceStore.getState();
return {
connection: { status: mockConnectionStatus },
taskSessions: { items: mockSessionItems },
setTaskSession: mockSetTaskSession,
setSessionAgentctlStatus: mockSetSessionAgentctlStatus,
setResumeSkipped: mockSetResumeSkipped,
userSettings: { preventAutoStartAgentOnOpen: mockPreventAutoStart },
tasks: { resumeSkippedSessionIds: {} },
sessionAgentctl: workspaceState.sessionAgentctl,
...(mockWorkspaceRestorationEnabled
? {
environmentIdBySessionId: { [SESSION_ID]: "environment-1" },
workspaceRestoration: workspaceState.workspaceRestoration,
beginWorkspaceRestoration: workspaceState.beginWorkspaceRestoration,
completeWorkspaceRestoration: workspaceState.completeWorkspaceRestoration,
failWorkspaceRestoration: workspaceState.failWorkspaceRestoration,
clearWorkspaceRestoration: workspaceState.clearWorkspaceRestoration,
bumpWorkspaceFilesRefresh: workspaceState.bumpWorkspaceFilesRefresh,
}
: {}),
};
}
vi.mock("@/lib/ws/connection", () => ({
getWebSocketClient: () => ({ request: mockRequest }),
}));
vi.mock("@/components/state-provider", () => ({
useAppStore: (selector: (state: Record<string, unknown>) => unknown) =>
selector({
connection: { status: mockConnectionStatus },
taskSessions: { items: mockSessionItems },
setTaskSession: mockSetTaskSession,
setSessionAgentctlStatus: mockSetSessionAgentctlStatus,
setResumeSkipped: mockSetResumeSkipped,
userSettings: { preventAutoStartAgentOnOpen: mockPreventAutoStart },
tasks: { resumeSkippedSessionIds: {} },
}),
useAppStore: (selector: (state: Record<string, unknown>) => unknown) => selector(getMockState()),
useAppStoreApi: () => ({
getState: () => ({ taskSessions: { items: mockSessionItems } }),
getState: getMockState,
}),
}));
@@ -45,6 +63,13 @@ const LATER_AT = "2026-01-02T00:00:00.000Z";
const RESUME_TRANSPORT_ERROR = "Resume transport failed";
const WORKSPACE_RESTORE_ERROR = "Workspace restore failed";
afterEach(() => {
vi.useRealTimers();
mockRequest.mockReset();
mockWorkspaceRestorationEnabled = false;
mockWorkspaceStore = createAppStore();
});
import {
decideResumeAction,
resumeWithSilentFallback,
@@ -53,6 +78,7 @@ import {
type ResumptionState,
type SessionRecoveryFailure,
} from "./use-session-resumption";
import { WebSocketRequestTimeoutError } from "@/lib/ws/client";
import { resumeViaLaunch } from "./use-session-resumption-operations";
import { buildGuardedSetters } from "./use-session-resumption-request-guard";
@@ -451,6 +477,7 @@ describe("resumeWithSilentFallback", () => {
});
});
// eslint-disable-next-line max-lines-per-function -- status and recovery cases share one lifecycle harness.
describe("useSessionResumption", () => {
beforeEach(() => {
vi.clearAllMocks();
@@ -479,6 +506,169 @@ describe("useSessionResumption", () => {
expect(mockSetTaskSession).toHaveBeenCalledWith(expect.objectContaining({ updated_at: "" }));
});
it("retries a timed out status check before processing the response", async () => {
vi.useFakeTimers();
mockRequest
.mockRejectedValueOnce(new WebSocketRequestTimeoutError(STATUS_ACTION))
.mockResolvedValueOnce({
session_id: SESSION_ID,
task_id: TASK_ID,
state: "WAITING_FOR_INPUT",
is_agent_running: false,
is_resumable: false,
needs_resume: false,
});
renderHook(() => useSessionResumption(TASK_ID, SESSION_ID));
await act(async () => {
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
await Promise.resolve();
});
await act(async () => {
await vi.advanceTimersByTimeAsync(1000);
});
expect(mockRequest).toHaveBeenCalledTimes(2);
expect(mockRequest).toHaveBeenNthCalledWith(
1,
STATUS_ACTION,
{ task_id: TASK_ID, session_id: SESSION_ID },
10000,
);
expect(mockSetTaskSession).toHaveBeenCalledWith(
expect.objectContaining({ state: "WAITING_FOR_INPUT" }),
);
});
it("surfaces a status timeout as status-unavailable feedback after the retry budget", async () => {
vi.useFakeTimers();
mockRequest
.mockRejectedValueOnce(new WebSocketRequestTimeoutError(STATUS_ACTION))
.mockRejectedValueOnce(new WebSocketRequestTimeoutError(STATUS_ACTION));
const { result } = renderHook(() => useSessionResumption(TASK_ID, SESSION_ID));
await act(async () => {
await Promise.resolve();
await Promise.resolve();
await vi.advanceTimersByTimeAsync(1000);
await Promise.resolve();
await Promise.resolve();
});
expect(result.current.resumptionState).toBe("error");
expect(result.current.error).toBeNull();
expect(result.current.recoveryFailure).toEqual({
outcome: "status_unavailable",
kind: "timeout",
statusError: "WebSocket request timed out: task.session.status",
});
expect(mockRequest).toHaveBeenCalledTimes(2);
mockRequest.mockResolvedValueOnce({
session_id: SESSION_ID,
task_id: TASK_ID,
state: "RUNNING",
is_agent_running: true,
is_resumable: false,
needs_resume: false,
});
await act(async () => {
await result.current.retrySessionStatus();
});
expect(mockRequest).toHaveBeenCalledTimes(3);
expect(mockRequest).not.toHaveBeenCalledWith(
LAUNCH_ACTION,
expect.anything(),
expect.anything(),
);
expect(result.current.recoveryFailure).toBeNull();
expect(result.current.resumptionState).toBe("running");
});
it("keeps a workspace restore failure as launch feedback with a launch retry", async () => {
mockWorkspaceRestorationEnabled = true;
mockRequest
.mockResolvedValueOnce({
session_id: SESSION_ID,
task_id: TASK_ID,
state: "WAITING_FOR_INPUT",
is_agent_running: false,
is_resumable: false,
needs_resume: false,
needs_workspace_restore: true,
})
.mockRejectedValueOnce(new Error(WORKSPACE_RESTORE_ERROR));
const { result } = renderHook(() => useSessionResumption(TASK_ID, SESSION_ID));
await waitFor(() => expect(result.current.workspaceRestoration.status).toBe("error"));
expect(result.current.workspaceRestoration.attempt?.details).toBe(WORKSPACE_RESTORE_ERROR);
expect(result.current.resumptionState).toBe("error");
expect(result.current.error).toBeNull();
expect(result.current.recoveryFailure).toBeNull();
expect(mockRequest).toHaveBeenNthCalledWith(
2,
LAUNCH_ACTION,
expect.objectContaining({ intent: "restore_workspace" }),
expect.any(Number),
);
mockRequest.mockResolvedValueOnce({
success: true,
task_id: TASK_ID,
session_id: SESSION_ID,
state: "STARTING",
});
await act(async () => {
await result.current.resumeSession();
});
expect(mockRequest).toHaveBeenNthCalledWith(
3,
LAUNCH_ACTION,
expect.objectContaining({ intent: "resume" }),
expect.any(Number),
);
});
it("keeps a backend status error visible after a timeout retry without launching", async () => {
vi.useFakeTimers();
mockRequest
.mockRejectedValueOnce(new WebSocketRequestTimeoutError(STATUS_ACTION))
.mockRejectedValueOnce(new WebSocketRequestTimeoutError(STATUS_ACTION));
const { result } = renderHook(() => useSessionResumption(TASK_ID, SESSION_ID));
await act(async () => {
await Promise.resolve();
await Promise.resolve();
await vi.advanceTimersByTimeAsync(1000);
await Promise.resolve();
await Promise.resolve();
});
mockRequest.mockResolvedValueOnce({
session_id: SESSION_ID,
task_id: TASK_ID,
state: "WAITING_FOR_INPUT",
is_agent_running: false,
is_resumable: false,
needs_resume: false,
error: "session not found",
});
await act(async () => {
await result.current.retrySessionStatus();
});
expect(result.current.resumptionState).toBe("error");
expect(result.current.error).toBe("session not found");
expect(result.current.recoveryFailure).toBeNull();
expect(mockRequest.mock.calls.filter(([action]) => action === LAUNCH_ACTION)).toHaveLength(0);
});
it("refreshes the status and embedded editor capability after a successful resume", async () => {
mockRequest
.mockResolvedValueOnce({
@@ -511,10 +701,14 @@ describe("useSessionResumption", () => {
await waitFor(() => {
expect(result.current.sessionStatus?.capabilities?.embedded_vscode).toBe(true);
});
expect(mockRequest).toHaveBeenLastCalledWith(STATUS_ACTION, {
task_id: TASK_ID,
session_id: SESSION_ID,
});
expect(mockRequest).toHaveBeenLastCalledWith(
STATUS_ACTION,
{
task_id: TASK_ID,
session_id: SESSION_ID,
},
10000,
);
});
it("clears stale feedback when an external recovery makes the live session active", async () => {
@@ -650,10 +844,14 @@ describe("useSessionResumption prevent-auto-start gate", () => {
renderHook(() => useSessionResumption(TASK_ID, SESSION_ID));
await waitFor(() => {
expect(mockRequest).toHaveBeenCalledWith(STATUS_ACTION, {
task_id: TASK_ID,
session_id: SESSION_ID,
});
expect(mockRequest).toHaveBeenCalledWith(
STATUS_ACTION,
{
task_id: TASK_ID,
session_id: SESSION_ID,
},
10000,
);
});
// The skip branch must consult the live row (RUNNING) and refuse.
expect(mockSetResumeSkipped).not.toHaveBeenCalled();
@@ -690,10 +888,14 @@ describe("useSessionResumption monotonic terminal hydration", () => {
renderHook(() => useSessionResumption("t1", "s1"));
await waitFor(() => {
expect(mockRequest).toHaveBeenCalledWith(STATUS_ACTION, {
task_id: "t1",
session_id: "s1",
});
expect(mockRequest).toHaveBeenCalledWith(
STATUS_ACTION,
{
task_id: "t1",
session_id: "s1",
},
10000,
);
});
// The stale COMPLETED must not overwrite the newer live FAILED.
expect(mockSetTaskSession).not.toHaveBeenCalledWith(
@@ -722,10 +924,14 @@ describe("useSessionResumption monotonic terminal hydration", () => {
renderHook(() => useSessionResumption("t1", "s1"));
await waitFor(() => {
expect(mockRequest).toHaveBeenCalledWith(STATUS_ACTION, {
task_id: "t1",
session_id: "s1",
});
expect(mockRequest).toHaveBeenCalledWith(
STATUS_ACTION,
{
task_id: "t1",
session_id: "s1",
},
10000,
);
});
// The stale STARTING must not overwrite the newer live FAILED.
expect(mockSetTaskSession).not.toHaveBeenCalledWith(
@@ -781,10 +987,14 @@ describe("useSessionResumption monotonic terminal hydration", () => {
renderHook(() => useSessionResumption("t1", "s1"));
await waitFor(() => {
expect(mockRequest).toHaveBeenCalledWith(STATUS_ACTION, {
task_id: "t1",
session_id: "s1",
});
expect(mockRequest).toHaveBeenCalledWith(
STATUS_ACTION,
{
task_id: "t1",
session_id: "s1",
},
10000,
);
});
// WAITING_FOR_INPUT means the agent is alive; a stale older response must
// not downgrade it to a stopped-looking state.
@@ -836,10 +1046,14 @@ describe("useSessionResumption completed-session admission", () => {
renderHook(() => useSessionResumption(TASK_ID, SESSION_ID));
await waitFor(() => {
expect(mockRequest).toHaveBeenCalledWith(STATUS_ACTION, {
task_id: TASK_ID,
session_id: SESSION_ID,
});
expect(mockRequest).toHaveBeenCalledWith(
STATUS_ACTION,
{
task_id: TASK_ID,
session_id: SESSION_ID,
},
10000,
);
});
expect(mockRequest).not.toHaveBeenCalledWith(
LAUNCH_ACTION,
@@ -908,10 +1122,14 @@ describe("useSessionResumption stale-callback guard after navigation", () => {
);
await waitFor(() => {
expect(mockRequest).toHaveBeenCalledWith(STATUS_ACTION, {
task_id: TASK_ID,
session_id: SESSION_ID,
});
expect(mockRequest).toHaveBeenCalledWith(
STATUS_ACTION,
{
task_id: TASK_ID,
session_id: SESSION_ID,
},
10000,
);
});
// Navigate to another session before the first status response resolves.
@@ -1,4 +1,11 @@
/* eslint-disable max-lines -- session resumption coordinates one guarded lifecycle. */
import { useCallback, useEffect, useLayoutEffect, useRef, useState } from "react";
import {
isWebSocketRequestTimeoutError,
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
SESSION_ENTRY_RETRY_DELAY_MS,
} from "@/lib/ws/client";
import { getWebSocketClient } from "@/lib/ws/connection";
import { launchSession } from "@/lib/services/session-launch-service";
import {
@@ -67,6 +74,33 @@ type CheckAndResumeParams = {
};
const TERMINAL_STATES = new Set<TaskSessionState>(["FAILED", "CANCELLED", "COMPLETED"]);
const MAX_SESSION_STATUS_ATTEMPTS = 2;
function sessionStatusFailure(
error: unknown,
): Extract<SessionRecoveryFailure, { outcome: "status_unavailable" }> {
return {
outcome: "status_unavailable",
kind: isWebSocketRequestTimeoutError(error) ? "timeout" : "request",
statusError: error instanceof Error && error.message ? error.message : t("common:unknownError"),
};
}
/** Apply permanent outcomes returned inside an otherwise successful status response. */
function applyStatusResponseOutcome(status: SessionStatus, setters: ResumeStateSetter): boolean {
if (status.error) {
setters.setRecoveryFailure?.(null);
setters.setResumptionState("error");
setters.setError(status.error);
setters.setNotice?.(null);
return true;
}
if (status.resume_reason === TASK_ARCHIVED_KIND) {
clearArchiveRecovery(setters);
return true;
}
return false;
}
type LiveSessionLike = (SessionLike & { state?: string }) | null;
@@ -137,6 +171,42 @@ type RefreshSessionStatusParams = {
canContinue: () => boolean;
};
function waitForSessionStatusRetry(canContinue: () => boolean): Promise<boolean> {
return new Promise((resolve) => {
const timer = window.setTimeout(() => resolve(canContinue()), SESSION_ENTRY_RETRY_DELAY_MS);
if (!canContinue()) {
window.clearTimeout(timer);
resolve(false);
}
});
}
async function requestSessionStatusWithRetry({
client,
taskId,
sessionId,
canContinue,
}: Omit<
RefreshSessionStatusParams,
"session" | "setSessionStatus" | "setters"
>): Promise<SessionStatus | null> {
for (let attempt = 0; attempt < MAX_SESSION_STATUS_ATTEMPTS; attempt += 1) {
if (!canContinue()) return null;
try {
return await client.request<SessionStatus>(
"task.session.status",
{ task_id: taskId, session_id: sessionId },
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
);
} catch (error) {
const canRetry =
isWebSocketRequestTimeoutError(error) && attempt + 1 < MAX_SESSION_STATUS_ATTEMPTS;
if (!canRetry || !(await waitForSessionStatusRetry(canContinue))) throw error;
}
}
return null;
}
async function refreshSessionStatus({
client,
taskId,
@@ -145,14 +215,17 @@ async function refreshSessionStatus({
setSessionStatus,
setters,
canContinue,
}: RefreshSessionStatusParams): Promise<void> {
if (!canContinue()) return;
}: RefreshSessionStatusParams): Promise<SessionStatus | null> {
if (!canContinue()) return null;
try {
const status = await client.request<SessionStatus>("task.session.status", {
task_id: taskId,
session_id: sessionId,
const status = await requestSessionStatusWithRetry({
client,
taskId,
sessionId,
canContinue,
});
if (!canContinue()) return;
if (!status || !canContinue()) return null;
if (applyStatusResponseOutcome(status, setters)) return null;
setSessionStatus(status);
applyStatusToState(status, taskId, sessionId, session, setters);
// A status response confirming the agent is running must clear any
@@ -164,8 +237,17 @@ async function refreshSessionStatus({
if (status.is_agent_running && setters.setAgentctlReady) {
setters.setAgentctlReady(sessionId);
}
return status;
} catch (err) {
console.error("[refreshSessionStatus] failed to refresh session status", { sessionId, err });
if (isTaskArchivedConflict(err) || !canContinue()) {
clearArchiveRecovery(setters);
return null;
}
setters.setResumptionState("error");
setters.setError(null);
setters.setNotice?.(null);
setters.setRecoveryFailure?.(sessionStatusFailure(err));
return null;
}
}
@@ -244,16 +326,7 @@ async function processResumeStatus({
preventAutoStart,
canContinue,
}: ProcessResumeStatusParams): Promise<boolean> {
if (status.error) {
setters.setRecoveryFailure?.(null);
setters.setResumptionState("error");
setters.setError(status.error);
return false;
}
if (status.resume_reason === TASK_ARCHIVED_KIND) {
clearArchiveRecovery(setters);
return false;
}
if (applyStatusResponseOutcome(status, setters)) return false;
applyStatusToState(status, taskId, sessionId, session, setters);
if (status.is_agent_running && setters.setAgentctlReady) {
setters.setAgentctlReady(sessionId);
@@ -269,6 +342,7 @@ async function processResumeStatus({
});
}
// eslint-disable-next-line complexity -- status, restore, archive, and stale-request outcomes share one guarded transition.
async function checkAndResume({
taskId,
sessionId,
@@ -286,12 +360,28 @@ async function checkAndResume({
setters.setError(null);
setters.setNotice?.(null);
setters.setRecoveryFailure?.(null);
let status: SessionStatus | null;
try {
const status = await client.request<SessionStatus>("task.session.status", {
task_id: taskId,
session_id: sessionId,
status = await requestSessionStatusWithRetry({
client,
taskId,
sessionId,
canContinue,
});
if (!canContinue()) return;
} catch (err) {
if (isTaskArchivedConflict(err) || !canContinue()) {
clearArchiveRecovery(setters);
return;
}
setters.setResumptionState("error");
setters.setError(null);
setters.setNotice?.(null);
setters.setRecoveryFailure?.(sessionStatusFailure(err));
return;
}
if (!status || !canContinue()) return;
if (applyStatusResponseOutcome(status, setters)) return;
try {
setSessionStatus(status);
const resumed = await processResumeStatus({
status,
@@ -319,9 +409,9 @@ async function checkAndResume({
return;
}
setters.setResumptionState("error");
setters.setRecoveryFailure?.(null);
setters.setError(err instanceof Error ? err.message : t("common:unknownError"));
setters.setNotice?.(null);
setters.setRecoveryFailure?.(null);
}
}
@@ -335,6 +425,7 @@ interface UseSessionResumptionReturn {
worktreePath: string | null;
worktreeBranch: string | null;
resumeSession: () => Promise<boolean>;
retrySessionStatus: () => Promise<void>;
workspaceRestoration: WorkspaceRestorationResult;
}
@@ -347,6 +438,7 @@ type SessionResetAndCheckResult = {
sessionStatus: SessionStatus | null;
captureRequest: () => SessionRequestIdentity;
buildGuardedSettersFor: (capturedRequest: SessionRequestIdentity) => ResumeStateSetter;
retryStatus: () => Promise<void>;
};
type ResetAndCheckParams = {
@@ -366,6 +458,7 @@ const getSessionRequestKey = (
) => JSON.stringify([taskId, sessionId, taskArchiveState]);
/** Extracted effects: reset state on session/task change, auto-check/resume, and remote retry. */
// eslint-disable-next-line max-lines-per-function -- one hook owns the request identity and lifecycle effects.
function useSessionResetAndCheck({
taskId,
sessionId,
@@ -456,10 +549,14 @@ function useSessionResetAndCheck({
if (!client) return;
remoteStatusRetryCount.current += 1;
try {
const nextStatus = await client.request<SessionStatus>("task.session.status", {
task_id: taskId,
session_id: sessionId,
});
const nextStatus = await client.request<SessionStatus>(
"task.session.status",
{
task_id: taskId,
session_id: sessionId,
},
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
);
if (isCurrentRequest(activeRequestRef.current, capturedRequest)) {
setSessionStatus({ requestKey: capturedRequest.key, status: nextStatus });
}
@@ -471,11 +568,47 @@ function useSessionResetAndCheck({
return () => window.clearTimeout(timer);
}, [taskId, sessionId, connectionStatus, sessionStatus, taskArchiveState]);
const retryStatus = useCallback(async () => {
if (!taskId || !sessionId || connectionStatus !== "connected" || taskArchiveState !== false) {
return;
}
const capturedRequest = activeRequestRef.current;
const guardedSetters = buildGuardedSetters(activeRequestRef, capturedRequest, setters);
const canContinue = () => isCurrentRequest(activeRequestRef.current, capturedRequest);
guardedSetters.setResumptionState("checking");
guardedSetters.setError(null);
guardedSetters.setNotice?.(null);
guardedSetters.setRecoveryFailure?.(null);
const client = getWebSocketClient();
if (!client) return;
const status = await refreshSessionStatus({
client,
taskId,
sessionId,
session,
setSessionStatus: (nextStatus) => {
if (canContinue()) {
setSessionStatus({ requestKey: capturedRequest.key, status: nextStatus });
}
},
setters: guardedSetters,
canContinue,
});
if (!status || !canContinue()) return;
guardedSetters.setError(null);
guardedSetters.setNotice?.(null);
guardedSetters.setRecoveryFailure?.(null);
guardedSetters.setResumptionState(
status.is_agent_running || status.state === "RUNNING" ? "running" : "idle",
);
}, [connectionStatus, session, sessionId, setters, taskArchiveState, taskId]);
return {
sessionStatus,
captureRequest: () => activeRequestRef.current,
buildGuardedSettersFor: (capturedRequest) =>
buildGuardedSetters(activeRequestRef, capturedRequest, setters),
retryStatus,
};
}
@@ -627,15 +760,16 @@ export function useSessionResumption(
useSessionRecoveryFeedback(sessionId, session?.state, error, notice, setters);
const { sessionStatus, captureRequest, buildGuardedSettersFor } = useSessionResetAndCheck({
taskId,
sessionId,
connectionStatus,
session,
setters,
preventAutoStart: preventAutoStartAgentOnOpen,
taskArchiveState,
});
const { sessionStatus, captureRequest, buildGuardedSettersFor, retryStatus } =
useSessionResetAndCheck({
taskId,
sessionId,
connectionStatus,
session,
setters,
preventAutoStart: preventAutoStartAgentOnOpen,
taskArchiveState,
});
const resumeSession = useManualResumeSession({
taskId,
@@ -656,6 +790,7 @@ export function useSessionResumption(
worktreePath,
worktreeBranch,
resumeSession,
retrySessionStatus: retryStatus,
workspaceRestoration,
};
}
+83 -16
View File
@@ -1,6 +1,6 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { WebSocketClient, WebSocketRequestError } from "./client";
import { WebSocketClient, WebSocketRequestError, WebSocketRequestTimeoutError } from "./client";
type SentRequest = {
id: string;
@@ -9,6 +9,9 @@ type SentRequest = {
payload: unknown;
};
const SESSION_ID = "sess-1";
const SESSION_SUBSCRIBE_ACTION = "session.subscribe";
class FakeWebSocket {
static readonly OPEN = 1;
static readonly CLOSED = 3;
@@ -66,7 +69,9 @@ function connectClient(options?: ConstructorParameters<typeof WebSocketClient>[2
}
function sessionSubscribeRequest(socket: FakeWebSocket, index = 0) {
const request = socket.sent.filter((message) => message.action === "session.subscribe")[index];
const request = socket.sent.filter((message) => message.action === SESSION_SUBSCRIBE_ACTION)[
index
];
if (!request) throw new Error("No session.subscribe request was sent");
return request;
}
@@ -89,10 +94,70 @@ afterEach(() => {
vi.useRealTimers();
});
// eslint-disable-next-line max-lines-per-function -- the readiness suite covers the shared lifecycle contract.
describe("session subscription readiness", () => {
it("accepts a registration acknowledgement that arrives after seven seconds", async () => {
vi.useFakeTimers();
const { client, socket } = connectClient();
const subscription = client.subscribeSessionWithReady(SESSION_ID);
await vi.advanceTimersByTimeAsync(7000);
acknowledge(socket, sessionSubscribeRequest(socket));
await expect(subscription.ready).resolves.toBeUndefined();
subscription.unsubscribe();
});
it("recovers timed out registration without a visibility change", async () => {
vi.useFakeTimers();
const { client, socket } = connectClient();
const subscription = client.subscribeSessionWithReady(SESSION_ID);
await vi.advanceTimersByTimeAsync(10000);
expect(
socket.sent.filter((message) => message.action === SESSION_SUBSCRIBE_ACTION),
).toHaveLength(1);
await vi.advanceTimersByTimeAsync(1000);
const retryRequest = sessionSubscribeRequest(socket, 1);
acknowledge(socket, retryRequest);
await expect(subscription.ready).resolves.toBeUndefined();
subscription.unsubscribe();
});
it("stops after the bounded registration retry budget", async () => {
vi.useFakeTimers();
const { client, socket } = connectClient();
const subscription = client.subscribeSessionWithReady(SESSION_ID);
await vi.advanceTimersByTimeAsync(21000);
await expect(subscription.ready).rejects.toBeInstanceOf(WebSocketRequestTimeoutError);
expect(
socket.sent.filter((message) => message.action === SESSION_SUBSCRIBE_ACTION),
).toHaveLength(2);
subscription.unsubscribe();
});
it("cancels a scheduled registration retry when the last consumer unsubscribes", async () => {
vi.useFakeTimers();
const { client, socket } = connectClient();
const subscription = client.subscribeSessionWithReady(SESSION_ID);
await vi.advanceTimersByTimeAsync(10000);
subscription.unsubscribe();
await expect(subscription.ready).rejects.toThrow("Session subscription released");
await vi.advanceTimersByTimeAsync(1000);
expect(
socket.sent.filter((message) => message.action === SESSION_SUBSCRIBE_ACTION),
).toHaveLength(1);
});
it("resolves only after the server acknowledges the registration", async () => {
const { client, socket } = connectClient();
const subscription = client.subscribeSessionWithReady("sess-1");
const subscription = client.subscribeSessionWithReady(SESSION_ID);
const request = sessionSubscribeRequest(socket);
let ready = false;
@@ -111,11 +176,13 @@ describe("session subscription readiness", () => {
it("shares one in-flight acknowledgement between ref-counted consumers", async () => {
const { client, socket } = connectClient();
const first = client.subscribeSessionWithReady("sess-1");
const second = client.subscribeSessionWithReady("sess-1");
const first = client.subscribeSessionWithReady(SESSION_ID);
const second = client.subscribeSessionWithReady(SESSION_ID);
expect(second.ready).toBe(first.ready);
expect(socket.sent.filter((message) => message.action === "session.subscribe")).toHaveLength(1);
expect(
socket.sent.filter((message) => message.action === SESSION_SUBSCRIBE_ACTION),
).toHaveLength(1);
acknowledge(socket, sessionSubscribeRequest(socket));
await expect(second.ready).resolves.toBeUndefined();
@@ -126,7 +193,7 @@ describe("session subscription readiness", () => {
it("allows a failed registration to be retried with fresh readiness", async () => {
const { client, socket } = connectClient();
const subscription = client.subscribeSessionWithReady("sess-1");
const subscription = client.subscribeSessionWithReady(SESSION_ID);
const firstRequest = sessionSubscribeRequest(socket);
socket.receive({
@@ -136,7 +203,7 @@ describe("session subscription readiness", () => {
});
await expect(subscription.ready).rejects.toThrow("session is not ready");
const retry = client.resubscribeSession("sess-1");
const retry = client.resubscribeSession(SESSION_ID);
const retryRequest = sessionSubscribeRequest(socket, 1);
expect(retry).not.toBe(subscription.ready);
@@ -148,7 +215,7 @@ describe("session subscription readiness", () => {
it("tracks the re-registration after reconnect", async () => {
vi.useFakeTimers();
const { client, socket } = connectClient({ enabled: true, initialDelay: 0, maxAttempts: 1 });
const initial = client.subscribeSessionWithReady("sess-1");
const initial = client.subscribeSessionWithReady(SESSION_ID);
acknowledge(socket, sessionSubscribeRequest(socket));
await expect(initial.ready).resolves.toBeUndefined();
@@ -157,9 +224,9 @@ describe("session subscription readiness", () => {
const reconnectedSocket = FakeWebSocket.latest();
reconnectedSocket.open();
const reconnected = client.subscribeSessionWithReady("sess-1");
const reconnected = client.subscribeSessionWithReady(SESSION_ID);
const reconnectRequest = sessionSubscribeRequest(reconnectedSocket, 0);
expect(reconnectRequest.payload).toEqual({ session_id: "sess-1" });
expect(reconnectRequest.payload).toEqual({ session_id: SESSION_ID });
expect(reconnected.ready).not.toBe(initial.ready);
acknowledge(reconnectedSocket, reconnectRequest);
@@ -255,12 +322,12 @@ describe("session subscription reconnect recovery", () => {
it("keeps queued hydration behind the reconnect subscription acknowledgement", async () => {
vi.useFakeTimers();
const { client, socket } = connectClient({ enabled: true, initialDelay: 0, maxAttempts: 1 });
const initial = client.subscribeSessionWithReady("sess-1");
const initial = client.subscribeSessionWithReady(SESSION_ID);
acknowledge(socket, sessionSubscribeRequest(socket));
await expect(initial.ready).resolves.toBeUndefined();
socket.close();
const reconnectReadiness = client.getSessionSubscriptionReadiness("sess-1");
const reconnectReadiness = client.getSessionSubscriptionReadiness(SESSION_ID);
let readinessResolved = false;
void reconnectReadiness.then(() => {
readinessResolved = true;
@@ -269,7 +336,7 @@ describe("session subscription reconnect recovery", () => {
id: "hydration-1",
type: "request",
action: "message.list",
payload: { session_id: "sess-1" },
payload: { session_id: SESSION_ID },
});
await Promise.resolve();
expect(readinessResolved).toBe(false);
@@ -279,7 +346,7 @@ describe("session subscription reconnect recovery", () => {
reconnectedSocket.open();
expect(reconnectedSocket.sent.map((message) => message.action)).toEqual([
"session.subscribe",
SESSION_SUBSCRIBE_ACTION,
"message.list",
]);
const reconnectRequest = sessionSubscribeRequest(reconnectedSocket);
@@ -296,7 +363,7 @@ describe("session subscription reconnect recovery", () => {
it("rejects an active readiness when reconnect recovery is disabled", async () => {
const { client, socket } = connectClient({ enabled: false });
const subscription = client.subscribeSessionWithReady("sess-1");
const subscription = client.subscribeSessionWithReady(SESSION_ID);
socket.close();
await expect(subscription.ready).rejects.toThrow("WebSocket connection closed");
+52 -6
View File
@@ -1,10 +1,21 @@
/* eslint-disable max-lines -- the shared WebSocket client owns all transport lifecycle paths. */
import type { BackendMessageMap, BackendMessageType } from "@/lib/types/backend";
import type { ConnectionStatus } from "@/lib/types/connection";
import { generateUUID } from "@/lib/utils";
import { createDebugLogger, isDebug } from "@/lib/debug/log";
import { dispatchToPluginWsHandlers } from "@/lib/ws/plugin-bridge";
import { toWebSocketRequestError } from "./request-error";
export { WebSocketRequestError, type WebSocketRequestErrorDetails } from "./request-error";
import {
isWebSocketRequestTimeoutError,
toWebSocketRequestError,
WebSocketRequestTimeoutError,
} from "./request-error";
export {
isWebSocketRequestTimeoutError,
WebSocketRequestError,
WebSocketRequestTimeoutError,
type WebSocketRequestErrorDetails,
} from "./request-error";
const debugDispatch = createDebugLogger("ws:dispatch");
@@ -43,8 +54,14 @@ type SessionSubscriptionReadiness = {
reject: (reason: unknown) => void;
requestStarted: boolean;
settled: boolean;
attempt: number;
retryTimer: ReturnType<typeof setTimeout> | null;
};
export const SESSION_ENTRY_REQUEST_TIMEOUT_MS = 10000;
export const SESSION_ENTRY_RETRY_DELAY_MS = 1000;
const MAX_SESSION_SUBSCRIPTION_ATTEMPTS = 2;
const DEFAULT_RECONNECT_OPTIONS: Required<ReconnectOptions> = {
enabled: true,
maxAttempts: 10,
@@ -178,7 +195,7 @@ export class WebSocketClient {
return new Promise((resolve, reject) => {
const timeout = setTimeout(() => {
this.pendingRequests.delete(id);
reject(new Error(`WebSocket request timed out: ${action}`));
reject(new WebSocketRequestTimeoutError(action));
}, timeoutMs);
this.pendingRequests.set(id, {
resolve: resolve as (payload: unknown) => void,
@@ -587,6 +604,8 @@ export class WebSocketClient {
reject,
requestStarted: false,
settled: false,
attempt: 0,
retryTimer: null,
};
// subscribeSession() consumers do not await readiness, so handle failures
// while returning the original promise to readiness-aware consumers.
@@ -596,18 +615,37 @@ export class WebSocketClient {
}
private startSessionSubscription(sessionId: string, readiness: SessionSubscriptionReadiness) {
if (this.sessionSubscriptionReadiness.get(sessionId) !== readiness) return;
if (readiness.requestStarted) return;
readiness.requestStarted = true;
void this.request("session.subscribe", { session_id: sessionId }).then(
readiness.attempt += 1;
void this.request(
"session.subscribe",
{ session_id: sessionId },
SESSION_ENTRY_REQUEST_TIMEOUT_MS,
).then(
() => {
if (this.sessionSubscriptionReadiness.get(sessionId) !== readiness) return;
readiness.retryTimer = null;
readiness.settled = true;
readiness.resolve();
},
(error: unknown) => {
if (this.sessionSubscriptionReadiness.get(sessionId) === readiness) {
this.sessionSubscriptionReadiness.delete(sessionId);
if (this.sessionSubscriptionReadiness.get(sessionId) !== readiness) return;
readiness.requestStarted = false;
if (
isWebSocketRequestTimeoutError(error) &&
readiness.attempt < MAX_SESSION_SUBSCRIPTION_ATTEMPTS &&
this.status === "connected" &&
this.socket
) {
readiness.retryTimer = setTimeout(() => {
readiness.retryTimer = null;
this.startSessionSubscription(sessionId, readiness);
}, SESSION_ENTRY_RETRY_DELAY_MS);
return;
}
this.sessionSubscriptionReadiness.delete(sessionId);
readiness.settled = true;
readiness.reject(error);
},
@@ -618,6 +656,10 @@ export class WebSocketClient {
const readiness = this.sessionSubscriptionReadiness.get(sessionId);
if (!readiness) return;
this.sessionSubscriptionReadiness.delete(sessionId);
if (readiness.retryTimer) {
clearTimeout(readiness.retryTimer);
readiness.retryTimer = null;
}
if (!readiness.settled) {
readiness.settled = true;
readiness.reject(new Error("Session subscription released"));
@@ -628,6 +670,10 @@ export class WebSocketClient {
const readinessEntries = [...this.sessionSubscriptionReadiness.entries()];
this.sessionSubscriptionReadiness.clear();
for (const [, readiness] of readinessEntries) {
if (readiness.retryTimer) {
clearTimeout(readiness.retryTimer);
readiness.retryTimer = null;
}
if (readiness.settled) continue;
readiness.settled = true;
readiness.reject(error);
+18
View File
@@ -1,5 +1,23 @@
export type WebSocketRequestErrorDetails = Record<string, unknown>;
/** Error returned when a WebSocket request does not receive a response in time. */
export class WebSocketRequestTimeoutError extends Error {
readonly action: string;
readonly kind = "timeout" as const;
constructor(action: string) {
super(`WebSocket request timed out: ${action}`);
this.name = "WebSocketRequestTimeoutError";
this.action = action;
}
}
export function isWebSocketRequestTimeoutError(
error: unknown,
): error is WebSocketRequestTimeoutError {
return error instanceof WebSocketRequestTimeoutError;
}
/** Error returned when the backend rejects a request with a typed payload. */
export class WebSocketRequestError extends Error {
readonly code?: string;
+5
View File
@@ -1973,6 +1973,11 @@
"sessionRecoveryDetails": "Recovery details",
"sessionRecoveryResumeAttempt": "Resume attempt",
"sessionRecoveryRestoreAttempt": "Workspace restore attempt",
"sessionHistoryRetrying": "Taking longer than usual. Retrying...",
"sessionHistoryUnavailable": "Conversation could not load.",
"sessionHistoryUnknownError": "The history request failed.",
"sessionStatusUnavailable": "Session status is unavailable.",
"sessionStatusUnavailableDetail": "The session remains available. Retry the status check.",
"sessionBootstrapRecoveryTitle": "Session startup needs attention",
"sessionBootstrapRecoverySummary": "The agent could not start. Retry resume or restore the workspace as read-only.",
"sessionBootstrapCauseAuthenticationRequired": "Agent authentication is required.",
+5
View File
@@ -1973,6 +1973,11 @@
"sessionRecoveryDetails": "Ŕēćōvēŕŷ ďēţàĩĺś",
"sessionRecoveryResumeAttempt": "Ŕēśũḿē àţţēḿƥţ",
"sessionRecoveryRestoreAttempt": "Ŵōŕķśƥàćē ŕēśţōŕē àţţēḿƥţ",
"sessionHistoryRetrying": "Ţàķĩńĝ ĺōńĝēŕ ţĥàń ũśũàĺ. Ŕēţŕŷĩńĝ...",
"sessionHistoryUnavailable": "Ćōńvēŕśàţĩōń ćōũĺď ńōţ ĺōàď.",
"sessionHistoryUnknownError": "Ţĥē ĥĩśţōŕŷ ŕēqũēśţ ƒàĩĺēď.",
"sessionStatusUnavailable": "Śēśśĩōń śţàţũś ĩś ũńàvàĩĺàƀĺē.",
"sessionStatusUnavailableDetail": "Ţĥē śēśśĩōń ŕēḿàĩńś àvàĩĺàƀĺē. Ŕēţŕŷ ţĥē śţàţũś ćĥēćķ.",
"sessionBootstrapRecoveryTitle": "Śēśśĩōń śţàŕţũƥ ńēēďś àţţēńţĩōń",
"sessionBootstrapRecoverySummary": "Ţĥē àĝēńţ ćōũĺď ńōţ śţàŕţ. Ŕēţŕŷ ŕēśũḿē ōŕ ŕēśţōŕē ţĥē ŵōŕķśƥàćē àś ŕēàď-ōńĺŷ.",
"sessionBootstrapCauseAuthenticationRequired": "Àĝēńţ àũţĥēńţĩćàţĩōń ĩś ŕēqũĩŕēď.",
+5
View File
@@ -1949,6 +1949,11 @@
"sessionRecoveryDetails": "Detalhes da recuperação",
"sessionRecoveryResumeAttempt": "Tentativa de retomar",
"sessionRecoveryRestoreAttempt": "Tentativa de restaurar o workspace",
"sessionHistoryRetrying": "Está a demorar mais do que o habitual. A tentar novamente...",
"sessionHistoryUnavailable": "Não foi possível carregar a conversa.",
"sessionHistoryUnknownError": "O pedido do histórico falhou.",
"sessionStatusUnavailable": "O estado da sessão não está disponível.",
"sessionStatusUnavailableDetail": "A sessão continua disponível. Tente verificar o estado novamente.",
"sessionBootstrapRecoveryTitle": "O arranque da sessão precisa de atenção",
"sessionBootstrapRecoverySummary": "Não foi possível iniciar o agente. Tente retomar ou restaurar o workspace em modo só de leitura.",
"sessionBootstrapCauseAuthenticationRequired": "É necessária autenticação do agente.",
+5
View File
@@ -1949,6 +1949,11 @@
"sessionRecoveryDetails": "恢复详情",
"sessionRecoveryResumeAttempt": "恢复尝试",
"sessionRecoveryRestoreAttempt": "工作区还原尝试",
"sessionHistoryRetrying": "耗时比平时长。正在重试…",
"sessionHistoryUnavailable": "无法加载对话。",
"sessionHistoryUnknownError": "历史记录请求失败。",
"sessionStatusUnavailable": "会话状态不可用。",
"sessionStatusUnavailableDetail": "会话仍可用。请重试状态检查。",
"sessionBootstrapRecoveryTitle": "会话启动需要处理",
"sessionBootstrapRecoverySummary": "智能体无法启动。请重试恢复,或将工作区还原为只读模式。",
"sessionBootstrapCauseAuthenticationRequired": "需要进行智能体身份验证。",
+5
View File
@@ -1949,6 +1949,11 @@
"sessionRecoveryDetails": "恢復詳情",
"sessionRecoveryResumeAttempt": "恢復嘗試",
"sessionRecoveryRestoreAttempt": "工作區還原嘗試",
"sessionHistoryRetrying": "耗時比平時長。正在重試…",
"sessionHistoryUnavailable": "無法載入對話。",
"sessionHistoryUnknownError": "歷史記錄請求失敗。",
"sessionStatusUnavailable": "工作階段狀態不可用。",
"sessionStatusUnavailableDetail": "工作階段仍可用。請重試狀態檢查。",
"sessionBootstrapRecoveryTitle": "工作階段啓動需要處理",
"sessionBootstrapRecoverySummary": "代理程式無法啓動。請重試恢復,或將工作區還原為只讀模式。",
"sessionBootstrapCauseAuthenticationRequired": "需要進行代理程式身份驗證。",
+5
View File
@@ -1949,6 +1949,11 @@
"sessionRecoveryDetails": "恢復詳情",
"sessionRecoveryResumeAttempt": "恢復嘗試",
"sessionRecoveryRestoreAttempt": "工作區還原嘗試",
"sessionHistoryRetrying": "耗時比平時長。正在重試…",
"sessionHistoryUnavailable": "無法載入對話。",
"sessionHistoryUnknownError": "歷史記錄請求失敗。",
"sessionStatusUnavailable": "工作階段狀態不可用。",
"sessionStatusUnavailableDetail": "工作階段仍可用。請重試狀態檢查。",
"sessionBootstrapRecoveryTitle": "工作階段啟動需要處理",
"sessionBootstrapRecoverySummary": "代理程式無法啟動。請重試恢復,或將工作區還原為只讀模式。",
"sessionBootstrapCauseAuthenticationRequired": "需要進行代理程式身分驗證。",
+183
View File
@@ -0,0 +1,183 @@
---
created: 2026-09-12
status: complete
requirements:
- REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002
- REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-001
system_design:
- ../../specs/platform/system-design/session-subscription-recovery.md
legacy_specs: []
---
# Implementation Plan: Recover delayed session entry
## Overview
Restore chat entry after temporary transport delays, then present accurate loading and recovery states.
Platform owns this work because it owns subscription readiness and shared session recovery.
The [requirement](../../specs/platform/requirements/session-subscription-recovery.md) extends the existing contract.
The [system design](../../specs/platform/system-design/session-subscription-recovery.md) defines timing, ownership, and presentation.
## Evidence and confirmed root cause
On 2026-09-12, task `41227cde-bb02-4738-b9c0-1b776929d68e` received status and subscription responses after 6.9 seconds.
The five-second client deadline had already discarded both requests.
History waited on rejected subscription readiness and never sent its initial request.
Foreground refresh later loaded 100 messages in 351 milliseconds.
The status exception became a generic startup-failure banner; failed history became an empty invitation.
The message-fetch catch also clears cached history, which compounds recovery failures.
Backend registration occurred promptly. The downstream delivery bottleneck remains unconfirmed.
This package repairs the demonstrated timeout, retry, and presentation defects.
It does not claim a backend or network performance fix.
## Scope
### In scope
- Bounded, shared subscription recovery and read-only status retries.
- History retry with preserved cache and truthful initialization state.
- Desktop, phone, and preview feedback with localized recovery controls.
- Regression tests for timing, ownership, mutations, and rendered states.
### Out of scope
- Global timeout increases, automatic mutation retries, and server replay protocols.
- Agent startup policy, archive admission, and unrelated error redesign.
- Proxy tuning or a diagnosis of the original delivery bottleneck.
## Technical approach
Task 01 implements the design's client and hook state. It retains acknowledgement-before-history ordering.
Task 02 connects those outcomes to the shared chat renderer and session feedback, then proves both viewports.
Use typed timeout classification in `lib/ws/request-error.ts`; keep actual backend errors distinct.
Keep registration ownership in `WebSocketClient` and mutation admission in existing session recovery code.
New state belongs to session identity and connection generation, not a global pending boolean.
## ASCII UI preview
UI-01: Open existing task, current failure (desktop and phone).
```text
[!] Couldn't start a session
WebSocket request timed out: task.session.status
[Retry]
No messages yet. Start the conversation!
```
UI-02: Open existing task, proposed desktop chat region.
```text
Loading: (spinner) Loading conversation...
Retrying: (spinner) Taking longer than usual. Retrying...
Exhausted: Conversation could not load. [Retry] [Details v]
Ready: <conversation, or confirmed empty invitation>
```
UI-03: Same task, proposed phone chat region.
```text
+------------------------------------+
| Conversation could not load. |
| [ Retry ] [ Details v ] |
| |
| <cached conversation stays visible>|
+------------------------------------+
| <existing composer and navigation> |
+------------------------------------+
```
UI-04: Status-only exhaustion with a readable transcript.
```text
Session status is unavailable. [Retry] [Details v]
<conversation remains visible and scrollable>
```
Details expands inline with wrapped technical text. Loading and retrying use a
neutral status region. Neither uses a destructive alert or the empty invitation.
Phone actions occupy their own row and have at least 44-pixel targets.
Desktop actions use the normal 28-pixel size. The chat remains the scroll owner;
existing composer placement and safe-area behavior remain unchanged.
These hierarchy and state choices are required; wording and spacing are illustrative.
All text is localized. Covers AC-002.4 through AC-002.6 and AC-002.8 through AC-002.10
(where AC-002 denotes AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002).
## Tests
- `lib/ws/client.test.ts`: seven-second acknowledgement, timeout retry, exhausted budget, shared consumers, last unsubscribe, and reconnect.
- `hooks/domains/session/use-session-messages.test.ts`: automatic history recovery, no request before acknowledgement, preserved cache, empty success, and session-switch cancellation.
- `hooks/domains/session/use-session-subscription-retry.test.ts`: unknown-session retry joins current registration without resetting its budget.
- `hooks/domains/session/use-session-resumption.test.ts`: seven-second status success, bounded retry, immediate permanent rejection, and one mutation after success.
- `components/task/chat/session-entry-feedback.test.tsx` (new): loading, unavailable, cached, confirmed-empty, and disclosure states.
- Existing `components/task/ensure-session-error.test.tsx`: actual launch and profile errors retain their actions.
Task 01 covers AC-002.1 through AC-002.7 and AC-002.10, plus AC-001.1 through AC-001.8.
Task 02 covers all AC-002 criteria through rendering and E2E.
Use fake timers for deterministic unit timing. Record the expected behavioral RED before changing production code.
## E2E tests
New files: `apps/web/e2e/tests/session/session-entry-recovery.spec.ts` (chromium)
and `apps/web/e2e/tests/session/mobile-session-entry-recovery.spec.ts` (mobile-chrome).
Seed an existing stopped session with known persisted messages and use a real WebSocket proxy.
Delay selected subscription/status responses beyond five seconds with `injectLatency`.
Verify automatic chat appearance without reload or tab visibility changes.
Drop a first response to prove retry; drop every attempt to prove bounded exhaustion and explicit recovery.
Assert request counts, no duplicate launch, no stale startup banner, and no false empty invitation.
Also cover cached refresh failure, a successful empty history, and permanent authorization rejection.
On phone, tap Retry and Details, measure targets, and check overflow at phone width and around the 768-pixel boundary.
Reuse `e2e/helpers/session-capabilities.ts` for frame routing and `causal-waits.ts` for event waits.
The new shared helper is `e2e/helpers/session-entry-recovery.ts`.
## Work orders
- [x] [Task 01: Recover entry requests](task-01-entry-recovery.md)
- [x] [Task 02: Present conversation recovery](task-02-recovery-feedback.md)
## Verification results
Implementation complete. Design-package checks and implementation checks passed:
- `python3 scripts/lint-spec-files.test.py`: 36 tests passed.
- `python3 scripts/lint-spec-files.py --all`: passed.
- `git diff --check -- docs/specs docs/plans/session-entry-recovery docs/plans/session-subscription-recovery`: passed.
- `cd apps/web && pnpm exec vitest run lib/ws/client.test.ts hooks/domains/session/use-session-messages.test.ts hooks/domains/session/use-session-subscription-retry.test.ts hooks/domains/session/use-session-resumption.test.ts hooks/domains/session/use-session-resumption.archive.test.ts hooks/domains/session/use-session-message-fetch.test.ts components/task/chat/session-entry-feedback.test.tsx components/task/ensure-session-error.test.tsx components/task/chat/message-list-shared.test.tsx`: 170 tests passed across 9 files.
- `cd apps/web && pnpm run typecheck`: passed.
- The zero-warning ESLint commands listed in both work-order Results: passed.
- `cd apps/web && pnpm run i18n:zh-hant`: passed; generated the Traditional Chinese pair.
- `cd apps/web && pnpm run i18n:check`: passed; all five complete catalogs and pseudo locale are synchronized.
- `cd apps/web && pnpm run i18n:ratchet`: passed; no new untranslated copy.
- `cd apps/web && pnpm e2e:run --project chromium tests/session/session-entry-recovery.spec.ts tests/session/session-resume-recovery.spec.ts`: 5 passed, including the production web build.
- `cd apps/web && pnpm e2e:run --project mobile-chrome tests/session/mobile-session-entry-recovery.spec.ts tests/session/mobile-session-resume-recovery.spec.ts`: 2 passed, including phone touch geometry and overflow checks.
- `git diff --check`: passed.
- Public documentation impact: no public docs change is required for this recovery behavior.
Review remediation validation:
- Status restoration and backend error-payload regressions passed, including the assertion that a
permanent status error does not launch a session.
- Deferred history generation regressions passed for terminal fetch, session A to B navigation,
session A to B to A re-entry, unmount-safe finalization, and fresh manual Retry.
- The focused recovery suite passed 170 tests across 9 files after review remediation. It
includes truthful workspace-restore failures, permanent status error payloads, terminal and
manual history refreshes, A-to-B and A-to-B-to-A deferred races, stale loading finalization,
and the single live status retry path. Typecheck, zero-warning ESLint, and formatting checks
passed.
- Follow-up CI remediation passed the deferred cached-refresh rerender regression, the desktop
unread-divider suite (5 tests), and the mobile unread-divider suite (3 tests). Cached refresh
completion now remains valid across same-session message rerenders and is still invalidated by
navigation or unmount.
## Risks
- Retries can duplicate launches if they wrap `processResumeStatus` instead of its status read.
- Per-consumer retry loops can amplify traffic or replace another consumer's readiness promise.
- Clearing loading flags before recovery finishes can reintroduce the false empty state.
- A longer entry deadline improves tolerance but does not remove the original delivery delay.
## Related package
[Original ordering repair](../session-subscription-recovery/plan.md) remains implemented.
Its historical checks do not count as validation of this package.
@@ -0,0 +1,135 @@
---
id: "01-entry-recovery"
title: "Recover entry requests"
status: done
wave: 1
depends_on: []
plan: "plan.md"
requirements:
- REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002
- REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-001
acceptance_criteria:
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.1
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.2
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.3
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.4
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.5
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.6
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.7
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.10
system_design:
- ../../specs/platform/system-design/session-subscription-recovery.md
---
# Task 01: Recover entry requests
## Summary
Implement bounded entry recovery and truthful history state in the existing client and hooks.
Preserve shared registration ownership, cached messages, and mutation admission.
## In scope
- Typed timeout errors and entry-specific deadlines.
- Shared subscription retry and independently bounded status/history reads.
- Hook recovery outcome and manual retry APIs for Task 02.
- Session/connection guards, cleanup, cache preservation, and permanent-error handling.
## Out of scope
Rendered UI, localization, backend changes, and automatic mutation retries.
## Acceptance
- A seven-second response succeeds; dropped responses retry within the design's limits without a visibility event.
- Failed history preserves cached messages; only a successful snapshot marks history initialized.
- Concurrent consumers share recovery; obsolete requests cannot act, and a successful status runs resume processing at most once.
## Verification
From the repository root, install dependencies once if this worktree lacks them:
```bash
(cd apps && pnpm install --frozen-lockfile)
```
```bash
(cd apps/web && pnpm exec vitest run lib/ws/client.test.ts hooks/domains/session/use-session-messages.test.ts hooks/domains/session/use-session-subscription-retry.test.ts hooks/domains/session/use-session-resumption.test.ts hooks/domains/session/use-session-resumption.archive.test.ts)
(cd apps/web && pnpm run typecheck)
(cd apps/web && pnpm exec eslint --max-warnings 0 lib/ws/client.ts lib/ws/request-error.ts hooks/domains/session/use-session-messages.ts hooks/domains/session/use-session-message-fetch.ts hooks/domains/session/use-message-fetch-state.ts hooks/domains/session/use-session-subscription-retry.ts hooks/domains/session/use-session-resumption.ts)
git diff --check
```
Add a behavioral RED named `recovers timed out registration without a visibility change` before implementation.
Use existing FakeWebSocket and hook test mocks; add fake-timer assertions for exhaustion and late responses.
Run any extracted new helper tests explicitly and include their paths in Results.
## Files likely touched
- `apps/web/lib/ws/client.ts`, `client.test.ts`, and `request-error.ts`
- `apps/web/hooks/domains/session/use-session-messages.ts` and its tests
- `apps/web/hooks/domains/session/use-session-message-fetch.ts`
- `apps/web/hooks/domains/session/use-message-fetch-state.ts`
- `apps/web/hooks/domains/session/use-session-subscription-retry.ts` and its tests
- `apps/web/hooks/domains/session/use-session-resumption.ts` and its tests
Paths without full directories above refer to the preceding explicit directory.
## Dependencies
None.
## Risks
Ref-count churn, duplicate mutation admission, timer leaks, and stale response writes.
## Parallelism
`sequential`
## Inputs
- Requirement AC-002.1 through AC-002.7 and AC-002.10; preserve all AC-001 criteria.
- System design: Registration ordering, Entry recovery, and History state.
- Existing client fake socket and session hook tests.
## Results
Implemented typed WebSocket timeout classification, the 10-second session-entry deadline, and
bounded shared subscription recovery. Status and history reads now retry typed timeouts within
the design budget, while permanent errors remain distinct. History retains cached rows until a
successful snapshot and exposes explicit recovery state for the presentation task.
Verification:
- `cd apps/web && pnpm exec vitest run lib/ws/client.test.ts hooks/domains/session/use-session-messages.test.ts hooks/domains/session/use-session-subscription-retry.test.ts hooks/domains/session/use-session-resumption.test.ts hooks/domains/session/use-session-resumption.archive.test.ts hooks/domains/session/use-session-message-fetch.test.ts components/task/chat/session-entry-feedback.test.tsx components/task/ensure-session-error.test.tsx components/task/chat/message-list-shared.test.tsx` — 170 tests passed across 9 files.
- `cd apps/web && pnpm run typecheck` — passed.
- `cd apps/web && pnpm exec eslint --max-warnings 0 lib/ws/client.ts lib/ws/request-error.ts hooks/domains/session/use-session-messages.ts hooks/domains/session/use-session-message-fetch.ts hooks/domains/session/use-message-fetch-state.ts hooks/domains/session/use-session-subscription-retry.ts hooks/domains/session/use-session-resumption.ts` — passed.
- `git diff --check` — passed.
The implementation preserves acknowledgement-before-history ordering, cancels obsolete retry
timers, and does not retry session mutations.
Review remediation (complete):
- Status request failures are classified before resume processing. Workspace-restore and launch
failures keep their launch recovery action, while status responses with `error` or an archive
outcome remain visible and never start a session.
- History feedback now carries a session and connection generation through subscription,
terminal-state, entry, visibility, settle, running-backfill, and manual-retry fetches. Stale
completions release the original session's store bookkeeping without changing the active hook
state.
- Manual history Retry clears the settled hydration marker before fetching, so it always requests a
fresh snapshot while preserving cached rows until that snapshot succeeds.
- Added regressions for workspace-restore failure classification, timeout-then-error status
responses, A-to-B and A-to-B-to-A deferred history races, terminal-state fetches, and fresh
manual history retry.
- Focused review-remediation validation: 170 tests passed across 9 files. It covers
environment-scoped workspace-restore errors and launch retry behavior, permanent status error
payloads after a timeout, generation-fenced terminal and manual history fetches, settled
request-cache refreshes, and local loading cleanup for stale fetches. Typecheck, zero-warning
ESLint, and formatting checks passed.
- Follow-up CI remediation keeps cached-entry refresh finalization on the session/connection
generation. A live message that rerenders the entry while its refresh is pending can no longer
strand the loading flag or suppress initial divider placement. The deferred regression passed,
as did the desktop unread-divider suite (5 tests) and mobile unread-divider suite (3 tests).
@@ -0,0 +1,195 @@
---
id: "02-recovery-feedback"
title: "Present conversation recovery"
status: completed
wave: 2
depends_on: ["01-entry-recovery"]
plan: "plan.md"
requirements:
- REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002
- REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-001
acceptance_criteria:
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.1
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.2
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.3
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.4
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.5
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.6
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.7
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.8
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.9
- AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.10
system_design:
- ../../specs/platform/system-design/session-subscription-recovery.md
---
# Task 02: Present conversation recovery
## Summary
Connect entry recovery to compact chat feedback on desktop, phone, and preview.
Prove the reported delay and automatic recovery with isolated browser tests.
## In scope
- Explicit history loading/error/empty rendering and retry controls.
- Typed status-check feedback with collapsed technical details.
- Localized copy, accessible progress, phone targets, and request-aware browser tests.
## Out of scope
Changes to provider failures, launch policy, server transport, or global alert styling.
## Acceptance
- Unknown history never renders the empty invitation; cached messages remain visible after errors.
- Status timeouts use accurate compact feedback, and successful recovery clears that feedback without changing launch-error actions.
- Desktop and phone recover without reload; Retry and Details work through visible controls with correct touch geometry.
## ASCII UI preview
UI-01: Open existing task, current failure (desktop and phone).
```text
[!] Couldn't start a session
WebSocket request timed out: task.session.status
[Retry]
No messages yet. Start the conversation!
```
UI-02: Open existing task, proposed desktop chat region.
```text
Loading: (spinner) Loading conversation...
Retrying: (spinner) Taking longer than usual. Retrying...
Exhausted: Conversation could not load. [Retry] [Details v]
Ready: <conversation, or confirmed empty invitation>
```
UI-03: Same task, proposed phone chat region.
```text
+------------------------------------+
| Conversation could not load. |
| [ Retry ] [ Details v ] |
| |
| <cached conversation stays visible>|
+------------------------------------+
| <existing composer and navigation> |
+------------------------------------+
```
UI-04: Status-only exhaustion with a readable transcript.
```text
Session status is unavailable. [Retry] [Details v]
<conversation remains visible and scrollable>
```
Details expands inline with wrapped technical text. Loading and retrying use a
neutral status region. Neither uses a destructive alert or the empty invitation.
Phone actions occupy their own row and have at least 44-pixel targets.
Desktop actions use the normal 28-pixel size. The chat remains the scroll owner;
existing composer placement and safe-area behavior remain unchanged.
These hierarchy and state choices are required; wording and spacing are illustrative.
All text is localized. Covers AC-002.4 through AC-002.6 and AC-002.8 through AC-002.10
(where AC-002 denotes AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002).
Full combined preview: [plan](plan.md#ascii-ui-preview).
## Verification
Write component RED for the false empty state before rendering changes.
Use a controlled response barrier for E2E RED where feasible; record any fixture limitation.
```bash
(cd apps/web && pnpm exec vitest run components/task/chat/session-entry-feedback.test.tsx components/task/ensure-session-error.test.tsx hooks/domains/session/use-session-messages.test.ts hooks/domains/session/use-session-resumption.test.ts)
(cd apps/web && pnpm run typecheck)
(cd apps/web && pnpm exec eslint --max-warnings 0 components/task/ensure-session-error.tsx components/task/task-page-inner.tsx components/task/preview-session-tabs.tsx components/task/chat components/task/task-chat-panel.tsx e2e/helpers/session-entry-recovery.ts e2e/tests/session/session-entry-recovery.spec.ts e2e/tests/session/mobile-session-entry-recovery.spec.ts)
(cd apps/web && pnpm run i18n:zh-hant)
(cd apps/web && pnpm run i18n:check)
(cd apps/web && pnpm run i18n:ratchet)
(cd apps/web && pnpm e2e:run --project chromium tests/session/session-entry-recovery.spec.ts tests/session/session-resume-recovery.spec.ts)
(cd apps/web && pnpm e2e:run --project mobile-chrome tests/session/mobile-session-entry-recovery.spec.ts tests/session/mobile-session-resume-recovery.spec.ts)
git diff --check
```
Managed E2E commands rebuild production assets and use isolated backends.
Run desktop and mobile commands sequentially. Inspect their screenshots against UI-02 through UI-04.
The desktop suite also exercises preview; phone uses the existing direct task route.
Capture request IDs/actions without payload contents. Preserve unrelated frames in the proxy.
## Files likely touched
- `apps/web/components/task/ensure-session-error.tsx` and its tests
- `apps/web/components/task/task-page-inner.tsx`
- `apps/web/components/task/preview-session-tabs.tsx`
- `apps/web/components/task/chat/use-chat-panel-state.ts`
- `apps/web/components/task/chat/message-list-shared.tsx`
- `apps/web/components/task/task-chat-panel.tsx`
- New `apps/web/components/task/chat/session-entry-feedback.test.tsx`
- `apps/web/src/locales/{en,pt-pt,zh-cn,zh-hk,zh-tw}/task.json`
- New `apps/web/e2e/helpers/session-entry-recovery.ts`
- New `apps/web/e2e/tests/session/session-entry-recovery.spec.ts`
- New `apps/web/e2e/tests/session/mobile-session-entry-recovery.spec.ts`
## Dependencies
Task 01 supplies typed recovery state and bounded retry actions.
## Risks
Duplicate feedback from page and transcript consumers, desktop sizing leaking to phone, and untranslated diagnostic copy.
## Parallelism
`sequential`
## Inputs
- Requirement AC-002.1 through AC-002.10 and system design Presentation section.
- Task 01 hook results; existing shared transcript loading renderer.
- Mobile exemplar: `components/task/task-layout.tsx` dedicated phone chat composition.
- Browser routing exemplar: `e2e/helpers/session-capabilities.ts`.
## Results
Connected the recovery outcomes to localized desktop, phone, and preview feedback. History now
has loading, retrying, ready, and unavailable states; cached rows remain visible during failed
refreshes; and the empty invitation renders only after a confirmed empty snapshot. Status-only
failures use a compact neutral notice with collapsed technical details and a retry action that
does not launch the agent.
Verification:
- `cd apps/web && pnpm exec vitest run lib/ws/client.test.ts hooks/domains/session/use-session-messages.test.ts hooks/domains/session/use-session-subscription-retry.test.ts hooks/domains/session/use-session-resumption.test.ts hooks/domains/session/use-session-resumption.archive.test.ts hooks/domains/session/use-session-message-fetch.test.ts components/task/chat/session-entry-feedback.test.tsx components/task/ensure-session-error.test.tsx components/task/chat/message-list-shared.test.tsx` — 170 tests passed across 9 files.
- `cd apps/web && pnpm run typecheck` — passed.
- `cd apps/web && pnpm exec eslint --max-warnings 0 components/task/ensure-session-error.tsx components/task/task-page-inner.tsx components/task/preview-session-tabs.tsx components/task/chat e2e/helpers/session-entry-recovery.ts e2e/tests/session/session-entry-recovery.spec.ts e2e/tests/session/mobile-session-entry-recovery.spec.ts` — passed.
- `cd apps/web && pnpm run i18n:zh-hant` — passed.
- `cd apps/web && pnpm run i18n:check` — passed; English, Portuguese, Simplified Chinese, Traditional Chinese, and pseudo catalogs are synchronized.
- `cd apps/web && pnpm run i18n:ratchet` — passed.
- `cd apps/web && pnpm e2e:run --project chromium tests/session/session-entry-recovery.spec.ts tests/session/session-resume-recovery.spec.ts` — 5 passed; the managed run built the backend and production web assets.
- `cd apps/web && pnpm e2e:run --project mobile-chrome tests/session/mobile-session-entry-recovery.spec.ts tests/session/mobile-session-resume-recovery.spec.ts` — 2 passed.
- `git diff --check` — passed.
The request-aware WebSocket proxy records action and request IDs while preserving unrelated frames.
The phone scenario verifies 44-pixel-class controls and no horizontal overflow.
Review remediation (complete):
- The existing recovery feedback keeps launch and workspace-restore failures on the launch retry
path, while status-only failures remain on the mutation-free status retry path.
- Backend status error payloads remain visible through the shared feedback components instead of
being cleared as successful status refreshes.
- Deferred history race coverage confirms that old session work cannot overwrite feedback for the
current desktop or phone conversation, including manual retry after re-entry.
- History loading finalization is generation-fenced on navigation and connection changes, while
the original session's request bookkeeping is still released. Settled message requests cannot
satisfy a later terminal refresh, so explicit Retry sends a new snapshot request.
- The loading spinner is hidden from the live status subtree so screen readers receive one
progress announcement. The status-retry E2E records the launch count before and after Retry
and verifies that status recovery stays mutation-free.
- Follow-up CI validation confirmed that cached history refreshes release their pending state after
a live message rerender, so divider placement remains correct on desktop and phone. The exact
unread-divider regressions passed with 5 desktop and 3 mobile tests.
@@ -136,3 +136,8 @@ contract, so they are intentionally one sequential task.
- The race-specific reload fallback can be removed; the general bounded
`waitForLoad` reload remains for unrelated startup/SSR hydration failures.
## Follow-up package
The [delayed-entry package](../session-entry-recovery/plan.md) extends recovery after a registration timeout.
The completed ordering work and its recorded results remain unchanged.
@@ -28,49 +28,27 @@ A fast task session can change state or persist a reply while the browser is wai
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-001.7:** **GIVEN** one client-side consumer is already registering a session and a second consumer mounts, **WHEN** the second consumer requests hydration, **THEN** it waits for the shared registration acknowledgement and does not race an earlier `message.list` request ahead of registration.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-001.8:** **GIVEN** a session subscription is restored after a WebSocket reconnect, **WHEN** the restored registration is acknowledged, **THEN** message reconciliation can run against the restored subscription and later events are delivered normally.
## Migrated source detail
### REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002: Recover session entry from temporary delays
## Why
**Intent:** Users can open an existing conversation during a temporary connection delay without a false startup failure or lost history.
A fast task session can change state or persist a reply while the browser is
waiting for the backend to register its `session.subscribe` request. The live
notification is then unavailable to that client, leaving the task looking busy
or the transcript missing a reply until the user reloads the page.
#### Acceptance criteria
## What
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.1:** A successful subscription or status response delayed by seven seconds shall complete session entry without a startup-failure alert.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.2:** After a temporary entry timeout, the visible session shall retry automatically without a reload, navigation, or visibility change.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.3:** On a continuously connected socket, unsuccessful automatic recovery shall stop within 45 seconds. The user shall then have an explicit Retry action.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.4:** Until history loads successfully, an uncached conversation shall show loading, retrying, or unavailable feedback. It shall never show the empty-conversation invitation.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.5:** A failed refresh shall preserve cached messages, drafts, and scroll position. Successful recovery shall clear its stale feedback.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.6:** A status timeout shall identify a status-check problem, not a failed session start. Technical details shall remain collapsed by default.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.7:** Recovery shall not duplicate agent launches or bypass archive, permission, or manual-start restrictions. Changing sessions shall cancel obsolete recovery effects.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.8:** Desktop, phone, and task preview shall expose equivalent recovery outcomes. Feedback shall preserve chat access and keyboard focus without horizontal overflow.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.9:** Phone and coarse-pointer recovery controls shall have touch targets of at least 44 pixels. Progress shall have a localized status announcement.
- **AC-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002.10:** Permission and missing-session failures shall stop automatic retries. A successful empty history response shall still show the normal empty-conversation invitation.
- A session subscription exposes a readiness point that resolves only after the
backend has acknowledged registration.
- Transcript reconciliation for a newly mounted or reconnected session starts
after that readiness point, so the persisted message view covers the interval
before registration and live notifications cover the interval after it.
- Ref-counted consumers share an in-flight registration readiness point and do
not unregister one another's subscription.
- Reconnect and delayed-session retry paths use the same acknowledgement-aware
registration behavior.
- A failed or disconnected registration does not create an unhandled client
rejection or leave a stale readiness promise that prevents the next
connection from hydrating the session.
## Implementation plans
## Scenarios
- **GIVEN** a browser sends `session.subscribe` while a session state change or
reply is persisted before the backend registers the subscription, **WHEN** the
backend acknowledges the subscription and the client reconciles messages,
**THEN** the client displays the authoritative state and persisted reply
without a page reload.
- **GIVEN** one client-side consumer is already registering a session and a
second consumer mounts, **WHEN** the second consumer requests hydration,
**THEN** it waits for the shared registration acknowledgement and does not
race an earlier `message.list` request ahead of registration.
- **GIVEN** a session subscription is restored after a WebSocket reconnect,
**WHEN** the restored registration is acknowledged, **THEN** message
reconciliation can run against the restored subscription and later events are
delivered normally.
- **GIVEN** the registration fails or the socket disconnects before the
acknowledgement, **WHEN** the client retries on the next connected socket,
**THEN** the retry can establish a fresh readiness point and no stale promise
or unhandled rejection blocks hydration.
- [Acknowledgement ordering](../../../plans/session-subscription-recovery/plan.md)
- [Delayed session entry](../../../plans/session-entry-recovery/plan.md)
## Out of scope
@@ -0,0 +1,95 @@
---
status: draft
system: platform
requirements:
- REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-001
- REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002
---
# Session subscription recovery
## Ownership and mapping
Platform owns shared session subscription readiness and recovery. This design extends that contract across chat, task preview, and session-status feedback.
| Requirement | Design |
| --- | --- |
| REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-001 | Registration ordering and ownership |
| REQ-PLATFORM-SESSION-SUBSCRIPTION-RECOVERY-002 | Entry recovery, history state, and presentation |
## Registration ordering and ownership
The gateway registers membership before acknowledging `session.subscribe` in `apps/backend/internal/gateway/websocket/client.go`.
The frontend `WebSocketClient` owns reference counts and a shared readiness promise per session and connection generation.
`fetchAndStoreMessages` in `use-session-messages.ts` waits for readiness before requesting history.
This ordering remains mandatory. Recovery must not fetch history ahead of registration.
## Entry recovery
Use explicit 10-second deadlines for subscription registration and entry status checks. Keep the general request default unchanged.
Add a typed timeout error in `apps/web/lib/ws/request-error.ts`, retaining its existing diagnostic message.
Only this typed timeout permits the new automatic retry. Backend rejection codes retain their current meaning.
The client owns subscription retries within one shared readiness promise: two attempts, with one second between attempts.
Do not increase reference counts for a retry. Concurrent consumers and unknown-session retries join an active registration episode.
Only the current connection generation can settle readiness. Last unsubscribe cancels its timer and invalidates pending results.
A disconnect cancels the episode; existing reconnect registration starts a fresh generation after reconnection.
Do not queue episode retries while disconnected. Existing inactive-thread subscription limits remain unchanged.
The resumption hook separately retries only the `task.session.status` request with the same two-attempt policy.
Keep `processResumeStatus` outside that retry operation. It runs once after a current successful status response.
Never wrap resume, restore, launch, or prompt mutations in a retry loop.
Preserve archive-state checks, request-identity guards, and existing resume eligibility policy.
After registration, history uses its existing 10-second request deadline and at most one additional timeout retry after one second.
A continuously connected initial episode therefore takes at most 42 seconds: 21 for registration and 21 for history.
Status progresses independently and takes at most 21 seconds. Automatic failure feedback appears within the 45-second requirement.
Use timers with cleanup and fake-timer tests. Background throttling is outside the continuously visible timing guarantee.
Foreground refresh and reconnect remain recovery triggers, but they join an active episode instead of resetting its budget.
After exhaustion, explicit Retry starts a new bounded episode. Disable it while that episode is active.
## History state
Extend the existing message hooks with explicit load outcome and retry action: loading, retrying, ready, or unavailable.
Keep this state associated with session identity and hydration generation. Share in-flight fetches across mounted consumers.
The hook result supplies the state to `use-chat-panel-state.ts`, `task-chat-panel.tsx`, and the shared message-list components.
A successful snapshot alone establishes history initialization, including a successful empty snapshot.
`doFetchMessages` in `use-session-message-fetch.ts` must not clear messages on rejection or mark failed hydration as successful.
Preserve message reconciliation, prompt-history independence, draft state, and scroll anchoring.
An obsolete response cannot overwrite another session or clear a newer episode's error.
Partial live messages remain visible while initial reconciliation is pending.
## Presentation
Use the existing dedicated phone chat layout in `components/task/task-layout.tsx` as the mobile composition exemplar.
Reuse the inline loading region in `components/task/chat/message-list-shared.tsx`.
History feedback lives inside the chat region. Cached messages remain below a compact notice during recovery.
The transcript remains the scroll owner. Do not add an overlay, drawer, or separate nested scroller for short feedback.
The existing dynamic viewport and composer safe-area behavior remain intact.
Use localized copy such as `Loading conversation...`, `Taking longer than usual. Retrying...`, and `Conversation could not load.`
For exhausted status checks, use `Session status is unavailable.` with Retry and a collapsed Details disclosure.
Route entry-status feedback through a typed failure kind in `SessionRecoveryFeedback`, including task page and preview consumers.
Do not globally rename actual launch failures in `describeEnsureError` or suppress provider, profile, and workspace recovery actions.
A status-only failure must not replace the transcript loading result or hide usable cached chat.
When both fail, show one history notice in chat and retain status details without a duplicate large banner.
Desktop actions use the normal 28-pixel control size. Phone or coarse-pointer actions use at least 44-pixel targets.
Phone notices place actions below the message; desktop notices keep actions on the same row when space permits.
Use one polite `role="status"` per visible operation and keep Retry's accessible name stable.
Details use a keyboard-accessible disclosure with wrapped text. Do not move focus on automatic progress.
All new copy uses existing i18n catalogs in five languages, with generated Traditional Chinese variants.
## Persistence, security, and observability
No database, protocol, public API, or permission changes are required.
Existing server-side authorization and mutation admission remain authoritative.
Retries preserve typed backend errors and expose only existing transport diagnostics in Details.
Tests correlate request action and ID; they do not record chat payloads or credentials.
The cause of the observed seven-second delivery delay is unconfirmed. This design does not claim to remove that underlying latency.
## Implementation plans
- [Original ordering repair](../../../plans/session-subscription-recovery/plan.md)
- [Delayed session entry](../../../plans/session-entry-recovery/plan.md)