diff --git a/apps/mobile/src/features/threads/thread-list-v2-items.tsx b/apps/mobile/src/features/threads/thread-list-v2-items.tsx index 84e9494759cad..3e3f5abd8843e 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -76,6 +76,8 @@ import { type ThreadListV2Status, } from "./threadListV2"; import { QueuedMessageIcon } from "./queued-message-icon"; +import { useThreadHasNoPendingWake } from "./use-thread-has-no-pending-wake"; +import { RelativeTime } from "../../components/RelativeTime"; import { shouldRecedeThreadRow } from "./thread-row-emphasis"; import { ThreadSearchMatchExcerpt } from "./thread-search-match"; import { useScaledTextRole } from "../settings/appearance/useScaledTextRole"; @@ -132,6 +134,12 @@ const STATUS_LABEL_BY_STATUS: Partial> = iconTintClassName: "accent-warning-foreground", }, }; +const IDLE_STATUS_LABEL: StatusLabel = { + label: "Idle", + icon: "circle", + className: "text-foreground-muted", + iconTintClassName: "accent-foreground-muted", +}; const DONE_STATUS_LABEL: StatusLabel = { label: "Done", icon: "checkmark.circle", @@ -739,10 +747,15 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { const isUnread = status === "ready" && threadHasUnseenCompletion(thread); const workingLabel = STATUS_LABEL_BY_STATUS[status]; const showsWoke = status === "ready" && props.wokeAt !== null; + // Settled and snoozed rows show the user's own decision, and queued messages will start a turn. + const isIdle = useThreadHasNoPendingWake( + thread, + variant === "card" && !snoozedRow && props.hasQueuedMessages !== true, + ); const statusLabel = (status === "working" && workingLabel !== undefined && thread.goal?.status === "active" ? { ...workingLabel, label: "Goal" } - : workingLabel) ?? (isUnread ? DONE_STATUS_LABEL : undefined); + : workingLabel) ?? (isUnread ? DONE_STATUS_LABEL : isIdle ? IDLE_STATUS_LABEL : undefined); const recede = shouldRecedeThreadRow({ status, selected }); const isRegeneratingTitle = thread.titleRegeneration != null; const rowAccessibilityLabel = `${thread.title}${props.hasQueuedMessages ? ", messages queued to send" : ""}${isRegeneratingTitle ? ", regenerating title" : ""}`; @@ -1205,6 +1218,11 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { ) : status === "waiting" ? ( + ) : statusLabel === IDLE_STATUS_LABEL && thread.latestRun?.completedAt ? ( + <> + {" "} + + ) : null} diff --git a/apps/mobile/src/features/threads/use-thread-has-no-pending-wake.ts b/apps/mobile/src/features/threads/use-thread-has-no-pending-wake.ts new file mode 100644 index 0000000000000..1e2eae0477863 --- /dev/null +++ b/apps/mobile/src/features/threads/use-thread-has-no-pending-wake.ts @@ -0,0 +1,25 @@ +import { useAtomValue } from "@effect/atom-react"; +import { scopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment"; +import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/models"; +import { threadHasNoPendingWake } from "@t3tools/client-runtime/state/thread-inbox"; +import { Atom } from "effect/reactivity"; + +import { serverEnvironment } from "../../state/server"; + +const UNKNOWN_SCHEDULED_WAKE_ATOM = Atom.make(null); + +/** Only threads that are otherwise idle subscribe to the scheduled task list. */ +export function useThreadHasNoPendingWake( + thread: EnvironmentThreadShell, + enabled: boolean, +): boolean { + const mayBeIdle = enabled && threadHasNoPendingWake(thread, false); + const hasScheduledWake = useAtomValue( + mayBeIdle + ? serverEnvironment.threadHasScheduledWake( + scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), + ) + : UNKNOWN_SCHEDULED_WAKE_ATOM, + ); + return mayBeIdle && hasScheduledWake === false; +} diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index 14bb72e3c78fb..b8107bb94afb3 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -648,6 +648,17 @@ it.effect("SQL queue readiness skips held runs without bypassing active work", ( const { runs } = yield* store.getThreadProjection(threadId); const run = runs[0]!; const now = yield* DateTime.now; + const assertShellsHoldQueue = Effect.fnUntraced(function* (held: boolean) { + const memoryShell = ProjectionStore.threadShellFromProjection( + yield* store.getThreadProjection(threadId), + ); + const sqlShell = (yield* store.getShellSnapshot()).threads.find( + (row) => row.id === threadId, + )!; + assert.equal(memoryShell.hasHeldQueue, held); + assert.equal(sqlShell.hasHeldQueue, held); + }); + yield* assertShellsHoldQueue(false); const held = { ...run, status: "queued" as const, queueHeld: true }; yield* store.apply({ id: EventId.make("held-queue:held"), @@ -657,6 +668,7 @@ it.effect("SQL queue readiness skips held runs without bypassing active work", ( payload: held, }); assert.isFalse(yield* store.canStartQueuedRun(threadId)); + yield* assertShellsHoldQueue(true); const wake = { ...run, id: RunId.make("held-queue:wake"), ordinal: 2 }; for (const status of ["running", "queued", "completed"] as const) { diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index a46c8017731d5..30f31365cff2b 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -990,6 +990,7 @@ type ShellThreadRow = { readonly latest_run_started_at: string | null; readonly latest_run_completed_at: string | null; readonly latest_run_stop_requested: number; + readonly has_held_queue: number; readonly active_run_id: string | null; readonly active_run_status: string | null; readonly activity_run_status: string | null; @@ -1608,6 +1609,7 @@ export function threadShellFromProjection( projection.turnItems.some( (item) => item.runId === latestRun.id && item.type === "run_interrupt_request", ), + hasHeldQueue: runs.some((run) => run.status === "queued" && run.queueHeld === true), activeRunId: activeRun?.id ?? null, blockedOnThreadWait: deriveBlockedOnThreadWait({ activeRun, turnItems: projection.turnItems }), activityRunStatus: activityRun?.status ?? null, @@ -1725,6 +1727,7 @@ type ShellThreadState = { readonly latestRunStartedAt: DateTime.Utc | null; readonly latestRunCompletedAt: DateTime.Utc | null; readonly latestRunStopRequested: boolean; + readonly hasHeldQueue: boolean; readonly activeRunId: RunId | null; readonly blockedOnThreadWait: boolean; readonly activityRunStatus: ShellActivityRunStatus | null; @@ -1885,6 +1888,7 @@ function shellFromState(input: { latestRunStartedAt: input.state.latestRunStartedAt, latestRunCompletedAt: input.state.latestRunCompletedAt, latestRunStopRequested: input.state.latestRunStopRequested, + hasHeldQueue: input.state.hasHeldQueue, activeRunId: input.state.activeRunId, blockedOnThreadWait: input.state.blockedOnThreadWait, activityRunStatus: input.state.activityRunStatus, @@ -6099,6 +6103,12 @@ export const layer: Layer.Layer = WHERE stop_request.thread_id = t.thread_id AND stop_request.run_id = presented.run_id AND stop_request.type = 'run_interrupt_request' ) AS latest_run_stop_requested, + EXISTS ( + SELECT 1 FROM orchestration_v2_projection_runs held + WHERE held.thread_id = t.thread_id AND held.status = 'queued' + AND json_extract(held.payload_json, '$.queueHeld') IS 1 + AND NOT ${isSilentShellRun("held")} + ) AS has_held_queue, active.run_id AS active_run_id, active.status AS active_run_status, ( @@ -6779,6 +6789,7 @@ export const layer: Layer.Layer = latestRunStartedAt, latestRunCompletedAt, latestRunStopRequested, + hasHeldQueue: row.has_held_queue === 1, activeRunId: row.active_run_id === null ? null : RunId.make(row.active_run_id), blockedOnThreadWait: deriveBlockedOnThreadWait({ activeRun: diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index 3e2e8be2ce18d..3a7fdd82b9495 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -10,6 +10,7 @@ import { deleteSelectedThreadEntries, getFallbackThreadIdAfterDelete, hasUnseenCompletion, + threadHeaderMayShowIdle, resolveSidebarRowThreadId, belongsInSidebarWorkingShelf, isSidebarEnvironmentDisconnected, @@ -44,7 +45,14 @@ import { } from "@t3tools/client-runtime/state/thread-sort"; import { resolveProjectExpanded } from "../uiStateStore"; import { threadSearchMatchKey } from "@t3tools/client-runtime/state/thread-search"; -import { EnvironmentId, ProjectId, ProviderInstanceId, RunId, ThreadId } from "@t3tools/contracts"; +import { + EnvironmentId, + ProjectId, + ProviderInstanceId, + RunId, + ThreadId, + type OrchestrationV2ThreadShell, +} from "@t3tools/contracts"; import { DEFAULT_INTERACTION_MODE, @@ -901,6 +909,38 @@ describe("navigation after parking a thread", () => { }); describe("unseen completion with background work", () => { + it("lets the header show Idle only where the sidebar row would", () => { + const now = "2026-06-20T02:00:00.000Z"; + const read = "2026-06-20T01:30:00.000Z"; + const ended = (overrides: Partial = {}) => + presentThreadShell(localEnvironmentId, { + ...makeThreadFixture().source, + latestRunId: RunId.make("run-header-idle"), + status: "completed", + latestRunCompletedAt: DateTime.makeUnsafe("2026-06-20T01:00:00.000Z"), + ...overrides, + }); + expect(threadHeaderMayShowIdle(ended(), { lastVisitedAt: read, now })).toBe(true); + // Done comes first while the completion is unread. + const unread = { lastVisitedAt: "2026-06-20T00:30:00.000Z", now }; + expect(threadHeaderMayShowIdle(ended(), unread)).toBe(false); + const settled = ended({ settledOverride: "settled" }); + expect(threadHeaderMayShowIdle(settled, { lastVisitedAt: read, now })).toBe(false); + const snoozed = (until: string) => + ended({ + snoozedAt: DateTime.makeUnsafe("2026-06-20T01:35:00.000Z"), + snoozedUntil: DateTime.makeUnsafe(until), + }); + expect( + threadHeaderMayShowIdle(snoozed("2026-06-20T03:00:00.000Z"), { lastVisitedAt: read, now }), + ).toBe(false); + // Woke comes first until the user opens the thread after it woke. + const woke = snoozed("2026-06-20T01:50:00.000Z"); + expect(threadHeaderMayShowIdle(woke, { lastVisitedAt: read, now })).toBe(false); + const seen = { lastVisitedAt: "2026-06-20T01:55:00.000Z", now }; + expect(threadHeaderMayShowIdle(woke, seen)).toBe(true); + }); + it.each([ { roster: "dev server", diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index c20789e2a9160..872efdacbe212 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -27,6 +27,8 @@ import type { SidebarProjectSortOrder, SidebarThreadSortOrder } from "@t3tools/c import type { AsyncResult } from "effect/reactivity"; import { effectiveSnoozed, + isThreadListedActive, + threadWokeAt, type ThreadSnoozeShell, } from "@t3tools/client-runtime/state/thread-settled"; import { @@ -38,6 +40,7 @@ import { } from "@t3tools/client-runtime/state/thread-sort"; import type { SidebarThreadSummary, Thread } from "../types"; import { normalizeSearchText } from "../lib/utils"; +import { parseTimestampDate } from "../timestampFormat"; import { isLatestRunSettled } from "../session-logic"; import { observeIntersection } from "../lib/observeIntersection"; import { deadlineAfter, deadlineStore } from "@t3tools/client-runtime/clock"; @@ -730,6 +733,36 @@ export function hasUnseenCompletion(thread: ThreadStatusInput): boolean { return completedAt > lastVisitedAt; } +/** A thread woke from its snooze after the user last opened it. */ +export function isSidebarThreadWoke(input: { + readonly wokeAt: string | null; + readonly lastVisitedAt: string | undefined; + readonly settled: boolean; +}): boolean { + const wokeAtDate = input.wokeAt === null ? null : parseTimestampDate(input.wokeAt); + const lastVisitedDate = + input.lastVisitedAt === undefined ? null : parseTimestampDate(input.lastVisitedAt); + return ( + wokeAtDate !== null && + (lastVisitedDate === null || lastVisitedDate < wokeAtDate) && + !input.settled + ); +} + +/** The header may show Idle only where the thread's sidebar row would: active, read, not woke. */ +export function threadHeaderMayShowIdle( + thread: ThreadStatusInput & ThreadSnoozeShell & Pick, + options: { readonly lastVisitedAt: string | undefined; readonly now: string }, +): boolean { + if (!isThreadListedActive(thread, { now: options.now })) return false; + if (hasUnseenCompletion({ ...thread, lastVisitedAt: options.lastVisitedAt })) return false; + return !isSidebarThreadWoke({ + wokeAt: threadWokeAt(thread, { now: options.now }), + lastVisitedAt: options.lastVisitedAt, + settled: false, + }); +} + /** Unopened threads' failures still count as unseen, unlike completions. */ function hasUnseenFailure(thread: ThreadStatusInput): boolean { if (thread.latestRun?.status !== "failed") return false; diff --git a/apps/web/src/components/chat/ChatHeader.tsx b/apps/web/src/components/chat/ChatHeader.tsx index 49f21a1ddc829..d4502cbb2c330 100644 --- a/apps/web/src/components/chat/ChatHeader.tsx +++ b/apps/web/src/components/chat/ChatHeader.tsx @@ -4,7 +4,7 @@ import { type ScopedThreadRef, type ThreadId, } from "@t3tools/contracts"; -import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import { scopeThreadRef, scopedThreadKey } from "@t3tools/client-runtime/environment"; import type { EnvironmentProject } from "@t3tools/client-runtime/state/shell"; import { isAtomCommandInterrupted, @@ -19,7 +19,12 @@ import { type KeyboardEvent as ReactKeyboardEvent, type MouseEvent as ReactMouseEvent, } from "react"; -import { isTrailingDoubleClick } from "../Sidebar.logic"; +import { + isTrailingDoubleClick, + resolveThreadLastVisitedAt, + threadHeaderMayShowIdle, +} from "../Sidebar.logic"; +import { useUiStateStore } from "../../uiStateStore"; import { Badge } from "../ui/badge"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; import { toastManager } from "../ui/toast"; @@ -42,6 +47,32 @@ import { cn } from "~/lib/utils"; import { textDirection } from "~/textDirection"; import { useClientSettings } from "../../hooks/useSettings"; import { GuestShareIndicator } from "./GuestShareControls"; +import { useQuietThreadShell } from "../../state/entities"; +import { useNowAfter } from "~/hooks/useDeadline"; +import { useThreadHasNoPendingWake } from "../../hooks/useThreadHasNoPendingWake"; + +// Its own component, so shell updates during a turn re-render only the badge. +function ThreadIdleBadge(props: { threadRef: ScopedThreadRef }) { + const shell = useQuietThreadShell(props.threadRef); + const localLastVisitedAt = useUiStateStore( + (state) => state.threadLastVisitedAtById[scopedThreadKey(props.threadRef)], + ); + // A clock read in render would stay frozen in the compiler's cache; this one moves when a snooze ends. + const now = new Date(useNowAfter(Date.parse(shell?.snoozedUntil ?? "") || null)).toISOString(); + const mayShowIdle = + shell !== null && + threadHeaderMayShowIdle(shell, { + lastVisitedAt: resolveThreadLastVisitedAt(shell.lastVisitedAt, localLastVisitedAt), + now, + }); + if (!useThreadHasNoPendingWake(shell, mayShowIdle)) return null; + return ( + + }>Idle + Nothing will resume this thread + + ); +} interface ChatHeaderProps { activeThreadEnvironmentId: EnvironmentId; @@ -418,6 +449,7 @@ export const ChatHeader = memo(function ChatHeader({ {activeThreadTitle} )} + {renamingTitle === null ? : null} {renamingTitle === null && activeThreadLinkOrigin ? ( }> diff --git a/apps/web/src/components/sidebar/SidebarThreadRow.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.tsx index ed3a8276be8bd..119cc7e7d88c8 100644 --- a/apps/web/src/components/sidebar/SidebarThreadRow.tsx +++ b/apps/web/src/components/sidebar/SidebarThreadRow.tsx @@ -37,6 +37,7 @@ import { CircleCheckIcon, CircleDashedIcon, CircleDotDashedIcon, + CircleIcon, ClockIcon, GitBranchIcon, MessageCircleQuestionIcon, @@ -68,7 +69,7 @@ import { useNowSecond } from "../../hooks/useNowSecond"; import { vcsEnvironment } from "../../state/vcs"; import { useEnvironmentQuery } from "../../state/query"; import { readEnvironmentScope, useEnvironmentScope } from "../../state/session"; -import { formatRelativeTimeLabel, parseTimestampDate } from "../../timestampFormat"; +import { formatRelativeTimeLabel } from "../../timestampFormat"; import type { SidebarThreadSummary } from "../../types"; import type { EnvironmentProject } from "@t3tools/client-runtime/state/shell"; import { cn } from "~/lib/utils"; @@ -77,6 +78,7 @@ import { EnvironmentMachineIcon } from "../EnvironmentMachineIcon"; import { animateSidebarLayoutChanges, hasUnseenCompletion, + isSidebarThreadWoke, resolveSidebarRowAccessibility, type SidebarDropVerb, resolveSidebarThreadStatus, @@ -114,6 +116,7 @@ import { Menu, MenuItem, MenuPopup, MenuSeparator, MenuShortcut, MenuTrigger } f import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; import { MiddleTruncate } from "../ui/middle-truncate"; import { useThreadHasUnsentDraft } from "../../composerDraftStore"; +import { useThreadHasNoPendingWake } from "../../hooks/useThreadHasNoPendingWake"; // The fields thread rows show. Provider status pushes refresh `checkedAt`, // status and auth on every provider; leaving those out keeps each @@ -519,6 +522,7 @@ const dropVerbBadge: Record = { const TOP_STATUS_ICONS = { working: CircleDashedIcon, waiting: CircleDotDashedIcon, + idle: CircleIcon, input: MessageCircleQuestionIcon, request: UserRoundIcon, approval: ShieldQuestionIcon, @@ -708,12 +712,13 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP const isUnread = hasUnseenCompletion({ ...thread, lastVisitedAt }); const status = resolveSidebarThreadStatus(thread); - const lastVisitedDate = lastVisitedAt === undefined ? null : parseTimestampDate(lastVisitedAt); - const wokeAtDate = props.wokeAt === null ? null : parseTimestampDate(props.wokeAt); - const isWoke = - wokeAtDate !== null && - (lastVisitedDate === null || lastVisitedDate < wokeAtDate) && - thread.settledOverride !== "settled"; + // Settled and snoozed rows already show the user's own decision instead. + const isIdle = useThreadHasNoPendingWake(thread, variantAction === "settle"); + const isWoke = isSidebarThreadWoke({ + wokeAt: props.wokeAt, + lastVisitedAt, + settled: thread.settledOverride === "settled", + }); const shouldRecede = shouldRecedeSidebarThread({ status, isUnread, @@ -776,7 +781,13 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP icon: "done" as const, className: "text-success-foreground", } - : null; + : isIdle + ? { + label: "Idle", + icon: "idle" as const, + className: "text-muted-foreground", + } + : null; const isWokeStatus = topStatus?.icon === "woke"; const branchMismatch = resolveLocalCheckoutBranchMismatch({ @@ -1504,6 +1515,12 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP + ) : topStatus.icon === "idle" ? ( + + + ) : null} ) diff --git a/apps/web/src/hooks/useThreadHasNoPendingWake.ts b/apps/web/src/hooks/useThreadHasNoPendingWake.ts new file mode 100644 index 0000000000000..83052ff3d44c8 --- /dev/null +++ b/apps/web/src/hooks/useThreadHasNoPendingWake.ts @@ -0,0 +1,25 @@ +import { useAtomValue } from "@effect/atom-react"; +import { scopedThreadKey, scopeThreadRef } from "@t3tools/client-runtime/environment"; +import type { EnvironmentThreadShell } from "@t3tools/client-runtime/state/models"; +import { threadHasNoPendingWake } from "@t3tools/client-runtime/state/thread-inbox"; +import { Atom } from "effect/reactivity"; + +import { serverEnvironment } from "../state/server"; + +const UNKNOWN_SCHEDULED_WAKE_ATOM = Atom.make(null); + +/** Only threads that are otherwise idle subscribe to the scheduled task list. */ +export function useThreadHasNoPendingWake( + thread: EnvironmentThreadShell | null, + enabled = true, +): boolean { + const mayBeIdle = enabled && thread !== null && threadHasNoPendingWake(thread, false); + const hasScheduledWake = useAtomValue( + mayBeIdle + ? serverEnvironment.threadHasScheduledWake( + scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)), + ) + : UNKNOWN_SCHEDULED_WAKE_ATOM, + ); + return mayBeIdle && hasScheduledWake === false; +} diff --git a/docs/user/thread-sidebar.md b/docs/user/thread-sidebar.md index 6a3145d17512f..b453017feec26 100644 --- a/docs/user/thread-sidebar.md +++ b/docs/user/thread-sidebar.md @@ -148,6 +148,15 @@ choice. While this is on, the active list is ordered by when each thread last came back to you, so you cannot drag or move threads within it. Your saved order returns when you turn it off. +### Idle threads + +A thread whose latest turn ended with nothing left to start another one shows **Idle** in the +sidebar, in the thread header on web and desktop, and in the thread list on mobile. No running +background work, watched pull request, delegated thread, queued message, or scheduled task bound +to the thread will resume it, even when the agent's last message says it will continue later. +Send a message to continue it. A thread whose agent waits on background work, a watched pull +request, or a delegated thread shows **Waiting** instead. + ### Group threads by project (beta) Turn on **Settings → Labs → Boring sidebar (group by project)** on web and desktop to put active diff --git a/packages/client-runtime/src/state/entities.test.ts b/packages/client-runtime/src/state/entities.test.ts index 6cc9bc91595da..88718603da4db 100644 --- a/packages/client-runtime/src/state/entities.test.ts +++ b/packages/client-runtime/src/state/entities.test.ts @@ -8,6 +8,7 @@ import { RunId, RuntimeRequestId, TurnItemId, + type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import { describe, expect, it } from "vite-plus/test"; @@ -17,7 +18,7 @@ import { v2Projection, v2ThreadShell } from "./orchestrationV2TestFixtures.ts"; import { deriveLatestThreadRun, deriveThreadRuntime } from "./threadExecution.ts"; import { isThreadWorking } from "./threadInbox.ts"; import { derivePendingThreadRequests } from "./threadRequests.ts"; -import { canSnooze, effectiveSnoozed } from "./threadSettled.ts"; +import { canSnooze, effectiveSnoozed, isThreadListedActive } from "./threadSettled.ts"; const environmentId = EnvironmentId.make("environment-v2"); @@ -43,6 +44,30 @@ describe("V2 client presentation", () => { expect(effectiveSnoozed(active, { now: "2026-06-20T00:05:00.000Z" })).toBe(false); }); + it("lists a thread as active only while it is neither settled nor snoozed", () => { + const now = "2026-06-20T00:05:00.000Z"; + const ended = (overrides: Partial) => + presentThreadShell(environmentId, { + ...v2ThreadShell, + latestRunId: RunId.make("run-ended"), + status: "completed", + ...overrides, + }); + const snoozedUntil = (until: string) => ({ + snoozedAt: DateTime.makeUnsafe("2026-06-19T23:00:00.000Z"), + snoozedUntil: DateTime.makeUnsafe(until), + }); + expect(isThreadListedActive(ended({}), { now })).toBe(true); + expect(isThreadListedActive(ended({ settledOverride: "settled" }), { now })).toBe(false); + expect(isThreadListedActive(ended(snoozedUntil("2026-06-20T01:00:00.000Z")), { now })).toBe( + false, + ); + // A snooze that already ended puts the thread back in the active list. + expect(isThreadListedActive(ended(snoozedUntil("2026-06-20T00:00:00.000Z")), { now })).toBe( + true, + ); + }); + it("moves a running thread with any question out of Working", () => { for (const capability of [{ blocking: false }, { blocking: true }, {}]) { const shell = presentThreadShell(environmentId, { diff --git a/packages/client-runtime/src/state/models.ts b/packages/client-runtime/src/state/models.ts index 196b5ff39e171..1b337c00d019b 100644 --- a/packages/client-runtime/src/state/models.ts +++ b/packages/client-runtime/src/state/models.ts @@ -154,6 +154,8 @@ export interface EnvironmentThreadShell { readonly latestRun: ThreadRunSummary | null; readonly runtime: ThreadRuntimeSummary | null; readonly blockedOnThreadWait?: boolean; + /** Queued messages wait, held after a failure or a stop, until released. Unknown on older servers. */ + readonly hasHeldQueue?: boolean; readonly latestUserMessageAt: string | null; readonly latestUserAuthoredMessageAt?: string | null; readonly hasPendingApprovals: boolean; @@ -302,6 +304,7 @@ export function presentThreadShell( latestRun, runtime, blockedOnThreadWait: thread.blockedOnThreadWait ?? false, + ...(thread.hasHeldQueue === undefined ? {} : { hasHeldQueue: thread.hasHeldQueue }), latestUserMessageAt: nullableIso(thread.latestUserMessageAt), ...(thread.latestUserAuthoredMessageAt === undefined ? {} diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index 04a2d66321517..c4eedd6bdcd8e 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -52,7 +52,9 @@ import { type EnvironmentRpcInput, } from "../rpc/client.ts"; import type { RpcSession } from "../rpc/session.ts"; +import { parseScopedThreadKey } from "../environment/scoped.ts"; import { followStreamInEnvironment } from "./runtime.ts"; +import { scheduledTasksWakeThread } from "./threadInbox.ts"; import { applyServerConfigProjection, reuseUnchanged, @@ -1200,6 +1202,11 @@ export function createServerEnvironmentAtoms( concurrency: configConcurrency, }); + const scheduledTasksLive = createEnvironmentRpcSubscriptionAtomFamily(runtime, { + label: "environment-data:server:scheduled-tasks:live", + tag: WS_METHODS.scheduledTasksSubscribe, + }); + return { configValueAtom, runStorageCleanup: createEnvironmentRpcCommand(runtime, { @@ -1323,9 +1330,16 @@ export function createServerEnvironmentAtoms( label: "environment-data:server:scheduled-task:webhook-delivery", tag: WS_METHODS.scheduledTasksGetWebhookDelivery, }), - scheduledTasksLive: createEnvironmentRpcSubscriptionAtomFamily(runtime, { - label: "environment-data:server:scheduled-tasks:live", - tag: WS_METHODS.scheduledTasksSubscribe, + scheduledTasksLive, + /** Keyed by scoped thread key; a list push re-renders only threads whose answer flips. */ + threadHasScheduledWake: Atom.family((threadKey: string) => { + const ref = parseScopedThreadKey(threadKey); + return Atom.make((get): boolean | null => { + if (ref === null) return null; + const result = get(scheduledTasksLive({ environmentId: ref.environmentId, input: {} })); + const tasks = AsyncResult.getOrElse(result, () => null)?.tasks; + return tasks === undefined ? null : scheduledTasksWakeThread(tasks, ref.threadId); + }).pipe(Atom.withLabel(`environment-data:server:thread-scheduled-wake:${threadKey}`)); }), // A cold transcript scan is measured in seconds, so keep the result around // long enough that switching windows or re-rendering does not rescan. diff --git a/packages/client-runtime/src/state/threadInbox.test.ts b/packages/client-runtime/src/state/threadInbox.test.ts index 35c7fa8355e39..fe7bc82df0b16 100644 --- a/packages/client-runtime/src/state/threadInbox.test.ts +++ b/packages/client-runtime/src/state/threadInbox.test.ts @@ -1,11 +1,23 @@ -import { EnvironmentId, ProviderInstanceId, RunId, ThreadId } from "@t3tools/contracts"; +import { + EnvironmentId, + ProviderInstanceId, + RunId, + RuntimeRequestId, + ThreadId, + type OrchestrationV2ThreadShell, + type ScheduledTask, +} from "@t3tools/contracts"; import { afterEach, beforeEach, describe, expect, it, vi } from "vite-plus/test"; +import { presentThreadShell } from "./models.ts"; +import { v2ThreadShell } from "./orchestrationV2TestFixtures.ts"; import { createInboxReturnTracker, isThreadWorking, sortInboxThreadsByReturn, sortWorkingThreadsBySend, + scheduledTasksWakeThread, + threadHasNoPendingWake, } from "./threadInbox.ts"; const environmentId = EnvironmentId.make("environment-1"); @@ -241,3 +253,119 @@ describe("sortWorkingThreadsBySend", () => { ).toEqual(["sent-last", "sent-first", "launched"]); }); }); + +describe("threadHasNoPendingWake", () => { + const runId = RunId.make("run-ended"); + const ended = { ...v2ThreadShell, latestRunId: runId, status: "completed" as const }; + const present = (shell: Partial) => + presentThreadShell(environmentId, { ...ended, hasHeldQueue: false, ...shell }); + const request = (kind: "user_input" | "dynamic_tool_call") => ({ + id: RuntimeRequestId.make("request-open"), + kind, + createdAt: v2ThreadShell.updatedAt, + }); + const watch = { + taskId: "pull-request-watch:github.com/acme/app#1", + kind: "monitor" as const, + awaited: true as const, + }; + + it("is idle once the turn ended with nothing that could resume it", () => { + for (const status of ["completed", "interrupted", "cancelled"] as const) { + expect(threadHasNoPendingWake(present({ status }), false)).toBe(true); + } + }); + + it.each([ + { name: "no turn ran yet", shell: { latestRunId: null, status: "idle" as const } }, + { name: "a turn is running", shell: { activeRunId: runId, status: "running" as const } }, + { name: "the turn awaits its checkpoint", shell: { status: "waiting" as const } }, + { name: "the turn failed", shell: { status: "failed" as const } }, + // A rolled-back latest run hides the pending work list, watches included. + { name: "the turn was rolled back", shell: { status: "rolled_back" as const } }, + { + name: "an awaited background command runs", + shell: { + pendingBackgroundTasks: [ + { taskId: "tests", kind: "command" as const, awaited: true as const }, + ], + }, + }, + { + // Claude wakes the agent when any background command it started exits. + name: "an unawaited background command runs", + shell: { pendingBackgroundTasks: [{ taskId: "build", kind: "command" as const }] }, + }, + { + name: "a delegated thread is open", + shell: { + pendingBackgroundTasks: [ + { taskId: "delegate", kind: "subagent" as const, childThreadId: ThreadId.make("child") }, + ], + }, + }, + { name: "a pull request is watched", shell: { pendingBackgroundTasks: [watch] } }, + { + name: "an approval is pending", + shell: { pendingRuntimeRequest: request("dynamic_tool_call") }, + }, + { name: "a question is pending", shell: { pendingRuntimeRequest: request("user_input") } }, + { name: "a guest message waits", shell: { pendingGuestMessageCount: 1 } }, + // Includes a message queued during the turn and held by a stop, which predates the turn's end. + { name: "a message waits in a held queue", shell: { hasHeldQueue: true } }, + ])("is not idle when $name", ({ shell }) => { + expect(threadHasNoPendingWake(present(shell), false)).toBe(false); + }); + + it("is not idle while a child thread asks for approval or input", () => { + for (const childRequest of ["approval", "user_input"] as const) { + const shell = presentThreadShell( + environmentId, + { ...ended, hasHeldQueue: false }, + childRequest, + ); + expect(threadHasNoPendingWake(shell, false)).toBe(false); + } + }); + + it("is not idle when an older server does not report its held queue", () => { + expect(threadHasNoPendingWake(presentThreadShell(environmentId, ended), false)).toBe(false); + }); + + it("is not idle while a scheduled task can run into it, or before the task list loads", () => { + expect(threadHasNoPendingWake(present({}), true)).toBe(false); + expect(threadHasNoPendingWake(present({}), null)).toBe(false); + }); +}); + +describe("scheduledTasksWakeThread", () => { + const threadId = v2ThreadShell.id; + const task = ( + overrides: Partial< + Pick + > = {}, + ) => ({ + threadId, + enabled: true, + nextRunAt: "2026-06-21T00:00:00.000Z", + schedule: { type: "interval" as const, everyMs: 3_600_000 }, + ...overrides, + }); + + it("counts a bound task with a next run or an enabled webhook", () => { + expect(scheduledTasksWakeThread([task()], threadId)).toBe(true); + const webhook = task({ nextRunAt: null, schedule: { type: "webhook", signature: null } }); + expect(scheduledTasksWakeThread([webhook], threadId)).toBe(true); + }); + + it("ignores tasks that cannot run into this thread again", () => { + const tasks = [ + task({ threadId: ThreadId.make("other-thread") }), + task({ threadId: null }), + task({ enabled: false, nextRunAt: null }), + task({ nextRunAt: null, schedule: { type: "once", runAt: "2026-06-19T00:00:00.000Z" } }), + task({ blockedByPolicy: "Webhooks are disabled by policy." }), + ]; + expect(scheduledTasksWakeThread(tasks, threadId)).toBe(false); + }); +}); diff --git a/packages/client-runtime/src/state/threadInbox.ts b/packages/client-runtime/src/state/threadInbox.ts index 2669416db42fe..64c289796ad6e 100644 --- a/packages/client-runtime/src/state/threadInbox.ts +++ b/packages/client-runtime/src/state/threadInbox.ts @@ -1,4 +1,6 @@ // @effect-diagnostics globalTimers:off - Shelf placement timers run in web and native clients, outside an Effect runtime. +import type { ScheduledTask, ThreadId } from "@t3tools/contracts"; + import { threadRuntimeIsActive, type EnvironmentThreadShell } from "./models.ts"; import { toSortableTimestamp } from "./threadSort.ts"; import { effectiveSnoozed } from "./threadSettled.ts"; @@ -30,6 +32,57 @@ export function isThreadWorking(thread: WorkingThreadInput): boolean { return !(thread.interactionMode === "plan" && thread.hasActionableProposedPlan && runSettled); } +type PendingWakeThreadInput = Pick< + EnvironmentThreadShell, + | "hasPendingApprovals" + | "hasPendingUserInput" + | "pendingSubagentRequest" + | "pendingGuestMessageCount" + | "pendingBackgroundTasks" + | "runtime" + | "hasHeldQueue" +>; + +const ENDED_RUNTIME_STATUSES = new Set["status"]>([ + "completed", + "interrupted", + "cancelled", +]); + +/** + * The latest turn ended and nothing T3 knows of can start another. Any running background task + * counts, since some providers wake on unawaited ones. Failed and rolled-back turns keep their + * own state. `hasScheduledWake` is null while the scheduled task list is unknown. + */ +export function threadHasNoPendingWake( + thread: PendingWakeThreadInput, + hasScheduledWake: boolean | null, +): boolean { + if (thread.runtime === null || !ENDED_RUNTIME_STATUSES.has(thread.runtime.status)) return false; + if (thread.hasPendingApprovals || thread.hasPendingUserInput) return false; + if (thread.pendingSubagentRequest != null || (thread.pendingGuestMessageCount ?? 0) > 0) { + return false; + } + // A held queue keeps its messages out of the latest run, so the earlier turn still reads ended. + // Older servers omit it, and an unknown queue never reads as idle. + if (thread.hasHeldQueue !== false) return false; + return thread.pendingBackgroundTasks.length === 0 && hasScheduledWake === false; +} + +export function scheduledTasksWakeThread( + tasks: ReadonlyArray< + Pick + >, + threadId: ThreadId, +): boolean { + return tasks.some( + (task) => + task.threadId === threadId && + task.blockedByPolicy === undefined && + (task.nextRunAt !== null || (task.enabled && task.schedule.type === "webhook")), + ); +} + type InboxThreadInput = Pick< EnvironmentThreadShell, "id" | "environmentId" | "createdAt" | "unsettledAt" | "latestRun" diff --git a/packages/client-runtime/src/state/threadSettled.ts b/packages/client-runtime/src/state/threadSettled.ts index fa1c33fb8341f..ca23042b257c3 100644 --- a/packages/client-runtime/src/state/threadSettled.ts +++ b/packages/client-runtime/src/state/threadSettled.ts @@ -145,6 +145,14 @@ export function effectiveSnoozed( return !threadRaisedHandWhileSnoozed(shell); } +/** Neither settled nor snoozed, so the thread sits in the active list, as the sidebar sorts it. */ +export function isThreadListedActive( + shell: ThreadSnoozeShell & Pick, + options: { readonly now: string }, +): boolean { + return shell.settledOverride !== "settled" && !effectiveSnoozed(shell, options); +} + export function threadWokeAt( shell: ThreadSnoozeShell, options: { readonly now: string }, diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 785f41e3aa0fb..9ce85686e66be 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -2094,6 +2094,8 @@ export const OrchestrationV2ThreadShell = Schema.Struct({ latestRunStartedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), latestRunCompletedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), latestRunStopRequested: Schema.optional(Schema.Boolean), + /** Queued messages wait, held after a failure or a stop, until released. Absent from older servers. */ + hasHeldQueue: Schema.optional(Schema.Boolean), activeRunId: Schema.NullOr(RunId), activityRunStartedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), activityRunStatus: Schema.optional(