"use client"; import { useEffect, useRef } from "react"; import { useQueryClient, type InfiniteData, type QueryClient } from "@tanstack/react-query"; import type { WSClient } from "../api/ws-client"; import type { StoreApi, UseBoundStore } from "zustand"; import type { AuthState } from "../auth/store"; import { createLogger } from "../logger"; import { clearWorkspaceStorage } from "../platform/storage-cleanup"; import { defaultStorage } from "../platform/storage"; import { getCurrentWsId, getCurrentSlug } from "../platform/workspace-storage"; import { issueKeys } from "../issues/queries"; import { projectKeys } from "../projects/queries"; import { pinKeys } from "../pins/queries"; import { autopilotKeys } from "../autopilots/queries"; import { runtimeKeys } from "../runtimes/queries"; import { labelKeys } from "../labels/queries"; import { propertyKeys } from "../properties/queries"; import { agentTaskSnapshotKeys, workspaceWorkingAgentsKeys, agentActivityKeys, agentRunCountsKeys, agentTasksKeys, } from "../agents/queries"; import { githubKeys } from "../github/queries"; import { larkKeys } from "../lark/queries"; import { slackKeys } from "../slack/queries"; import { onIssueCreated, onIssueUpdated, onIssueDeleted, onIssueLabelsChanged, onIssuePropertiesChanged, onIssueMetadataChanged, } from "../issues/ws-updaters"; import { invalidateUpdatedAtSortedIssueLists } from "../issues/cache-coordinator"; import { onInboxNew, onInboxInvalidate, onInboxIssueStatusChanged, onInboxIssueDeleted, onInboxSummaryInvalidate } from "../inbox/ws-updaters"; import { inboxKeys } from "../inbox/queries"; import { notificationPreferenceOptions, notificationPreferenceKeys, } from "../notification-preferences/queries"; import { workspaceKeys, workspaceListOptions } from "../workspace/queries"; import { isWorkspaceDeletePending } from "../workspace/pending-delete"; import { showWebNotification, type SystemNotificationPayload, } from "../platform/system-notification"; import type { Workspace } from "../types/workspace"; import { chatKeys, mergeTaskMessagesBySeq, sortChatSessions } from "../chat/queries"; import { useChatStore } from "../chat"; import { resolvePostAuthDestination, useHasOnboarded } from "../paths"; import type { MemberAddedPayload, WorkspaceDeletedPayload, WorkspaceUpdatedPayload, MemberRemovedPayload, IssueUpdatedPayload, IssueCreatedPayload, IssueDeletedPayload, IssueLabelsChangedPayload, IssueMetadataChangedPayload, IssuePropertiesChangedPayload, InboxNewPayload, InboxItem, NotificationPreferenceResponse, CommentCreatedPayload, CommentUpdatedPayload, CommentDeletedPayload, CommentResolvedPayload, CommentUnresolvedPayload, ActivityCreatedPayload, ReactionAddedPayload, ReactionRemovedPayload, IssueReactionAddedPayload, IssueReactionRemovedPayload, SubscriberAddedPayload, SubscriberRemovedPayload, TaskMessagePayload, TaskQueuedPayload, TaskDispatchPayload, TaskRunningPayload, TaskWaitingLocalDirectoryPayload, TaskCompletedPayload, TaskFailedPayload, TaskCancelledPayload, ChatDonePayload, ChatCancelFinalizedPayload, ChatMessage, ChatPendingTask, ChatMessagesPage, ChatSession, InvitationCreatedPayload, } from "../types"; const chatWsLogger = createLogger("chat.ws"); const logger = createLogger("realtime-sync"); export function invalidateChatMessageQueries( qc: QueryClient, sessionId: string, ) { qc.invalidateQueries({ queryKey: chatKeys.messages(sessionId) }); qc.invalidateQueries({ queryKey: chatKeys.messagesPage(sessionId) }); } // refetchPendingChatAggregate marks the current user's cross-session pending // aggregate stale so it is refetched from the permission-filtering endpoint // (/api/chat/pending-tasks[/has-any]). // // SECURITY (review on PR #5018 / MUL-4159): this is deliberately an // invalidate, NOT an optimistic setQueryData. Chat `task:*` events are a // workspace fanout delivered to every member with no creator / agent // visibility in the payload, so optimistically writing the aggregate from them // would let one member's task flip another member's FAB to has_pending=true, // bypassing the server-side permission filter. Invalidation forces the // authoritative, creator+agent-scoped server response to be the source of // truth. The has-any key is nested under pendingTasks, so invalidating // pendingTasks refreshes both the detailed list and the boolean fast-path. export function refetchPendingChatAggregate( qc: QueryClient, wsId: string | null | undefined, ) { if (!wsId) return; qc.invalidateQueries({ queryKey: chatKeys.pendingTasks(wsId) }); } export function applyChatDoneToCache( qc: QueryClient, payload: ChatDonePayload, ) { const sessionId = payload.chat_session_id; const taskId = payload.task_id; const messageId = payload.message_id; const content = payload.content; if (messageId && content !== undefined) { const assistant: ChatMessage = { id: messageId, chat_session_id: sessionId, role: "assistant", content, task_id: taskId, created_at: payload.created_at ?? new Date().toISOString(), elapsed_ms: payload.elapsed_ms ?? null, // Carry the kind so a no_response turn renders its placeholder inline // without waiting for the reconciling refetch (MUL-4351). Missing → // "message" for older servers. message_kind: payload.message_kind ?? "message", }; qc.setQueryData( chatKeys.messages(sessionId), (old) => { if (!old) return old; // first fetch will pick it up // Idempotent against reconnect replay. if (old.some((m) => m.id === messageId)) return old; return [...old, assistant]; }, ); qc.setQueryData | undefined>( chatKeys.messagesPage(sessionId), (old) => patchLatestChatMessagePage(old, assistant), ); } // Replacement is in the messages list now; safe to drop pending. qc.setQueryData(chatKeys.pendingTask(sessionId), {}); // Authoritative refetch reconciles redaction / migrations / clients // that took the fallback branch above. invalidateChatMessageQueries(qc, sessionId); qc.invalidateQueries({ queryKey: chatKeys.pendingTask(sessionId) }); } function patchLatestChatMessagePage( old: InfiniteData | undefined, message: ChatMessage, ): InfiniteData | undefined { if (!old?.pages.length) return old; const seen = old.pages.some((page) => page.messages.some((m) => m.id === message.id)); if (seen) return old; return { ...old, pages: old.pages.map((page, index) => { if (index !== 0) return page; return { ...page, messages: [...page.messages, message], }; }), }; } type ChatSessionUpdatedPayload = { chat_session_id: string; title?: string; project_id?: string | null; pinned?: boolean; status?: "active" | "archived"; updated_at?: string; }; /** * Patch the cached sessions row for a `chat:session_updated` event (rename, * pin/unpin, archive/unarchive from any tab/device) instead of refetching the * whole list. `pinned` is present only on pin/unpin events and `status` only on * archive/unarchive; a plain rename omits both, so absent fields leave existing * state untouched. When either changes we re-sort so the row lands in the right * place (pin → top; archive → the other list) like the server order. * * Archiving MUST also zero the row's unread here: the server payload carries * only status/updated_at, and chatSessionsOptions is `staleTime: Infinity`, so a * stale cache in another tab/device would otherwise keep an archived session's * unread badge lit forever — the same MUL-4360 stuck-badge bug, one surface over. * This mirrors the archive mutation's optimistic patch and the backend deriving * unread_count=0 for archived rows. Unarchive does NOT fabricate a count — the * true unread state comes back from the server refetch (last_read_at is * untouched), so we leave the row's unread fields as-is for `active`. */ export function applyChatSessionUpdatedToCache( qc: QueryClient, wsId: string, payload: ChatSessionUpdatedPayload, ): void { qc.setQueryData(chatKeys.sessions(wsId), (old) => { if (!old) return old; const next = old.map((s) => s.id === payload.chat_session_id ? { ...s, title: payload.title ?? s.title, ...("project_id" in payload ? { project_id: payload.project_id } : {}), pinned: payload.pinned ?? s.pinned, status: payload.status ?? s.status, updated_at: payload.updated_at ?? s.updated_at, ...(payload.status === "archived" ? { unread_count: 0, has_unread: false } : {}), } : s, ); return payload.pinned === undefined && payload.status === undefined ? next : sortChatSessions(next); }); } function removeChatMessageFromPageCache( qc: QueryClient, sessionId: string, messageId: string, ) { qc.setQueryData | undefined>( chatKeys.messagesPage(sessionId), (old) => { if (!old) return old; return { ...old, pages: old.pages.map((page) => ({ ...page, messages: page.messages.filter((m) => m.id !== messageId), })), }; }, ); } export function removeChatMessageFromCaches( qc: QueryClient, sessionId: string, messageId: string, ) { qc.setQueryData( chatKeys.messages(sessionId), (old) => old?.filter((m) => m.id !== messageId) ?? old, ); removeChatMessageFromPageCache(qc, sessionId, messageId); } /** * Apply a chat:cancel_finalized event (#5219): the deferred outcome of a * cancelled chat task, settled after the daemon's transcript flush. * * - outcome "stopped": a late "Stopped." assistant row was persisted — * insert it exactly like a chat:done message. * - outcome "restored": the triggering user message was deleted — drop it * from the caches. The deleted prompt itself never rides this * workspace-wide broadcast: it is durable server-side and only the * initiator's client refetches it through the creator-authorized * draft-restores query, which the session's composer applies and consumes. * * The draft-restores invalidation is gated to the task's initiator and fails * closed when initiator_user_id is missing — nothing is lost either way, * because the durable restore is fetched again on the next composer mount or * network reconnect. Cache patches stay unconditional — they are no-ops for * anyone not viewing the session. */ export function applyChatCancelFinalizedToCache( qc: QueryClient, payload: ChatCancelFinalizedPayload, currentUserId?: string, ) { const sessionId = payload.chat_session_id; if (!sessionId) return; if (payload.outcome === "stopped") { applyChatDoneToCache(qc, { chat_session_id: sessionId, task_id: payload.task_id, message_id: payload.message_id, content: payload.content, elapsed_ms: payload.elapsed_ms, created_at: payload.created_at, message_kind: payload.message_kind, }); return; } if (payload.outcome === "restored") { if (payload.message_id) { removeChatMessageFromCaches(qc, sessionId, payload.message_id); } qc.setQueryData(chatKeys.pendingTask(sessionId), {}); invalidateChatMessageQueries(qc, sessionId); const isInitiator = !!payload.initiator_user_id && !!currentUserId && payload.initiator_user_id === currentUserId; if (isInitiator) { void qc.invalidateQueries({ queryKey: chatKeys.draftRestores(sessionId) }); } } } /** * Apply a workspace:updated event directly to the cached workspace list. * If the incoming `issue_prefix` differs from what's currently cached, also * invalidates issueKeys.all for that workspace, since every issue's rendered * identifier (`MUL-123`) is recomputed from the workspace prefix at read time. * * If the workspace isn't in the cached list (first observation), we * conservatively invalidate — the prefix is effectively "new" relative to * what's cached, so any issues already loaded under the old prefix would * be stale anyway. */ export function applyWorkspaceUpdatedToCache( qc: QueryClient, payload: WorkspaceUpdatedPayload, ): void { const next = payload.workspace; if (next?.id) { const list = qc.getQueryData(workspaceKeys.list()); const cached = list?.find((w) => w.id === next.id) ?? null; if (cached && cached.issue_prefix !== next.issue_prefix) { qc.invalidateQueries({ queryKey: issueKeys.all(next.id) }); } if (cached && list) { qc.setQueryData( workspaceKeys.list(), list.map((workspace) => (workspace.id === next.id ? next : workspace)), ); return; } // Do not seed an absent list with one workspace: staleTime is Infinity, // so doing so would hide every other membership until a hard refresh. qc.invalidateQueries({ queryKey: issueKeys.all(next.id) }); } qc.invalidateQueries({ queryKey: workspaceKeys.list() }); } /** * Resolves the slug of the workspace an inbox item originated from, via the * cached workspace list (fetched once when the cache is cold). * * Desktop notification routing must pin to the *source* workspace of the * inbox item, not the currently active one: the user can be on workspace B * when an `inbox:new` for workspace A arrives, and macOS Notification Center * holds banners across workspace switches. Returns null when the workspace * cannot be resolved — callers must NOT fall back to the current slug (that * recreates the wrong-workspace routing this exists to prevent, #3766) and * should show the notification without a deep link instead. */ export async function resolveInboxSourceSlug( qc: QueryClient, workspaceId: string, ): Promise { if (!workspaceId) return null; try { const workspaces = await qc.ensureQueryData(workspaceListOptions()); return workspaces?.find((w) => w.id === workspaceId)?.slug ?? null; } catch { // Workspace list unavailable (e.g. network hiccup): degrade to a // link-less notification rather than guessing a slug. return null; } } /** * Handles an `inbox:new` event end-to-end: inbox cache invalidation, the * focus / mute checks, and the native OS banner. Exported so the handler * behavior (not just slug resolution) is testable. * * Every workspace-scoped read here keys on the ITEM's workspace * (`item.workspace_id`), never the currently active one (#3766): the cache * invalidation must refresh the source workspace's inbox list / unread * count / dock badge, the mute check must honor the source workspace's * preference, and the deep link must carry the source workspace's slug. */ export async function handleInboxNew( qc: QueryClient, item: InboxItem, ): Promise { const sourceWsId = item.workspace_id; if (sourceWsId) onInboxNew(qc, sourceWsId, item); // A new item in ANY workspace can light the workspace-switcher dot, so // refresh the cross-workspace summary regardless of the active workspace. onInboxSummaryInvalidate(qc); // Fire a native OS notification only when the app isn't focused. When // the user is already looking at Multica, the inbox sidebar's unread // styling is enough — no need to interrupt with a banner. `desktopAPI` // is injected by the preload script; its absence (web app) skips silently. if (typeof document !== "undefined" && document.hasFocus()) return; // Resolve the source workspace's slug once: it pins BOTH the mute check // and the deep link to the workspace the inbox item BELONGS to, never the // currently active one. Reading `getCurrentSlug()` here was the source of // wrong-workspace routing (#3766): an `inbox:new` from workspace A arriving // while workspace B is active emitted a notification carrying B's slug and // A's issue id, deep-linking to an issue B doesn't have. const slug = await resolveInboxSourceSlug(qc, sourceWsId); // Respect the SOURCE workspace's system-notification preference. Keying the // query on `sourceWsId` is not enough: the request resolves its workspace // from the `X-Workspace-Slug` header, which follows the ACTIVE workspace — // so a cold-cache lookup while viewing B would read B's mute setting and // cache it under A's key. Passing the source slug scopes the fetch to A. // When the slug can't be resolved we read only an already-warm cache // (populated earlier with the correct workspace context) rather than fetch // with the wrong one; on network failure we fall through to the default // ("all") rather than swallow the banner. if (sourceWsId) { try { const prefData = slug ? await qc.ensureQueryData( notificationPreferenceOptions(sourceWsId, slug), ) : qc.getQueryData( notificationPreferenceKeys.all(sourceWsId), ); if (prefData?.preferences?.system_notifications === "muted") return; } catch { // Fall through with default behavior. } } // `issueKey` matches the inbox page's URL selector (issue id when the // item is attached to an issue, otherwise the inbox item id). `itemId` // is the inbox row's own id, needed to fire markInboxRead on click. // A null slug (workspace list unavailable / item from a workspace this // client can't see) still shows the banner — the user should learn about // the inbox item — but with an empty slug so the click is a no-op // (the inbox bridge ignores empty slugs) instead of routing wrong. const payload: SystemNotificationPayload = { slug: slug ?? "", itemId: item.id, issueKey: item.issue_id ?? item.id, title: item.title, body: item.body ?? "", }; const desktopAPI = ( globalThis as unknown as { desktopAPI?: { showNotification?: (payload: SystemNotificationPayload) => void; }; } ).desktopAPI; if (desktopAPI?.showNotification) { // Desktop: native OS banner rendered by the Electron main process. desktopAPI.showNotification(payload); return; } // Web: the browser Notification API. No-op without granted permission or on // SSR — the in-app inbox + unread badge still reflect the new item. showWebNotification(payload); } /** * Invalidates all workspace-scoped queries. Used after reconnect and when a * new WSClient instance is detected (workspace switch) to recover events * missed while disconnected. */ function invalidateWorkspaceScopedQueries(qc: QueryClient): void { const wsId = getCurrentWsId(); if (wsId) { qc.invalidateQueries({ queryKey: issueKeys.all(wsId) }); qc.invalidateQueries({ queryKey: inboxKeys.all(wsId) }); qc.invalidateQueries({ queryKey: workspaceKeys.agents(wsId) }); qc.invalidateQueries({ queryKey: workspaceKeys.members(wsId) }); qc.invalidateQueries({ queryKey: workspaceKeys.squads(wsId) }); qc.invalidateQueries({ queryKey: workspaceKeys.skills(wsId) }); qc.invalidateQueries({ queryKey: workspaceKeys.invitations(wsId) }); qc.invalidateQueries({ queryKey: projectKeys.all(wsId) }); qc.invalidateQueries({ queryKey: runtimeKeys.all(wsId) }); qc.invalidateQueries({ queryKey: autopilotKeys.all(wsId) }); qc.invalidateQueries({ queryKey: agentTaskSnapshotKeys.all(wsId) }); qc.invalidateQueries({ queryKey: workspaceWorkingAgentsKeys.all(wsId) }); qc.invalidateQueries({ queryKey: agentActivityKeys.all(wsId) }); qc.invalidateQueries({ queryKey: agentRunCountsKeys.all(wsId) }); qc.invalidateQueries({ queryKey: chatKeys.all(wsId) }); qc.invalidateQueries({ queryKey: labelKeys.all(wsId) }); qc.invalidateQueries({ queryKey: propertyKeys.all(wsId) }); } // Cross-workspace, so outside the wsId guard: a reconnect may have missed // inbox events from any workspace, so re-pull the switcher-dot summary. onInboxSummaryInvalidate(qc); // Per-issue caches are keyed without wsId, so the issueKeys.all(wsId) // prefix above does not reach them. They rely entirely on WS events for // freshness (staleTime: Infinity), so events missed while disconnected // left them stale until a full reload — the inbox showed an agent's new // comment while the issue timeline didn't (#3953). Inactive caches only // get marked stale here and refetch on next mount; the one mounted issue // refetches immediately, same as its own useWSReconnect already does. qc.invalidateQueries({ queryKey: issueKeys.timelineAll() }); qc.invalidateQueries({ queryKey: issueKeys.reactionsAll() }); qc.invalidateQueries({ queryKey: issueKeys.subscribersAll() }); qc.invalidateQueries({ queryKey: issueKeys.usageAll() }); qc.invalidateQueries({ queryKey: issueKeys.attachmentsAll() }); qc.invalidateQueries({ queryKey: issueKeys.tasksAll() }); // Per-chat-session caches are also keyed without wsId, so the // chatKeys.all(wsId) prefix above only reaches session lists / aggregates. // Message streams rely on WS invalidation with staleTime: Infinity; recover // sessions that missed chat/task events while the socket was disconnected. qc.invalidateQueries({ queryKey: chatKeys.messagesAll() }); qc.invalidateQueries({ queryKey: chatKeys.messagesPageAll() }); qc.invalidateQueries({ queryKey: chatKeys.pendingTaskAll() }); qc.invalidateQueries({ queryKey: chatKeys.taskMessagesAll() }); // A chat:cancel_finalized broadcast missed while disconnected is exactly // what the durable draft-restore rows exist for (#5219) — re-pull them so // a mounted composer recovers the prompt without a remount. qc.invalidateQueries({ queryKey: chatKeys.draftRestoresAll() }); qc.invalidateQueries({ queryKey: workspaceKeys.list() }); } function invalidateSquadMemberStatusQueries(qc: QueryClient, wsId: string): void { qc.invalidateQueries({ predicate: (query) => { const key = query.queryKey; return ( key[0] === "workspaces" && key[1] === wsId && key[2] === "squads" && key[4] === "members-status" ); }, }); } export interface RealtimeSyncStores { authStore: UseBoundStore>; } /** * Centralized WS -> store sync. Called once from WSProvider. * * Uses the "WS as invalidation signal + refetch" pattern: * - onAny handler extracts event prefix and calls the matching store refresh * - Debounce per-prefix prevents rapid-fire refetches (e.g. bulk issue updates) * - Precise handlers only for side effects (toast, navigation, self-check) * * Per-issue events (comments, activity, reactions, subscribers) are handled * both here (invalidation fallback) and by per-page useWSEvent hooks (granular * updates). Daemon register events invalidate runtimes globally; heartbeats * are skipped to avoid excessive refetches. * * @param ws - WebSocket client instance (null when not yet connected) * @param stores - Platform-created Zustand store instances for auth and workspace * @param onToast - Optional callback for showing toast messages (platform-specific) */ export function useRealtimeSync( ws: WSClient | null, stores: RealtimeSyncStores, onToast?: (message: string, type?: "info" | "error") => void, ) { const { authStore } = stores; const qc = useQueryClient(); // Captured via ref so the (rare) hasOnboarded change doesn't re-subscribe // every WS handler in this effect. The resolver reads `.current` at the // moment workspace-loss fires, which is what we want. const hasOnboarded = useHasOnboarded(); const hasOnboardedRef = useRef(hasOnboarded); hasOnboardedRef.current = hasOnboarded; // Main sync: onAny -> refreshMap with debounce useEffect(() => { if (!ws) return; const refreshMap: Record void> = { inbox: () => { const wsId = getCurrentWsId(); if (wsId) onInboxInvalidate(qc, wsId); // inbox:read / inbox:archived / inbox:unarchived / batch events arrive // here. They can originate from a workspace other than the active one // (personal events fan out to all the user's connections), so always // refresh the cross-workspace summary — its dot must clear when another // workspace's items are read/archived, and light again when an unread // item is restored from the archive. onInboxSummaryInvalidate(qc); }, agent: () => { const wsId = getCurrentWsId(); if (wsId) { qc.invalidateQueries({ queryKey: workspaceKeys.agents(wsId) }); qc.invalidateQueries({ queryKey: workspaceWorkingAgentsKeys.all(wsId) }); // Squad members status is derived per agent, so any agent // change (status flip, archive, runtime swap) needs to refresh the // per-squad members-status cache without refetching the static squad // list summary. invalidateSquadMemberStatusQueries(qc, wsId); } }, member: () => { const wsId = getCurrentWsId(); if (wsId) qc.invalidateQueries({ queryKey: workspaceKeys.members(wsId) }); }, // workspace:updated is handled by the specific handler below // (compares prefixes to decide whether to also invalidate issues). // This generic fallback still fires for workspace:deleted (paired // with the specific navigation handler) and any future workspace:* // events without dedicated handlers. workspace: () => { qc.invalidateQueries({ queryKey: workspaceKeys.list() }); }, skill: () => { const wsId = getCurrentWsId(); if (wsId) qc.invalidateQueries({ queryKey: workspaceKeys.skills(wsId) }); }, project: () => { const wsId = getCurrentWsId(); if (wsId) qc.invalidateQueries({ queryKey: projectKeys.all(wsId) }); }, squad: () => { const wsId = getCurrentWsId(); if (wsId) { qc.invalidateQueries({ queryKey: workspaceKeys.squads(wsId) }); // squad:deleted triggers assignee transfer — refresh issues too. qc.invalidateQueries({ queryKey: issueKeys.all(wsId) }); } }, label: () => { // Label catalogs are independently scoped to issues, agents, and // skills. The generic event prefix does not carry the scope into this // dispatcher, so refresh all three resource projections. Issue rows // embed label snapshots; agent/skill label pickers use their resource // cache plus the shared label query tree. const wsId = getCurrentWsId(); if (wsId) { qc.invalidateQueries({ queryKey: ["labels", wsId] }); qc.invalidateQueries({ queryKey: issueKeys.all(wsId) }); qc.invalidateQueries({ queryKey: workspaceKeys.agents(wsId) }); qc.invalidateQueries({ queryKey: workspaceKeys.skills(wsId) }); } }, pin: () => { const wsId = getCurrentWsId(); const userId = authStore.getState().user?.id; if (wsId && userId) qc.invalidateQueries({ queryKey: pinKeys.all(wsId, userId) }); }, daemon: () => { const wsId = getCurrentWsId(); if (wsId) { qc.invalidateQueries({ queryKey: runtimeKeys.all(wsId) }); // Runtime online/offline transitions move the derived status // for every agent that hosts on this runtime, which shifts the // working/idle/offline pill on the squad page. invalidateSquadMemberStatusQueries(qc, wsId); } }, autopilot: () => { const wsId = getCurrentWsId(); if (wsId) qc.invalidateQueries({ queryKey: autopilotKeys.all(wsId) }); }, github_installation: () => { const wsId = getCurrentWsId(); if (wsId) qc.invalidateQueries({ queryKey: githubKeys.installations(wsId) }); }, lark_installation: () => { const wsId = getCurrentWsId(); if (wsId) qc.invalidateQueries({ queryKey: larkKeys.installations(wsId) }); }, slack_installation: () => { const wsId = getCurrentWsId(); if (wsId) qc.invalidateQueries({ queryKey: slackKeys.installations(wsId) }); }, vcs_connection: () => { const wsId = getCurrentWsId(); if (wsId) qc.invalidateQueries({ queryKey: ["vcs", wsId] }); }, pull_request: () => { // PR list is keyed by issue id, not workspace, so we invalidate all // PR queries — the open issue detail page will refetch its own list. qc.invalidateQueries({ queryKey: ["github", "pull-requests"] }); }, // Powers the agent presence cache: any task lifecycle change // (dispatch / completed / failed / cancelled) refreshes the // workspace-wide agent-task-snapshot query so per-agent presence // reflects the change. task:message is NOT in this prefix path — it // stays in specificEvents to avoid an invalidate storm during long runs. task: () => { const wsId = getCurrentWsId(); if (!wsId) return; qc.invalidateQueries({ queryKey: agentTaskSnapshotKeys.list(wsId) }); qc.invalidateQueries({ queryKey: workspaceWorkingAgentsKeys.all(wsId) }); // The Table working-agent shortcut derives an assignee set from the // projection above. Refresh its server-owned graph alongside that set // so rows/groups/facets cannot remain on an old task transition while // the projection refetches (global staleTime is Infinity). qc.invalidateQueries({ queryKey: issueKeys.tableAll(wsId) }); // 30d activity series shares the same lifecycle signal — any task // completion / failure shifts the histogram. (Dispatch alone // doesn't change a completed_at-anchored series, but invalidating // here keeps the WS-handler shape uniform; the resulting refetch // is cheap.) Both the list (trailing 7d slice) and the detail // panel read off this single cache. qc.invalidateQueries({ queryKey: agentActivityKeys.last30d(wsId) }); // 30-day run count likewise increments per task lifecycle event. qc.invalidateQueries({ queryKey: agentRunCountsKeys.last30d(wsId) }); // Per-agent task list (Activity tab "Recent work"). Prefix match // catches every agent's list — the per-agent detail key sits // under agentTasks//. qc.invalidateQueries({ queryKey: agentTasksKeys.all(wsId) }); // Per-issue task list (issue-detail Execution log). Prefix match // across all issues — keeps the contract "any task: event makes // every list-of-tasks query stale" so cache stays fresh even // when the relevant component isn't currently mounted. qc.invalidateQueries({ queryKey: ["issues", "tasks"] }); // Per-issue token usage card (issue-detail right rail). Same // shape as the tasks invalidation above — any task lifecycle // event shifts the aggregated usage numbers. qc.invalidateQueries({ queryKey: ["issues", "usage"] }); // Squad members-status reads the same task lifecycle to flip // working ↔ idle for each agent member. invalidateSquadMemberStatusQueries(qc, wsId); // Comment trigger previews answer "who would a send wake right // now" — the pending-task dedup guard makes that answer // queue-dependent, so any task lifecycle change must refresh an // open composer's chips (e.g. an agent finishing its run becomes // triggerable again mid-typing). qc.invalidateQueries({ queryKey: issueKeys.commentTriggerPreviewAll() }); // Issue-trigger previews (assign/status/create/batch) are deliberately // NOT invalidated here. Unlike comment triggers, the assign source // (create / assignee change) cancels existing tasks before enqueuing, so // a task event can never change its verdict; only the status source's // pending dedup could, and that preview is advisory — the write path // re-evaluates authoritatively, so a rare stale label is harmless. // Refetching every mounted preview on every workspace task event caused // visible flicker, so the preview now refetches only on input change // (signature), mirroring its query design (MUL-3375). }, }; const timers = new Map>(); const debouncedRefresh = (prefix: string, fn: () => void) => { const existing = timers.get(prefix); if (existing) clearTimeout(existing); timers.set( prefix, setTimeout(() => { timers.delete(prefix); fn(); }, 100), ); }; // Event types handled by specific handlers below -- skip generic refresh const specificEvents = new Set([ "workspace:updated", "issue:updated", "issue:created", "issue:deleted", "issue_labels:changed", "issue_metadata:changed", "issue_properties:changed", "property:created", "property:updated", "inbox:new", "comment:created", "comment:updated", "comment:deleted", "comment:resolved", "comment:unresolved", "activity:created", "reaction:added", "reaction:removed", "issue_reaction:added", "issue_reaction:removed", "subscriber:added", "subscriber:removed", "daemon:heartbeat", // Chat events are handled explicitly below; do not double-invalidate. "chat:message", "chat:done", "chat:cancel_finalized", "chat:session_read", "chat:session_deleted", "chat:session_updated", // task:message stays out of the prefix path because it fires per // streamed message during a long run — invalidating the snapshot on // every message would flood the network. Specific chat handlers below // still receive it via ws.on() (a separate subscription channel). "task:message", // task:completed / task:failed deliberately NOT here. They go through // both the task-prefix invalidate (refreshes the agent-task-snapshot // cache) AND the chat-specific ws.on() handlers below. The two // channels are independent — onAny dispatch and ws.on are separate // subscriptions. ]); const unsubAny = ws.onAny((msg) => { if (specificEvents.has(msg.type)) return; const prefix = msg.type.split(":")[0] ?? ""; const refresh = refreshMap[prefix]; if (refresh) debouncedRefresh(prefix, refresh); }); // --- Specific event handlers (granular cache updates) --- // No self-event filtering: actor_id identifies the USER, not the TAB. // Filtering by actor_id would block other tabs of the same user. // Instead, both mutations and WS handlers use dedup checks to be idempotent. const unsubIssueUpdated = ws.on("issue:updated", (p) => { const payload = p as IssueUpdatedPayload; const { issue } = payload; if (!issue?.id) return; const wsId = getCurrentWsId(); if (wsId) { onIssueUpdated(qc, wsId, issue, { assigneeChanged: payload.assignee_changed, statusChanged: payload.status_changed, projectChanged: payload.project_changed, }); if (issue.status) { onInboxIssueStatusChanged(qc, wsId, issue.id, issue.status); } } }); const unsubIssueCreated = ws.on("issue:created", (p) => { const { issue } = p as IssueCreatedPayload; if (!issue) return; const wsId = getCurrentWsId(); if (wsId) onIssueCreated(qc, wsId, issue); }); const unsubIssueDeleted = ws.on("issue:deleted", (p) => { const { issue_id } = p as IssueDeletedPayload; if (!issue_id) return; const wsId = getCurrentWsId(); if (wsId) { onIssueDeleted(qc, wsId, issue_id); onInboxIssueDeleted(qc, wsId, issue_id); } }); const unsubIssueLabelsChanged = ws.on("issue_labels:changed", (p) => { const { issue_id, labels } = p as IssueLabelsChangedPayload; if (!issue_id) return; const wsId = getCurrentWsId(); if (wsId) onIssueLabelsChanged(qc, wsId, issue_id, labels ?? []); }); const unsubIssueMetadataChanged = ws.on("issue_metadata:changed", (p) => { const { issue_id, metadata } = p as IssueMetadataChangedPayload; if (!issue_id) return; const wsId = getCurrentWsId(); if (wsId) onIssueMetadataChanged(qc, wsId, issue_id, metadata ?? {}); }); const unsubIssuePropertiesChanged = ws.on("issue_properties:changed", (p) => { const { issue_id, properties } = p as IssuePropertiesChangedPayload; if (!issue_id) return; const wsId = getCurrentWsId(); if (wsId) { onIssuePropertiesChanged(qc, wsId, issue_id, properties ?? {}); // The catalog embeds per-definition usage counts; every value // set/unset shifts them. The list is tiny, so a refetch beats // trying to patch counts client-side. qc.invalidateQueries({ queryKey: propertyKeys.all(wsId) }); } }); // Definition changes (create / rename / options / archive) — refetch the // catalog; issue caches keep raw value bags so they stay valid. const unsubPropertyChanged = ["property:created", "property:updated"].map((event) => ws.on(event as "property:created" | "property:updated", () => { const wsId = getCurrentWsId(); if (wsId) { qc.invalidateQueries({ queryKey: propertyKeys.all(wsId) }); // Group order, supported group types, and unavailable option values // are derived from the property definition, not just issue rows. qc.invalidateQueries({ queryKey: issueKeys.tableAll(wsId) }); } }), ); const unsubInboxNew = ws.on("inbox:new", async (p) => { const { item } = p as InboxNewPayload; if (!item) return; await handleInboxNew(qc, item); }); // --- Timeline event handlers (global fallback) --- // These events are also handled granularly by useIssueTimeline when // IssueDetail is mounted. This global handler exists to mark the // timeline cache stale for issues whose IssueDetail is *not* mounted, // so stale data isn't served on next mount (staleTime: Infinity, set on // the QueryClient default, relies on this). // // `refetchType: "none"` is the load-bearing detail: without it, an // active IssueDetail observer would refetch the entire timeline on // every comment / activity / reaction event. The refetch replaces // every entry's reference and busts React.memo on every CommentCard // subtree (visible during AI streaming as a flash across all sibling // threads, MUL-1941). Inactive observers don't refetch either way; // when IssueDetail mounts later, the stale flag triggers the refetch // through `refetchOnMount`. Active observers stay fresh via the // granular setQueryData handlers in `useIssueTimeline`. const invalidateTimeline = (issueId: string) => { qc.invalidateQueries({ queryKey: issueKeys.timeline(issueId), refetchType: "none", }); }; const unsubCommentCreated = ws.on("comment:created", (p) => { const { comment } = p as CommentCreatedPayload; if (!comment?.issue_id) return; invalidateTimeline(comment.issue_id); // A new comment bumps the parent issue's updated_at server-side // (MUL-5009), so any open board/list sorted by "Updated date" has // drifted. Refetch just those keys to re-sort the commented card into // place; every other sort is untouched. Only comment:created bumps // updated_at, so the other comment events below deliberately do not. const wsId = getCurrentWsId(); if (wsId) invalidateUpdatedAtSortedIssueLists(qc, wsId); }); const unsubCommentUpdated = ws.on("comment:updated", (p) => { const { comment } = p as CommentUpdatedPayload; if (comment?.issue_id) invalidateTimeline(comment.issue_id); }); const unsubCommentDeleted = ws.on("comment:deleted", (p) => { const { issue_id } = p as CommentDeletedPayload; if (issue_id) invalidateTimeline(issue_id); }); const unsubCommentResolved = ws.on("comment:resolved", (p) => { const { comment } = p as CommentResolvedPayload; if (comment?.issue_id) invalidateTimeline(comment.issue_id); }); const unsubCommentUnresolved = ws.on("comment:unresolved", (p) => { const { comment } = p as CommentUnresolvedPayload; if (comment?.issue_id) invalidateTimeline(comment.issue_id); }); const unsubActivityCreated = ws.on("activity:created", (p) => { const { issue_id } = p as ActivityCreatedPayload; if (issue_id) invalidateTimeline(issue_id); }); const unsubReactionAdded = ws.on("reaction:added", (p) => { const { issue_id } = p as ReactionAddedPayload; if (issue_id) invalidateTimeline(issue_id); }); const unsubReactionRemoved = ws.on("reaction:removed", (p) => { const { issue_id } = p as ReactionRemovedPayload; if (issue_id) invalidateTimeline(issue_id); }); // --- Issue-level reactions & subscribers (global fallback) --- const unsubIssueReactionAdded = ws.on("issue_reaction:added", (p) => { const { issue_id } = p as IssueReactionAddedPayload; if (issue_id) qc.invalidateQueries({ queryKey: issueKeys.reactions(issue_id) }); }); const unsubIssueReactionRemoved = ws.on("issue_reaction:removed", (p) => { const { issue_id } = p as IssueReactionRemovedPayload; if (issue_id) qc.invalidateQueries({ queryKey: issueKeys.reactions(issue_id) }); }); const unsubSubscriberAdded = ws.on("subscriber:added", (p) => { const { issue_id } = p as SubscriberAddedPayload; if (issue_id) qc.invalidateQueries({ queryKey: issueKeys.subscribers(issue_id) }); }); const unsubSubscriberRemoved = ws.on("subscriber:removed", (p) => { const { issue_id } = p as SubscriberRemovedPayload; if (issue_id) qc.invalidateQueries({ queryKey: issueKeys.subscribers(issue_id) }); }); // --- Side-effect handlers (toast, navigation) --- // After the current workspace disappears (deleted or we were kicked out), // navigate to another workspace the user still has access to, or to the // create-workspace page. We use a full-page navigation: this reliably // tears down any in-flight queries / subscriptions tied to the dead // workspace without relying on framework-specific routers from here in // core. const relocateAfterWorkspaceLoss = async (lostWsId: string) => { const wsList = await qc.fetchQuery({ ...workspaceListOptions(), staleTime: 0, }); const remaining = wsList.filter((w) => w.id !== lostWsId); const target = resolvePostAuthDestination( remaining, hasOnboardedRef.current, ); if (typeof window !== "undefined") { window.location.assign(target); } }; const unsubWsUpdated = ws.on("workspace:updated", (p) => { applyWorkspaceUpdatedToCache(qc, p as WorkspaceUpdatedPayload); }); const unsubWsDeleted = ws.on("workspace:deleted", (p) => { const { workspace_id } = p as WorkspaceDeletedPayload; // Self-initiated delete: useDeleteWorkspace owns storage cleanup and // navigation (both run after the DELETE resolves). Reacting here too // would race that flow's navigation with a full-page relocate — the // CancelledError + reload combo this guard exists to prevent. This // handler only serves deletes initiated elsewhere (other user/device). if (isWorkspaceDeletePending(workspace_id)) return; // Event payload has UUID; look up slug from cached workspace list // since clearWorkspaceStorage keys are namespaced by slug. const wsList = qc.getQueryData<{ id: string; slug: string }[]>(workspaceKeys.list()) ?? []; const deletedSlug = wsList.find((w) => w.id === workspace_id)?.slug; if (deletedSlug) clearWorkspaceStorage(defaultStorage, deletedSlug); if (getCurrentWsId() === workspace_id) { logger.warn("current workspace deleted, switching"); onToast?.("This workspace was deleted", "info"); relocateAfterWorkspaceLoss(workspace_id); } }); const unsubMemberRemoved = ws.on("member:removed", (p) => { const { user_id } = p as MemberRemovedPayload; const myUserId = authStore.getState().user?.id; if (user_id === myUserId) { const slug = getCurrentSlug(); const wsId = getCurrentWsId(); if (slug && wsId) { clearWorkspaceStorage(defaultStorage, slug); logger.warn("removed from workspace, switching"); onToast?.("You were removed from this workspace", "info"); relocateAfterWorkspaceLoss(wsId); } } }); const unsubMemberAdded = ws.on("member:added", (p) => { const { member, workspace_name } = p as MemberAddedPayload; const myUserId = authStore.getState().user?.id; if (member.user_id === myUserId) { qc.invalidateQueries({ queryKey: workspaceKeys.list() }); qc.invalidateQueries({ queryKey: workspaceKeys.myInvitations() }); onToast?.( `You joined ${workspace_name ?? "a workspace"}`, "info", ); } }); // invitation:created — notify the invitee of a new pending invitation const unsubInvitationCreated = ws.on("invitation:created", (p) => { const { workspace_name } = p as InvitationCreatedPayload; qc.invalidateQueries({ queryKey: workspaceKeys.myInvitations() }); onToast?.( `You were invited to ${workspace_name ?? "a workspace"}`, "info", ); }); // invitation:accepted / declined / revoked — refresh invitation lists const unsubInvitationAccepted = ws.on("invitation:accepted", () => { const currentWsId = getCurrentWsId(); if (currentWsId) { qc.invalidateQueries({ queryKey: workspaceKeys.invitations(currentWsId) }); qc.invalidateQueries({ queryKey: workspaceKeys.members(currentWsId) }); } }); const unsubInvitationDeclined = ws.on("invitation:declined", () => { const currentWsId = getCurrentWsId(); if (currentWsId) { qc.invalidateQueries({ queryKey: workspaceKeys.invitations(currentWsId) }); } }); const unsubInvitationRevoked = ws.on("invitation:revoked", () => { qc.invalidateQueries({ queryKey: workspaceKeys.myInvitations() }); }); // --- Chat / task events (global, survives ChatWindow unmount) --- // // Single source of truth: the Query cache. No Zustand writes here — the // earlier mirror caused a race where the cache and store disagreed // during the invalidate → refetch window and the UI rendered duplicates. // // task:message is written directly into the task-messages cache so the // live timeline updates in place. chat:message / chat:done / // task:completed / task:failed invalidate messages + pending-task so the // DB remains authoritative. const unsubTaskMessage = ws.on("task:message", (p) => { const payload = p as TaskMessagePayload; qc.setQueryData( chatKeys.taskMessages(payload.task_id), (old = []) => mergeTaskMessagesBySeq(old, [payload]), ); chatWsLogger.debug("task:message (global)", { task_id: payload.task_id, seq: payload.seq, type: payload.type, }); }); // Helpers reused by chat lifecycle handlers. // // SECURITY (review on PR #5018 / MUL-4159): chat `task:*` events are a // *workspace fanout* — every member of the workspace receives them — and // the payload carries no creator / agent-visibility. So we must NEVER // optimistically write the cross-session pending AGGREGATE // (chatKeys.pendingTasks / chatKeys.pendingTasksHasAny) from these events: // member B starting a chat task would otherwise flip member A's FAB to // has_pending=true, bypassing the server-side permission filter on // /api/chat/pending-tasks[/has-any]. Instead we authoritatively (debounced) // invalidate the aggregate so it is refetched through the filtering // endpoint, which only returns the caller's own creator-owned, // accessible-agent tasks. // // The per-session `pendingTask` cache IS still written directly by the // handlers below — it is keyed by chat_session_id and only rendered for a // session the user is allowed to open (server-gated), so it is not a // cross-user aggregate leak. // // chat:message is intentionally NOT a trigger (it fires per streamed // message and would re-create the request storm MUL-4159 fixed); the // aggregate is refreshed only on task lifecycle transitions, which are // per-task and low-frequency, then coalesced by the debounce below. let aggregateRefreshTimer: ReturnType | null = null; const invalidatePendingAggregate = () => { if (aggregateRefreshTimer) clearTimeout(aggregateRefreshTimer); aggregateRefreshTimer = setTimeout(() => { aggregateRefreshTimer = null; refetchPendingChatAggregate(qc, getCurrentWsId()); }, 750); }; const invalidateSessionLists = () => { const id = getCurrentWsId(); if (id) qc.invalidateQueries({ queryKey: chatKeys.sessions(id) }); }; const unsubChatMessage = ws.on("chat:message", (p) => { const payload = p as { chat_session_id: string }; chatWsLogger.info("chat:message (global)", { chat_session_id: payload.chat_session_id }); invalidateChatMessageQueries(qc, payload.chat_session_id); qc.invalidateQueries({ queryKey: chatKeys.pendingTask(payload.chat_session_id) }); // NOTE: intentionally does NOT touch the pending aggregate. chat:message // fires per streamed message with no status; the aggregate is maintained // by the task lifecycle handlers below (MUL-4159). }); const unsubChatDone = ws.on("chat:done", (p) => { const payload = p as ChatDonePayload; chatWsLogger.info("chat:done (global)", { task_id: payload.task_id, chat_session_id: payload.chat_session_id, has_message: !!payload.message_id, }); // Inline-insert the assistant message into the messages cache BEFORE // clearing pending-task. Both writes land in the same React render // tick, so ChatMessageList sees `pendingAlreadyPersisted === true` // and the live TimelineView unmounts only after AssistantMessage has // mounted — no flicker window. This applies TkDodo's "combine // setQueryData (active query) + invalidateQueries (others)" pattern // (https://tkdodo.eu/blog/using-web-sockets-with-react-query). // // Falls back to invalidate-only when the server omits the message // payload (older builds). Older clients hitting a newer server also // work: they ignore the extra fields and rely on the invalidate // below, which keeps the old behavior alive. applyChatDoneToCache(qc, payload); // NOTE: the pending aggregate is left to the task:completed / task:failed // handlers (which carry the task_id needed to remove the right entry). // chat:done no longer invalidates it, so a chatty session doesn't refetch // the aggregate on every turn (MUL-4159). // Assistant message just landed → has_unread may have flipped to true. invalidateSessionLists(); }); // Deferred cancellation outcome (#5219): the server settles the // empty/non-empty judgment only after the daemon's transcript flush, so // this event arrives seconds after the cancel HTTP response — nothing // else re-fetches at that point. const unsubChatCancelFinalized = ws.on("chat:cancel_finalized", (p) => { const payload = p as ChatCancelFinalizedPayload; chatWsLogger.info("chat:cancel_finalized (global)", { task_id: payload.task_id, chat_session_id: payload.chat_session_id, outcome: payload.outcome, }); applyChatCancelFinalizedToCache(qc, payload, authStore.getState().user?.id); if (payload.outcome === "stopped") { // A Stopped. assistant row just landed → session previews change. invalidateSessionLists(); } }); // Chat task lifecycle writethrough: keep `chatKeys.pendingTask(sessionId)` // synchronized with the server state machine via setQueryData rather than // invalidate-refetch. Same pattern as task:message — the WS payload // carries everything we need, and an HTTP roundtrip just to read what we // already know would add latency to every stage transition. // // task:queued is emitted by EnqueueChatTask. The optimistic seed in // chat-window.tsx may have already populated the cache with a temporary // id; this handler upgrades it to the real task_id (and reaffirms status // when reconnect replays the event for an already-running task). const unsubTaskQueued = ws.on("task:queued", (p) => { const payload = p as TaskQueuedPayload; if (!payload.chat_session_id) return; qc.setQueryData( chatKeys.pendingTask(payload.chat_session_id), (old) => ({ ...(old ?? {}), task_id: payload.task_id, status: "queued", }), ); invalidatePendingAggregate(); }); // task:dispatch fires when the daemon claims the queued task. The daemon // immediately follows with StartTask, so dispatched→running is sub-second. // We collapse that window by writing "running" directly — the pill jumps // from "Queued" straight to "Thinking", skipping a meaningless "Starting" // frame. Stage decision in TaskStatusPill maps "running" + empty // taskMessages → "Thinking · Ns". const unsubTaskDispatch = ws.on("task:dispatch", (p) => { const payload = p as TaskDispatchPayload; if (!payload.chat_session_id) return; qc.setQueryData( chatKeys.pendingTask(payload.chat_session_id), (old) => { if (!old || old.task_id !== payload.task_id) return old; return { ...old, status: "running" }; }, ); invalidatePendingAggregate(); }); // task:running fires when the daemon transitions a previously-parked task // (waiting_local_directory) back into the run phase. The dispatch→running // path is collapsed in the handler above, so this handler exists mainly to // clear a stale `waiting_local_directory` pill — without it, the pill // would stay parked even after the daemon resumed work. const unsubTaskRunning = ws.on("task:running", (p) => { const payload = p as TaskRunningPayload; if (!payload.chat_session_id) return; qc.setQueryData( chatKeys.pendingTask(payload.chat_session_id), (old) => { if (!old || old.task_id !== payload.task_id) return old; return { ...old, status: "running" }; }, ); invalidatePendingAggregate(); }); // task:waiting_local_directory fires when the daemon dequeues a task but // can't acquire the local_directory path lock — another task on this // daemon is in the same directory. Write the status so TaskStatusPill // can render the "Waiting for local directory" stage instead of pinning // a stale "Starting / Thinking" frame. const unsubTaskWaitingLocalDir = ws.on( "task:waiting_local_directory", (p) => { const payload = p as TaskWaitingLocalDirectoryPayload; if (!payload.chat_session_id) return; qc.setQueryData( chatKeys.pendingTask(payload.chat_session_id), (old) => { if (!old || old.task_id !== payload.task_id) return old; return { ...old, status: "waiting_local_directory" }; }, ); invalidatePendingAggregate(); }, ); // task:cancelled reaches us when: // 1. handleStop already cleared the cache locally (this is a no-op confirm) // 2. another tab / admin / system cancels — this is the only path that // drops the pending pill in those cases. Without it the pill spins // forever in the second-tab scenario. // CancelTask also persists a best-effort assistant snapshot when the // stopped chat task had already streamed transcript rows, so refresh the // message page along with clearing pending. const unsubTaskCancelled = ws.on("task:cancelled", (p) => { const payload = p as TaskCancelledPayload; if (!payload.chat_session_id) return; chatWsLogger.info("task:cancelled (global, chat)", { task_id: payload.task_id, chat_session_id: payload.chat_session_id, }); qc.setQueryData(chatKeys.pendingTask(payload.chat_session_id), {}); invalidateChatMessageQueries(qc, payload.chat_session_id); invalidatePendingAggregate(); }); const unsubTaskCompleted = ws.on("task:completed", (p) => { const payload = p as TaskCompletedPayload; if (!payload.chat_session_id) return; // issue tasks handled elsewhere chatWsLogger.info("task:completed (global, chat)", { task_id: payload.task_id, chat_session_id: payload.chat_session_id, }); // `chat:done` (broadcast immediately before this event in CompleteTask) // already wrote the assistant message into the messages cache and // cleared `chatKeys.pendingTask`. This event is now only responsible // for refreshing the per-user cross-session aggregate that drives the // FAB indicator — `chat:done` is per-session and doesn't carry that // information. invalidatePendingAggregate(); }); const unsubTaskFailed = ws.on("task:failed", (p) => { const payload = p as TaskFailedPayload; if (!payload.chat_session_id) return; chatWsLogger.warn("task:failed (global, chat)", { task_id: payload.task_id, chat_session_id: payload.chat_session_id, }); // FailTask writes a failure chat_message (mirroring CompleteTask's // success message), so this path mirrors the task:completed handler: // clear the pending signal AND invalidate the messages list so the // failure bubble shows up without requiring a page refresh. Pre-#1823 // this branch only flipped pending — the comment "No new message" // was true then, but FailTask now persists a row. qc.setQueryData(chatKeys.pendingTask(payload.chat_session_id), {}); invalidateChatMessageQueries(qc, payload.chat_session_id); qc.invalidateQueries({ queryKey: chatKeys.pendingTask(payload.chat_session_id) }); invalidatePendingAggregate(); // FailTask persisted a failure chat_message, so the thread list's // last_message / unread_count / sort order changed too. Mirror the // chat:done (success) path and refresh the sessions list — otherwise the // left rail keeps a stale preview until the next full refetch. invalidateSessionLists(); }); const unsubChatSessionRead = ws.on("chat:session_read", (p) => { const payload = p as { chat_session_id: string }; chatWsLogger.info("chat:session_read (global)", payload); invalidateSessionLists(); }); // chat:session_updated fires after the creator renames, pins, or archives // a session in any tab/device. Patch the cached row inline so the dropdown // and badges reflect the change without a full sessions-list refetch — see // applyChatSessionUpdatedToCache for why archive must also zero unread. const unsubChatSessionUpdated = ws.on("chat:session_updated", (p) => { const payload = p as ChatSessionUpdatedPayload; chatWsLogger.info("chat:session_updated (global)", payload); const id = getCurrentWsId(); if (!id) return; applyChatSessionUpdatedToCache(qc, id, payload); }); // chat:session_deleted fires after a hard delete. The originating tab has // already optimistically dropped the row via useDeleteChatSession; this // handler keeps OTHER tabs/devices in sync and also clears the active // session pointer so a deleted session doesn't keep the chat window // pointed at vanished messages. const unsubChatSessionDeleted = ws.on("chat:session_deleted", (p) => { const payload = p as { chat_session_id: string }; chatWsLogger.info("chat:session_deleted (global)", payload); const id = getCurrentWsId(); if (id) { const drop = (old?: { id: string }[]) => old?.filter((s) => s.id !== payload.chat_session_id); qc.setQueryData(chatKeys.sessions(id), drop); } qc.removeQueries({ queryKey: chatKeys.messages(payload.chat_session_id) }); qc.removeQueries({ queryKey: chatKeys.pendingTask(payload.chat_session_id) }); invalidatePendingAggregate(); const chatState = useChatStore.getState?.(); if (chatState && chatState.activeSessionId === payload.chat_session_id) { chatState.setActiveSession(null); } }); return () => { unsubAny(); unsubIssueUpdated(); unsubIssueCreated(); unsubIssueDeleted(); unsubIssueLabelsChanged(); unsubIssueMetadataChanged(); unsubIssuePropertiesChanged(); unsubPropertyChanged.forEach((unsub) => unsub()); unsubInboxNew(); unsubCommentCreated(); unsubCommentUpdated(); unsubCommentDeleted(); unsubCommentResolved(); unsubCommentUnresolved(); unsubActivityCreated(); unsubReactionAdded(); unsubReactionRemoved(); unsubIssueReactionAdded(); unsubIssueReactionRemoved(); unsubSubscriberAdded(); unsubSubscriberRemoved(); unsubWsUpdated(); unsubWsDeleted(); unsubMemberRemoved(); unsubMemberAdded(); unsubInvitationCreated(); unsubInvitationAccepted(); unsubInvitationDeclined(); unsubInvitationRevoked(); unsubTaskMessage(); unsubChatMessage(); unsubChatDone(); unsubChatCancelFinalized(); unsubTaskQueued(); unsubTaskDispatch(); unsubTaskRunning(); unsubTaskWaitingLocalDir(); unsubTaskCancelled(); unsubTaskCompleted(); unsubTaskFailed(); unsubChatSessionRead(); unsubChatSessionDeleted(); unsubChatSessionUpdated(); if (aggregateRefreshTimer) clearTimeout(aggregateRefreshTimer); timers.forEach(clearTimeout); timers.clear(); }; }, [ws, qc, authStore, onToast]); // Reconnect -> refetch all data to recover missed events useEffect(() => { if (!ws) return; const unsub = ws.onReconnect(async () => { logger.info("reconnected, refetching all data"); try { invalidateWorkspaceScopedQueries(qc); } catch (e) { logger.error("reconnect refetch failed", e); } }); return unsub; }, [ws, qc]); // New WSClient instance (workspace switch) -> invalidate workspace-scoped // queries to recover events missed while the previous instance was torn down. // Skips the initial assignment to avoid a redundant refetch on first mount. const wsInstanceRef = useRef(null); useEffect(() => { if (!ws) return; if (wsInstanceRef.current === null) { // First non-null instance — store and skip invalidation. wsInstanceRef.current = ws; return; } if (wsInstanceRef.current === ws) return; wsInstanceRef.current = ws; logger.info("new WSClient instance detected, invalidating workspace queries"); invalidateWorkspaceScopedQueries(qc); }, [ws, qc]); }