diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index f0188af478c0..f91ae1c06375 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -39,6 +39,7 @@ import { submitCodexFeedback, type CodexFeedbackSubmission, } from "@t3tools/client-runtime/state/threads"; +import { environmentThreadStreamHealth } from "@t3tools/client-runtime/state/threadState"; import { parseScopedThreadKey, scopedThreadKey, @@ -2314,6 +2315,18 @@ function ChatViewContent(props: ChatViewProps) { ); const selectedProvider: ProviderDriverKind = lockedProvider ?? unlockedSelectedProvider; const phase = derivePhase(activeThread?.session ?? null); + // A running turn may only read as live work while the stream that would deliver + // the completion is attached; deleted navigates away via the route effect. + // Cold opens read Connecting until first attach — honest (nothing can deliver + // yet) and consistent with the sidebar vocabulary. Draft routes deliberately + // keep the detail-state hook empty; a promoted thread's progress arrives over + // the command channel until navigation, so only server routes carry a + // meaningful non-live signal. + const threadStreamHealth = environmentThreadStreamHealth(routeThreadState); + const streamHealthWhileWorking = + routeKind === "server" && threadStreamHealth !== "live" && threadStreamHealth !== "deleted" + ? (threadStreamHealth as "connecting" | "detached") + : null; const threadActivities = activeThread?.activities ?? EMPTY_ACTIVITIES; const activeContextWindow = useMemo( () => deriveLatestContextWindowSnapshot(threadActivities), @@ -6916,6 +6929,7 @@ function ChatViewContent(props: ChatViewProps) { key={activeThread.id} isWorking={isWorking} workingStepLabel={workingStepLabel} + streamHealth={phase === "running" ? streamHealthWhileWorking : null} activeTurnStartedAt={activeWorkStartedAt} listRef={legendListRef} timelineEntries={timelineEntries} diff --git a/apps/web/src/components/LegacySidebar.tsx b/apps/web/src/components/LegacySidebar.tsx index 6d6e26c18e35..91189f2d6128 100644 --- a/apps/web/src/components/LegacySidebar.tsx +++ b/apps/web/src/components/LegacySidebar.tsx @@ -115,6 +115,7 @@ import { useDesktopUpdateState } from "../state/desktopUpdate"; import { useThreadActions } from "../hooks/useThreadActions"; import { projectEnvironment } from "../state/projects"; import { useEnvironmentQuery } from "../state/query"; +import { useShellStreamHealth, useShellStreamHealthForEnvironments } from "../state/shell"; import { threadEnvironment, useEnvironmentThread } from "../state/threads"; import { vcsEnvironment } from "../state/vcs"; import { useEnvironment, useEnvironments, usePrimaryEnvironmentId } from "../state/environments"; @@ -452,11 +453,13 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr ); const isThreadRunning = thread.session?.status === "running" && thread.session.activeTurnId != null; + const shellStreamHealth = useShellStreamHealth(thread.environmentId); const threadStatus = resolveThreadStatusPill({ thread: { ...thread, lastVisitedAt, }, + streamHealth: shellStreamHealth, }); const linkedPullRequestStatus = useLinkedThreadPullRequest( thread.environmentId, @@ -1267,6 +1270,11 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec } return counts; }, [memberProjectByScopedKey, project.memberProjects, projectThreads]); + const memberEnvironmentIds = useMemo( + () => [...new Set(projectThreads.map((thread) => thread.environmentId))], + [projectThreads], + ); + const memberHealths = useShellStreamHealthForEnvironments(memberEnvironmentIds); const { projectStatus, visibleProjectThreads, orderedProjectThreadKeys } = useMemo(() => { const lastVisitedAtByThreadKey = new Map( @@ -1284,6 +1292,7 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec ...thread, ...(lastVisitedAt !== null && lastVisitedAt !== undefined ? { lastVisitedAt } : {}), }, + streamHealth: memberHealths[thread.environmentId] ?? "connecting", }); }; const visibleProjectThreads = sortThreads( @@ -1300,7 +1309,7 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec projectStatus, visibleProjectThreads, }; - }, [projectThreads, threadLastVisitedAts, threadSortOrder]); + }, [memberHealths, projectThreads, threadLastVisitedAts, threadSortOrder]); const pinnedCollapsedThread = useMemo(() => { const activeThreadKey = activeRouteThreadKey ?? undefined; if (!activeThreadKey || projectExpanded) { @@ -1336,6 +1345,7 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec ...thread, ...(lastVisitedAt !== null && lastVisitedAt !== undefined ? { lastVisitedAt } : {}), }, + streamHealth: memberHealths[thread.environmentId] ?? "connecting", }); }; const hasOverflowingThreads = visibleProjectThreads.length > sidebarThreadPreviewCount; @@ -1371,6 +1381,7 @@ const SidebarProjectItem = memo(function SidebarProjectItem(props: SidebarProjec pinnedCollapsedThread, projectExpanded, projectThreads, + memberHealths, sidebarThreadPreviewCount, threadLastVisitedAts, visibleProjectThreads, diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index ba75f2eaaf54..e94bee1792ac 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -1125,6 +1125,54 @@ describe("resolveThreadStatusPill", () => { ).toMatchObject({ label: "Working", pulse: true }); }); + it("shows Reconnecting instead of Working when the shell stream is detached", () => { + expect( + resolveThreadStatusPill({ + thread: baseThread, + streamHealth: "detached", + }), + ).toMatchObject({ label: "Reconnecting", pulse: true }); + }); + + it("shows Reconnecting for detached background liveness too, not only running sessions", () => { + expect( + resolveThreadStatusPill({ + thread: { ...baseThread, session: null, backgroundLiveness: "working" }, + streamHealth: "detached", + }), + ).toMatchObject({ label: "Reconnecting", pulse: true }); + expect( + resolveThreadStatusPill({ + thread: { ...baseThread, session: null, backgroundLiveness: "working" }, + }), + ).toMatchObject({ label: "Working", pulse: true }); + }); + + it("keeps Working when the shell stream is live (default)", () => { + expect(resolveThreadStatusPill({ thread: baseThread })).toMatchObject({ + label: "Working", + pulse: true, + }); + }); + + it("shows Connecting while the shell stream has never attached", () => { + expect( + resolveThreadStatusPill({ + thread: baseThread, + streamHealth: "connecting", + }), + ).toMatchObject({ label: "Connecting", pulse: true }); + }); + + it("does not detach non-working pills", () => { + expect( + resolveThreadStatusPill({ + thread: { ...baseThread, hasPendingApprovals: true }, + streamHealth: "detached", + }), + ).toMatchObject({ label: "Pending Approval" }); + }); + it("shows plan ready when a settled plan turn has a proposed plan ready for follow-up", () => { expect( resolveThreadStatusPill({ diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index 8067fdd59c15..8b851116b2f7 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -1,6 +1,7 @@ import * as React from "react"; import { defaultAnimateLayoutChanges, type AnimateLayoutChanges } from "@dnd-kit/sortable"; import type { ContextMenuItem } from "@t3tools/contracts"; +import type { ShellStreamHealth } from "@t3tools/client-runtime/state/shell"; import type { SidebarProjectSortOrder, SidebarThreadSortOrder } from "@t3tools/contracts/settings"; import { getThreadSortTimestamp, @@ -127,6 +128,7 @@ export function buildBulkTitleRegenerationContextMenuItem(input: { export interface ThreadStatusPill { label: | "Working" + | "Reconnecting" | "Monitoring" | "Connecting" | "Completed" @@ -145,6 +147,7 @@ const THREAD_STATUS_PRIORITY: Record = { "Pending Approval": 6, "Awaiting Input": 5, Working: 4, + Reconnecting: 4, Connecting: 4, "Plan Ready": 3, Monitoring: 2, @@ -644,8 +647,10 @@ export function formatWorkingDurationLabel(elapsedMs: number): string { export function resolveThreadStatusPill(input: { thread: ThreadStatusInput; + /** Shell stream health for this thread's environment (see useShellStreamHealth). */ + streamHealth?: ShellStreamHealth; }): ThreadStatusPill | null { - const { thread } = input; + const { thread, streamHealth = "live" } = input; if (thread.hasPendingApprovals) { return { @@ -665,13 +670,36 @@ export function resolveThreadStatusPill(input: { }; } - if (thread.session?.status === "running") { + // In-flight work may only read as healthy while the event stream that would + // deliver its completion is alive. Pre-attach reads as Connecting (same + // vocabulary as a starting session); a dropped stream reads Reconnecting. + const workingPill = (): ThreadStatusPill => { + if (streamHealth === "detached") { + return { + label: "Reconnecting", + colorClass: "text-sky-600 dark:text-sky-300/80", + dotClass: "bg-sky-500 dark:bg-sky-300/80", + pulse: true, + }; + } + if (streamHealth === "connecting") { + return { + label: "Connecting", + colorClass: "text-sky-600 dark:text-sky-300/80", + dotClass: "bg-sky-500 dark:bg-sky-300/80", + pulse: true, + }; + } return { label: "Working", colorClass: "text-sky-600 dark:text-sky-300/80", dotClass: "bg-sky-500 dark:bg-sky-300/80", pulse: true, }; + }; + + if (thread.session?.status === "running") { + return workingPill(); } if (thread.session?.status === "starting") { @@ -704,12 +732,7 @@ export function resolveThreadStatusPill(input: { // loops (a parent agent babysitting a PR, tailing checks) with no other // live work. Same recede treatment as Working per inbox-zero. if (thread.backgroundLiveness === "working") { - return { - label: "Working", - colorClass: "text-sky-600 dark:text-sky-300/80", - dotClass: "bg-sky-500 dark:bg-sky-300/80", - pulse: true, - }; + return workingPill(); } if (thread.backgroundLiveness === "monitoring") { diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 63681a54813a..fe7e61c63249 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -111,6 +111,7 @@ import { useProjects, useThreadShells } from "../state/entities"; import { environmentServerConfigsAtom, primaryServerKeybindingsAtom } from "../state/server"; import { vcsEnvironment } from "../state/vcs"; import { threadEnvironment } from "../state/threads"; +import { useShellStreamHealth } from "../state/shell"; import { useEnvironmentQuery } from "../state/query"; import { useAtomCommand } from "../state/use-atom-command"; import { @@ -815,6 +816,14 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { // switching sidebars must not light up every historical thread as unread. const isUnread = hasUnseenCompletion({ ...thread, lastVisitedAt }); const status = resolveSidebarThreadStatus(thread); + const shellStreamHealth = useShellStreamHealth(thread.environmentId); + const shellStreamNonLive = shellStreamHealth !== "live"; + const workingLabel = + shellStreamHealth === "detached" + ? "Reconnecting" + : shellStreamHealth === "connecting" + ? "Connecting" + : "Working"; // A woken thread reappears at its original position (the sort is // deliberately static), so the pill has to carry the weight. Snoozing is // an explicit act, so the pill clears only when the user re-engages: @@ -848,7 +857,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { const topStatus = status === "working" ? { - label: "Working", + label: workingLabel, icon: "working" as const, // No shimmer: a label that animates forever is noise in a sidebar // full of them (and repaints every vsync on high-refresh displays). @@ -1487,7 +1496,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { wrapper around the ticking duration would make screen readers announce every second. */} {topStatus.label} - {status === "working" ? ( + {status === "working" && !shellStreamNonLive ? ( diff --git a/apps/web/src/components/ThreadStatusIndicators.tsx b/apps/web/src/components/ThreadStatusIndicators.tsx index 843d310dd441..94a9bc90ca5a 100644 --- a/apps/web/src/components/ThreadStatusIndicators.tsx +++ b/apps/web/src/components/ThreadStatusIndicators.tsx @@ -13,6 +13,7 @@ import { useEnvironment, usePrimaryEnvironmentId } from "../state/environments"; import { useProject } from "../state/entities"; import { useEnvironmentQuery } from "../state/query"; import { linkedPullRequestDetailAtom } from "../state/pullRequests"; +import { useShellStreamHealth } from "../state/shell"; import { useThreadRunningTerminalIds } from "../state/terminalSessions"; import { vcsEnvironment } from "../state/vcs"; import { useUiStateStore } from "../uiStateStore"; @@ -554,11 +555,13 @@ export function ThreadRowLeadingStatus({ thread }: { thread: SidebarThreadSummar pr, linkedPullRequest?.sourceControlProvider ?? gitStatus.data?.sourceControlProvider, ); + const shellStreamHealth = useShellStreamHealth(thread.environmentId); const threadStatus = resolveThreadStatusPill({ thread: { ...thread, lastVisitedAt, }, + streamHealth: shellStreamHealth, }); if (!prStatus && !threadStatus) { diff --git a/apps/web/src/components/chat/MessagesTimeline.test.tsx b/apps/web/src/components/chat/MessagesTimeline.test.tsx index 7ee4514c3709..8510d4836ebc 100644 --- a/apps/web/src/components/chat/MessagesTimeline.test.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.test.tsx @@ -651,6 +651,34 @@ describe("MessagesTimeline", () => { ).not.toContain('data-maintain-scroll-at-end="enabled"'); }); + it("renders Reconnecting instead of Working while the detail stream is detached", () => { + const markup = renderToStaticMarkup( + , + ); + expect(markup).toContain("Reconnecting"); + expect(markup).not.toContain("Working for"); + }); + + it("renders Connecting while the detail stream has never attached", () => { + const markup = renderToStaticMarkup( + , + ); + expect(markup).toContain("Connecting"); + expect(markup).not.toContain("Working for"); + }); + it("renders collapse controls for long user messages", () => { const markup = renderToStaticMarkup( (null!); @@ -208,6 +210,7 @@ interface MessagesTimelineProps { onOpenAgents?: () => void; isWorking: boolean; workingStepLabel?: string | null; + streamHealth?: "connecting" | "detached" | null; activeTurnStartedAt: string | null; listRef: React.RefObject; timelineEntries: ReturnType; @@ -251,6 +254,7 @@ interface MessagesTimelineProps { export const MessagesTimeline = memo(function MessagesTimeline({ isWorking, workingStepLabel = null, + streamHealth = null, activeTurnStartedAt, agentPanelModel = EMPTY_AGENT_PANEL_MODEL, onOpenAgents = NOOP_OPEN_AGENTS, @@ -552,10 +556,10 @@ export const MessagesTimeline = memo(function MessagesTimeline({ isRevertingCheckpoint, latestTurnId: latestTurn?.turnId ?? null, workingStepLabel, + streamState: streamHealth, }), - [isRevertingCheckpoint, isWorking, latestTurn?.turnId, workingStepLabel], + [isRevertingCheckpoint, isWorking, latestTurn?.turnId, workingStepLabel, streamHealth], ); - // Stable renderItem — no closure deps. Row components read shared state // from TimelineRowCtx, which propagates through LegendList's memo. const renderItem = useCallback( @@ -1312,17 +1316,23 @@ const TurnPlanTimelineRow = memo(function TurnPlanTimelineRow({ }); function WorkingTimelineRow({ row }: { row: Extract }) { - const { workingStepLabel } = use(TimelineRowActivityCtx); + const { workingStepLabel, streamState } = use(TimelineRowActivityCtx); return (
- {row.createdAt ? ( + {streamState != null ? ( + streamState === "detached" ? ( + "Reconnecting…" + ) : ( + "Connecting…" + ) + ) : row.createdAt ? ( <> Working for ) : ( - "Working..." + "Working…" )} {workingStepLabel ? ( · {workingStepLabel} diff --git a/apps/web/src/state/shell.ts b/apps/web/src/state/shell.ts index dfb104e5c996..48e57971a30a 100644 --- a/apps/web/src/state/shell.ts +++ b/apps/web/src/state/shell.ts @@ -1,3 +1,4 @@ +import { useAtomValue } from "@effect/atom-react"; import { AVAILABLE_CONNECTION_STATE, connectionProjectionPhase, @@ -7,7 +8,10 @@ import { createEnvironmentShellSummaryAtom, createEnvironmentSnapshotAtom, createShellEnvironmentAtoms, + shellStreamHealth, + type ShellStreamHealth, } from "@t3tools/client-runtime/state/shell"; +import type { EnvironmentId } from "@t3tools/contracts"; import * as Option from "effect/Option"; import { AsyncResult, Atom } from "effect/unstable/reactivity"; @@ -46,3 +50,41 @@ export const allEnvironmentShellsBootstrappedAtom = Atom.make((get) => { } return true; }).pipe(Atom.withLabel("web-all-environment-shells-bootstrapped")); + +/** + * Derived per-environment shell stream health (the latch lives in the shell + * state machine itself, so environments without a mounted consumer still + * track attach/drop). Deriving here lets the atom's Object.is dedupe absorb + * per-item shell churn: rows re-render only on actual health flips. + */ +const shellStreamHealthAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make((get) => shellStreamHealth(get(environmentShell.stateValueAtom(environmentId)))).pipe( + Atom.withLabel(`web-shell-stream-health:${environmentId}`), + ), +); + +/** Shell stream health for one environment (drives Working/Connecting/Reconnecting pills). */ +export function useShellStreamHealth(environmentId: EnvironmentId): ShellStreamHealth { + return useAtomValue(shellStreamHealthAtom(environmentId)); +} + +const healthByEnvironmentInAtom = Atom.family((environmentsKey: string) => + Atom.make((get) => { + // Empty member list: no health lookups. + if (environmentsKey === "") return {}; + // Null prototype: environment ids are server data, and a "__proto__" key + // must not resolve to Object.prototype. + const healths: Record = Object.create(null); + for (const environmentId of environmentsKey.split("\u0000")) { + healths[environmentId] = get(shellStreamHealthAtom(environmentId as EnvironmentId)); + } + return healths; + }).pipe(Atom.withLabel(`web-shell-stream-health-by-environments:${environmentsKey}`)), +); + +/** Shell stream health for a set of environments, keyed by environment id (rollup paths). */ +export function useShellStreamHealthForEnvironments( + environmentIds: ReadonlyArray, +): Record { + return useAtomValue(healthByEnvironmentInAtom([...environmentIds].sort().join("\u0000"))); +} diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 38c1df18044e..c15bf3ceab08 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -29,3 +29,10 @@ pill** fallback because their colors are not controlled by T3 Code. To generate a fresh title from the conversation, open a thread's context menu and choose **Regenerate title**. While T3 Code is generating it, the action reads **Regenerating…** and cannot be selected again. The option is hidden when the connected environment needs a server update. + +## Connection interruptions + +If the connection to the server's event stream drops while a thread is working, the sidebar shows +**Reconnecting** instead of **Working** while the app keeps retrying. Transient interruptions +recover on their own; if the interruption persists, the thread stays in **Reconnecting** — +restarting the app or restoring the server connection clears it. diff --git a/packages/client-runtime/package.json b/packages/client-runtime/package.json index abed33998966..ed4f5732ab61 100644 --- a/packages/client-runtime/package.json +++ b/packages/client-runtime/package.json @@ -123,6 +123,10 @@ "types": "./src/state/shell.ts", "default": "./src/state/shell.ts" }, + "./state/threadState": { + "types": "./src/state/threadState.ts", + "default": "./src/state/threadState.ts" + }, "./state/source-control": { "types": "./src/state/sourceControl.ts", "default": "./src/state/sourceControl.ts" diff --git a/packages/client-runtime/src/state/entities.test.ts b/packages/client-runtime/src/state/entities.test.ts index d3bb6680208a..d21a6149505e 100644 --- a/packages/client-runtime/src/state/entities.test.ts +++ b/packages/client-runtime/src/state/entities.test.ts @@ -145,6 +145,7 @@ function shellState(snapshot: OrchestrationShellSnapshot): EnvironmentShellState return { snapshot: Option.some(snapshot), status: "live", + wasLive: true, error: Option.none(), }; } @@ -332,6 +333,7 @@ describe("environment entity projections", () => { AsyncResult.success({ data: Option.some(detail), status: "live", + wasLive: true, error: Option.none(), page: Option.none(), }), @@ -361,6 +363,7 @@ describe("environment entity projections", () => { }, }), status: "live", + wasLive: true, error: Option.none(), page: Option.none(), }), diff --git a/packages/client-runtime/src/state/shell.test.ts b/packages/client-runtime/src/state/shell.test.ts index f1326e0a5cbe..64255fa7fbe4 100644 --- a/packages/client-runtime/src/state/shell.test.ts +++ b/packages/client-runtime/src/state/shell.test.ts @@ -30,6 +30,7 @@ function shellState(input: { readonly snapshotSequence?: number; }): EnvironmentShellState { return { + wasLive: false, snapshot: input.updatedAt === undefined ? Option.none() diff --git a/packages/client-runtime/src/state/shell.ts b/packages/client-runtime/src/state/shell.ts index c150bbb75b8c..03cbc5218567 100644 --- a/packages/client-runtime/src/state/shell.ts +++ b/packages/client-runtime/src/state/shell.ts @@ -33,14 +33,32 @@ export interface EnvironmentShellState { readonly snapshot: Option.Option; readonly status: EnvironmentShellStatus; readonly error: Option.Option; + /** True once the shell stream has attached this session; distinguishes a + mid-session drop (detached) from a cold start (connecting). */ + readonly wasLive: boolean; } -const EMPTY_SHELL_STATE: EnvironmentShellState = { +export const EMPTY_SHELL_STATE: EnvironmentShellState = { snapshot: Option.none(), status: "empty", error: Option.none(), + wasLive: false, }; +/** Derived shell stream health: live, dropped-after-attach, or never attached. */ +export function shellStreamHealth(state: EnvironmentShellState): ShellStreamHealth { + if (state.status === "live") { + return "live"; + } + return state.wasLive ? "detached" : "connecting"; +} + +/** + * Shell stream health for one environment: "connecting" = never attached this + * session, "detached" = attached before but not now, "live" = attached. + */ +export type ShellStreamHealth = "live" | "connecting" | "detached"; + function shellStatusForSnapshot( snapshot: Option.Option, ): EnvironmentShellStatus { @@ -69,6 +87,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") const state = yield* SubscriptionRef.make({ snapshot: cachedSnapshot, status: shellStatusForSnapshot(cachedSnapshot), + wasLive: false, error: Option.none(), }); const awaitingCompletion = yield* Ref.make(false); @@ -142,7 +161,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.update(state, (current) => Option.isSome(current.snapshot) - ? { ...current, status: "live" as const, error: Option.none() } + ? { ...current, status: "live" as const, wasLive: true, error: Option.none() } : current, ); return; @@ -167,6 +186,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* SubscriptionRef.set(state, { snapshot: Option.some(nextSnapshot), status: waiting ? "synchronizing" : "live", + wasLive: waiting ? current.wasLive : true, error: Option.none(), }); if (item.kind === "snapshot") { @@ -237,6 +257,7 @@ export const makeEnvironmentShellState = Effect.fn("EnvironmentShellState.make") yield* SubscriptionRef.update(state, (value) => ({ ...value, status: "live" as const, + wasLive: true, error: Option.none(), })); } diff --git a/packages/client-runtime/src/state/stream-health.test.ts b/packages/client-runtime/src/state/stream-health.test.ts new file mode 100644 index 000000000000..5882043fb8b7 --- /dev/null +++ b/packages/client-runtime/src/state/stream-health.test.ts @@ -0,0 +1,86 @@ +import { describe, expect, it } from "@effect/vitest"; +import * as Option from "effect/Option"; + +import { EMPTY_SHELL_STATE, shellStreamHealth } from "./shell.ts"; +import { + EMPTY_ENVIRONMENT_THREAD_STATE, + environmentThreadStreamHealth, + type EnvironmentThreadState, +} from "./threadState.ts"; + +const threadState = (overrides: Partial): EnvironmentThreadState => ({ + ...EMPTY_ENVIRONMENT_THREAD_STATE, + ...overrides, +}); + +describe("environmentThreadStreamHealth", () => { + it("reports live only for the live status", () => { + expect(environmentThreadStreamHealth(threadState({ status: "live" }))).toBe("live"); + }); + + it("reports connecting for tracked-but-unerrored states that are not yet live", () => { + expect(environmentThreadStreamHealth(threadState({ status: "synchronizing" }))).toBe( + "connecting", + ); + expect(environmentThreadStreamHealth(EMPTY_ENVIRONMENT_THREAD_STATE)).toBe("connecting"); + // Opening a working thread from cache is the common pre-attach state. + expect(environmentThreadStreamHealth(threadState({ status: "cached" }))).toBe("connecting"); + }); + + it("distinguishes a mid-session drop from a cold start via wasLive", () => { + // Attached earlier this session, then dropped: detached, not connecting. + expect( + environmentThreadStreamHealth(threadState({ status: "synchronizing", wasLive: true })), + ).toBe("detached"); + expect(environmentThreadStreamHealth(threadState({ status: "cached", wasLive: true }))).toBe( + "detached", + ); + // The latch never demotes an active live stream. + expect(environmentThreadStreamHealth(threadState({ status: "live", wasLive: true }))).toBe( + "live", + ); + }); + + it("reports detached whenever an error is tracked and the stream is not live", () => { + expect( + environmentThreadStreamHealth(threadState({ status: "cached", error: Option.some("boom") })), + ).toBe("detached"); + expect( + environmentThreadStreamHealth( + threadState({ status: "synchronizing", error: Option.some("boom") }), + ), + ).toBe("detached"); + expect( + environmentThreadStreamHealth(threadState({ status: "empty", error: Option.some("boom") })), + ).toBe("detached"); + // Error precedence over live pins the branch order. + expect( + environmentThreadStreamHealth(threadState({ status: "live", error: Option.some("boom") })), + ).toBe("detached"); + }); + + it("reports deleted regardless of error state", () => { + expect(environmentThreadStreamHealth(threadState({ status: "deleted" }))).toBe("deleted"); + }); +}); + +describe("shellStreamHealth", () => { + it("is live only for the live status", () => { + expect(shellStreamHealth({ ...EMPTY_SHELL_STATE, status: "live", wasLive: true })).toBe("live"); + }); + + it("reads connecting before the first attach", () => { + expect(shellStreamHealth(EMPTY_SHELL_STATE)).toBe("connecting"); + expect(shellStreamHealth({ ...EMPTY_SHELL_STATE, status: "synchronizing" })).toBe("connecting"); + expect(shellStreamHealth({ ...EMPTY_SHELL_STATE, status: "cached" })).toBe("connecting"); + }); + + it("reads detached after an attach drops", () => { + expect(shellStreamHealth({ ...EMPTY_SHELL_STATE, status: "cached", wasLive: true })).toBe( + "detached", + ); + expect( + shellStreamHealth({ ...EMPTY_SHELL_STATE, status: "synchronizing", wasLive: true }), + ).toBe("detached"); + }); +}); diff --git a/packages/client-runtime/src/state/threadState.ts b/packages/client-runtime/src/state/threadState.ts index 8ba9696ec576..4f9f644cea1a 100644 --- a/packages/client-runtime/src/state/threadState.ts +++ b/packages/client-runtime/src/state/threadState.ts @@ -22,6 +22,8 @@ export interface EnvironmentThreadState { readonly status: EnvironmentThreadStatus; readonly error: Option.Option; readonly page: Option.Option; + /** True once the thread's detail stream has attached this session; distinguishes a mid-session drop (detached) from a cold start (connecting). */ + readonly wasLive: boolean; } export const EMPTY_ENVIRONMENT_THREAD_STATE: EnvironmentThreadState = { @@ -29,6 +31,7 @@ export const EMPTY_ENVIRONMENT_THREAD_STATE: EnvironmentThreadState = { status: "empty", error: Option.none(), page: Option.none(), + wasLive: false, }; /** Whether the thread has older turns that can be loaded with more pages. */ @@ -38,3 +41,22 @@ export function threadHasOlderTurns(state: EnvironmentThreadState): boolean { onSome: (page) => page.hasMore, }); } + +export type EnvironmentThreadStreamHealth = "live" | "connecting" | "detached" | "deleted"; + +/** + * Whether turn activity may render as healthy live work: "Working" may only + * render while the detail stream that would deliver the turn completion is + * alive. Pure and dependency-free so the web and mobile clients can share it. + * A state that attached this session (wasLive) reads detached once it stops + * being live — dropped-after-attach never masquerades as a cold start, matching + * the sidebar shell's wasLive-latched health vocabulary. + */ +export function environmentThreadStreamHealth( + state: EnvironmentThreadState, +): EnvironmentThreadStreamHealth { + if (state.status === "deleted") return "deleted"; + if (Option.isSome(state.error)) return "detached"; + if (state.status === "live") return "live"; + return state.wasLive ? "detached" : "connecting"; +} diff --git a/packages/client-runtime/src/state/threads.ts b/packages/client-runtime/src/state/threads.ts index 39561afdb416..bba48660c873 100644 --- a/packages/client-runtime/src/state/threads.ts +++ b/packages/client-runtime/src/state/threads.ts @@ -159,6 +159,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make // A cached windowed snapshot restores its page cursor so "load earlier" // works while rendering from cache; a cached full snapshot has no page. page: Option.flatMap(cached, (snapshot) => pageStateFromSnapshot(snapshot.page)), + wasLive: false, }); // Seed the resume cursor from the cached snapshot so a warm cache can catch up // via `afterSequence` instead of re-downloading the full thread body. @@ -264,6 +265,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make status: waiting ? ("synchronizing" as const) : ("live" as const), error: Option.none(), page: page === "keep" ? current.page : page, + wasLive: waiting ? current.wasLive : true, })); // Active threads can update many times per second and retain large tool // payloads. The server remains the source of truth while a turn is active; @@ -294,11 +296,13 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make const setDeleted = Effect.fn("EnvironmentThreadState.setDeleted")(function* () { yield* Ref.set(awaitingCompletion, false); yield* Ref.update(historyEpoch, (epoch) => epoch + 1); + const current = yield* SubscriptionRef.get(state); yield* SubscriptionRef.set(state, { data: Option.none(), status: "deleted", error: Option.none(), page: Option.none(), + wasLive: current.wasLive, }); yield* cache.removeThread(environmentId, threadId).pipe( Effect.catch((error) => @@ -321,7 +325,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make yield* Ref.set(awaitingCompletion, false); yield* SubscriptionRef.update(state, (current) => Option.isSome(current.data) && current.status !== "deleted" - ? { ...current, status: "live" as const, error: Option.none() } + ? { ...current, status: "live" as const, error: Option.none(), wasLive: true } : current, ); return; @@ -623,6 +627,7 @@ export const makeEnvironmentThreadState = Effect.fn("EnvironmentThreadState.make ...value, status: value.status === "deleted" ? value.status : ("live" as const), error: Option.none(), + wasLive: value.status === "deleted" ? value.wasLive : true, })); }