Files
PaperClipAI/ui/src/context/LiveUpdatesProvider.tsx
T
DottaandPaperclip 505e1a5582 feat(cases): add persistent workflows with same-task agent handoffs (#16959)
## Thinking Path

> - Paperclip manages AI agents and their work.
> - An agent workflow needs persistent facts and a safe way to transfer
work.
> - Creating a task for each phase splits the objective, discussion,
workspace, and review policy.
> - A case can retain the facts while one primary task moves between
agents.
> - Transfers must survive process failure and preserve private access,
budgets, and completion gates.
> - This pull request adds durable case handoffs, native case tools, and
read-only case pages.
> - Prompts describe the workflow. Existing task controls decide when
work can run and finish.

## Linked Issues or Issue Description

**Problem or motivation**

Prompt-driven agent teams need a persistent record that can move through
triage, implementation, verification, and correction. Ordinary
reassignment does not provide a durable instruction, safe release
barrier, or recoverable delivery receipt for an agent handing off its
own running task.

**Proposed solution**

Extend standalone cases with one primary task, a descriptive phase,
retained access sources, and a handoff ledger. Apply assignments after
source execution and workspace settlement. Deliver each transfer through
the existing durable wake queue. Provide `create_case`, `read_case`,
`update_case`, and `handoff_case` in the native runner. Creation
requires kickoff and explicit workflow text. An explicit handoff to
`done` closes the case when normal task completion is accepted.

**Alternatives considered**

Do not add a workflow engine, filesystem queue, configured state graph,
or one task per phase. The workflow document combines transition prose
with an optional Mermaid `stateDiagram-v2` map. The server does not
execute the diagram. The Pipelines experiment removed on master stays
removed.

**Roadmap alignment**

This extends Agent Reviews and Approvals, Enforced Outcomes, Bounded Run
Recovery, and Activity Log attribution. Existing approval, review,
budget, privacy, workspace, and responsible-user rules remain
authoritative.

Related work: [ordinary handoff wake
recovery](https://github.com/paperclipai/paperclip/pull/15019). This
feature adds a separate persisted case-handoff receipt and release
barrier.

## What Changed

- Add case routing fields, retained access sources, durable handoff
history, delivery leases, and replay-safe migrations.
- Keep the same primary task across agent phases. Preserve each
returning agent's own session and the task's workspace.
- Fence outgoing mutations, wait for provider and workspace release, and
recover delivery without duplicate task comments or runs. Recheck
current assignment permissions as the original caller before applying a
delayed transfer. Scope a run Stop to that run’s own handoff intents, so
repeated old Stop requests preserve newer work. Cancelling terminal
intent without a source run also requires fenced current task ownership,
including native retries and preparation with an old checkout. For Cloud
callers, revalidate current stack membership with an uncached Cloud RPC
and derive owner elevation from the current managed setting.
- Keep case reads, documents, assets, Audit, and live events inside
retained task/project permissions.
- Add native case creation/read/update/handoff tools. Store required
workflow text and initial documents on the first task. Page oversized
case facts and referenced document bodies through the dedicated tool
with bounded UTF-8 JSON, current routing versions, snapshot conflicts,
and current source authorization on every page.
- Add read-only case lists and detail pages, task-style list controls,
compact dotted field tables, linked handoff tasks, and case filters in
Audit.
- Add native/legacy integration tests, migration/restart tests, browser
coverage, and bounded live receipt fixtures. Extend the trusted sandbox
condition to cover the new subscription fixture profile.
- Hide obsolete paused-agent warnings on completed or cancelled task
threads.
- Synchronize the company-switch settings test with React effects after
a CI failure. Preserve its security assertions and production remount
behavior.

## Verification

The branch directly includes master `9ef434fe79`. The required `ci /
verify` and `ci / e2e` gates pass on head `765fccd67c`. [Current-head
CI](https://github.com/paperclipai/paperclip/actions/runs/39464348614)
passes typecheck, build, tests, and all eight browser shards. Security
checks pass. [Greptile gives this exact head
5/5](https://github.com/paperclipai/paperclip/pull/16959#issuecomment-6291156549),
with no unresolved review threads. The optional Paperclip Review check
remains queued because its assigned agent is paused. It is not a
required branch check. No check result was overridden.

Local workspace typecheck and build pass after the master merge, along
with database migration checks, token gates, map checks, Storybook
catalog checks, and the Copilot profile identity check. Standard builds
refresh Runner vendor output. Focused checks pass 2,008
API/paging/context tests and 7 strict case protocol tests. The final
commit changes one test fixture only; its ten tests, UI typecheck, and
token gates pass. No production source changed in that commit.

On current head `765fccd67c`, all **194 critical workflow tests**
executed and passed with no skipped tests in those files:

- 38 Stop/ownership tests, including board completion intent, native
retries, stale Stop, and preparation with an old checkout.
- 2 actual legacy process/HTTP handoff tests, including scratch failure,
a closed-stdio child that continues writing, and recovery through a
fresh worker after the process group stops.
- 5 migration tests, including SQL column metadata, preserved null
authority context, and all six prior-numbering receipts.
- 3 process-settlement and 5 scratch tests.
- 88 handoff-service, 18 heartbeat handoff, and 21 native case-tool
integration tests.
- 7 bounded case/document-read projection tests and 7 strict
protocol-contract tests.

The current CI checkout was `fed65d49`, merging `765fccd67c` into master
`9ef434fe79`. Its new document regressions reconstruct exact
Unicode/escaped bodies and revisions through real native authority and
authenticated HTTP with generic API tools disabled. They cover
case-owned, primary-task, and linked-task documents; routing on every
page; stale document snapshots; access loss and unlinking;
unrelated/foreign/private sources; system documents; and quarantined
content. Active-task document tools retain their existing behavior.
These are current-head test results from the first CI attempt; the
browser-only retry did not rerun these suites.

The full `64fdd3403f` CI failed one separate settings test: after a
company switch, it observed an unsettled API-key draft. The production
component and test were unchanged from prior green source. Production
switches remount both keyed settings layers; the test directly reused an
unkeyed view and used a timer tick instead of waiting for React effects.
The final commit replaces that fixture synchronization with React `act`.
The exact secret-clearing assertions and all timeouts remain unchanged.
The original assertion failure remains recorded; no unchanged retry was
used. The narrow unchanged baseline and fixed fixture each pass 10
tests.

The upstream result-paging merge preserves case capture namespaces and
current asset privacy. Its API integration suite executed 39 of 41 tests
successfully, with two skipped; the optional assigned-result PRP
integration test was skipped. Neither skipped suite is counted in the
194 critical tests or as live provider proof.

All 361 upstream migration entries and SQL bodies remain unchanged. Six
case migrations follow master as 0363–0368 with identical SQL hashes.
Official Drizzle generators produced the combined metadata and five
retained snapshots. Generation reports no schema changes, and the
snapshot drift test passes. The earlier `ce361ae5d0` CI passed on
attempt 1, including its then-current 179 critical tests. Fresh review
later identified unusable document references with generic API tools
disabled; the bounded document reads and ten additional authenticated
HTTP tests address that finding.

The full local `pnpm test:run` attempt was cancelled after this Mac
exhausted PostgreSQL semaphore capacity; it is **not** counted as a
pass. Later database attempts also stopped before assertions or skipped,
including one sandbox socket denial. Two original stale-Stop regressions
did reproduce the bug against the old source. A REST serialization
assertion passed after two preserved database-startup failures. No test
timeout was increased. All 194 assertions listed above executed and
passed in current-head CI. The complete 385-file, 3,963-test UI shard
also passed after the fixture repair.

A configured-provider receipt workflow passed 13 backend checks across
triage → implementation → verification → correction → done, using one
task and four work runs. The saved receipt is revision 2 with total 6
USD; case and task both closed. The original canonical attempt remains
failed because its browser assertion confused the terminal destination
with completed-task tooltips. The corrected full browser verifier passed
on the same retained database and original runs, with no additional
provider calls. The last clean-build browser readback is on
`6a86c54a53`; later changes affect recovery, authorization, bounded
reads, and settlement. Accounted work cost is a $1.010574 rate-card
estimate; provider billing and setup-probe usage are unavailable.

Copilot profile 49 passed the unchanged seven-fixture private protocol
pack (46/46 checks, no retries) and the product file-edit canary after
master's completion-replay change. All 60 bound implementation hashes
remain unchanged on the current source. Historical profile-47/48
identities and results remain intact. See the [profile-49 compatibility
record](https://github.com/paperclipai/paperclip/blob/codex/persistent-case-handoffs/doc/plans/2026-10-20-copilot-completion-recovery-compatibility-v49.json).
This is bounded compatibility evidence, not a fresh full matrix,
installed-package product test, or Daytona rollout.

Preserved CI failures and retries:

- On `20662930ac`, Slack managed setup failed a five-second success
assertion after reload. Its test/application path was unchanged from a
prior green head. One unchanged-source retry passed all 25 tests in that
shard; no source or timeout changed.
- On `9402b4f17a`, two CI workers received shutdown signals at 15:24:19
UTC. Neither log had an assertion failure before interruption. One
unchanged-source retry of each passed all Runner static checks and all
32 tests in browser shard 6; both aggregates passed. The next head,
`7b4374edaf`, passed all CI without retries.
- On `295f289b5a`, serialized shard 9 timed out during its first
device-login test, then the next test found an active login lease.
Direct login/test paths were unchanged from a prior green head. Cold
import plus a timed-out async body using the next test's mutable
actor/store is the supported fixture-race explanation; no request trace
proves that sequence. The original failure remains recorded. One
unchanged-source retry passed all 294 tests in 19 files, including all
41 device-login tests. No login source or timeout changed. The
subsequent `ce361ae5d0` CI passed without retries.
- On `64fdd3403f`, one company-switch settings assertion observed an
unsettled draft. The final test-only commit uses React `act` without
changing assertions, timeouts, or production code. On `765fccd67c`, all
ten settings tests and the complete 385-file, 3,963-test UI shard
passed.
- On `765fccd67c`, browser shard 3 completed the split-handoff
assertions, then cleanup received HTTP 500 from run cancellation after a
one-second PostgreSQL row-lock timeout. The extra chat run was reporting
completed work while teardown began. Cleanup stopped before restoring a
global experiment flag, causing a later assertion to fail. Relevant
production and test paths are unchanged from an earlier green commit.
Stop's lock order is consistent, and no nested self-lock was found; the
exact PostgreSQL blocker was not recorded. One unchanged-shard retry
passed. The original failure, trace, and downstream flag failure remain
recorded. No Stop implementation, cleanup assertion, or timeout changed.
The retry supports intermittent cleanup contention; it does not identify
the original blocking transaction.


## Risks

- Local process transfers require host-owned PID/group absence. Missing
or uncertain termination evidence keeps delivery pending, independently
of scratch setup.
- Handoff admission and task cancellation share execution ownership. New
regression tests cover temporary holds, late source writes, and
uncertain provider termination.
- Retained private sources use an intersection of existing permissions.
This intentionally fails closed when a source is missing or an existing
task cannot preserve that boundary.
- Migrations add tables and indexes. They preserve master history and
can replay over prior feature installs. Migration 0368 adds nullable
internal authorization provenance with no default or authority backfill.
Historical transfers still awaiting release without this context fail
closed; an operator must cancel them and request a new handoff.
- Delayed Cloud handoffs depend on the current Cloud membership RPC. An
unavailable, ambiguous, expired, or mismatched response holds the
handoff for retry; it does not reuse cached authority or grant access.
- Bounded reads reload and reauthorize the selected projection per page.
Underlying work grows with the data size and number of pages. Changed
data or routing requires restarting from offset zero without the old
snapshot.
- Phase names and workflow diagrams remain advisory. They do not satisfy
review or approval gates.
- This is a large, cross-cutting PR. The maintainer requested one PR for
the whole feature; the commits separate data, runtime, UI, and
qualification work.

## Model Used

OpenAI Codex, GPT-6 family, with tool use, code execution, and delegated
code review. The exact deployment model ID and context-window size were
not exposed in the session. Live fixture agents used the explicitly
configured provider model recorded in their test evidence.

## Checklist

- [x] I have included a thinking path that traces from project context
to this change
- [x] I have specified the model used (with version and capability
details)
- [x] I have checked ROADMAP.md and confirmed this PR does not duplicate
planned core work
- [x] I have searched GitHub for duplicate or related PRs and linked
them above
- [x] I have either (a) linked existing issues with `Fixes: #` / `Closes
#` / `Refs #` OR (b) described the issue in-PR following the relevant
issue template
- [x] I have not referenced internal/instance-local Paperclip issues or
links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip`
URLs)
- [x] My branch name describes the change (e.g. `docs/...`, `fix/...`)
and contains no internal Paperclip ticket id or instance-derived details
- [x] I have run tests locally and they pass
- [x] I have added or updated tests where applicable
- [x] I have updated relevant documentation to reflect my changes
- [x] I have considered and documented any risks above
- [x] All Paperclip CI gates are green
- [x] Greptile is 5/5 with no open P2s, recommendations, or follow-ups
- [x] I will address all Greptile and reviewer comments before
requesting merge

---------

Co-authored-by: Paperclip <noreply@paperclip.ing>
2026-10-20 12:46:14 -05:00

2201 lines
73 KiB
TypeScript

import { resolveAgentAppearance } from "@paperclipai/shared";
import { getPageVisibility, usePageVisibility } from "../lib/page-visibility";
import {
createContext,
useCallback,
useContext,
useEffect,
useMemo,
useRef,
type ReactNode,
} from "react";
import {
useQuery,
useQueryClient,
type InfiniteData,
type QueryClient,
} from "@tanstack/react-query";
import {
createCoalescingQueryClient,
createInvalidationBatcher,
} from "../lib/query-invalidation-batcher";
import {
patchRunStatusInList,
removeRunFromList,
} from "../lib/live-runs-cache";
import type {
Agent,
HeartbeatRun,
Issue,
IssueComment,
LiveEvent,
} from "@paperclipai/shared";
import type { RunForIssue } from "../api/activity";
import type { ActiveRunForIssue, LiveRunForIssue } from "../api/heartbeats";
import type { CompanyUserDirectoryResponse } from "../api/access";
import type { CaseDetail, IssueCaseLink } from "../api/cases";
import { issuesApi } from "../api/issues";
import { authApi } from "../api/auth";
import type { CompanyListResult } from "../api/companies-query";
import { healthApi } from "../api/health";
import { useCompany } from "./CompanyContext";
import type { ToastActor, ToastInput } from "./ToastContext";
import { useToastActions } from "./ToastContext";
import { upsertIssueCommentInPages } from "../lib/optimistic-issue-comments";
import {
clearIssueExecutionRun,
removeLiveRunById,
} from "../lib/optimistic-issue-runs";
import { queryKeys } from "../lib/queryKeys";
import { extractCompanyPrefixFromPath, toCompanyRelativePath } from "../lib/company-routes";
import { useLocation } from "../lib/router";
import { agentRouteRef } from "../lib/utils";
import { buildSameOriginWebSocketUrl } from "../lib/websocket-url";
import { tryCreateWebSocket } from "../lib/websocket";
const TOAST_COOLDOWN_WINDOW_MS = 10_000;
const TOAST_COOLDOWN_MAX = 3;
const RECONNECT_SUPPRESS_MS = 2000;
const RUN_TOAST_MAX_AGE_MS = 5 * 60_000;
const MAX_OBSERVED_RUN_OUTCOMES = 2000;
const DISCONNECTED_POLL_INTERVAL_MS = 15_000;
const SOCKET_CONNECTING = 0;
const SOCKET_OPEN = 1;
const TERMINAL_RUN_STATUSES = new Set([
"succeeded",
"interrupted",
"failed",
"cancelled",
"timed_out",
]);
type LiveUpdatesSocketLike = {
readyState: number;
onopen: ((this: WebSocket, ev: Event) => unknown) | null;
onmessage: ((this: WebSocket, ev: MessageEvent) => unknown) | null;
onerror: ((this: WebSocket, ev: Event) => unknown) | null;
onclose: ((this: WebSocket, ev: CloseEvent) => unknown) | null;
close: (code?: number, reason?: string) => void;
};
export type CompanyLiveEventHandler = (event: LiveEvent) => void;
interface LiveEventSubscription {
subscribe: (handler: CompanyLiveEventHandler) => () => void;
}
const LiveEventSubscriptionContext =
createContext<LiveEventSubscription | null>(null);
function dispatchLiveEventToSubscribers(
subscribers: Set<CompanyLiveEventHandler>,
expectedCompanyId: string,
event: LiveEvent,
) {
if (event.companyId !== expectedCompanyId) return;
// Snapshot so a handler that (un)subscribes mid-dispatch can't mutate the set
// we're iterating.
for (const handler of Array.from(subscribers)) {
try {
handler(event);
} catch {
// A misbehaving subscriber must never break the shared socket pipeline
// or the toast/invalidation handling that runs alongside it.
}
}
}
/**
* Subscribe to live company events off the single shared LiveUpdates socket.
* Components can react to `heartbeat.run.progress`, `activity.logged`, etc.
* without opening a WebSocket per mount. Events are already filtered to the
* active company. No-ops when rendered outside a LiveUpdatesProvider (e.g. in
* isolated tests), so callers get graceful degradation for free.
*/
export function useCompanyLiveEvent(handler: CompanyLiveEventHandler): void {
const subscription = useContext(LiveEventSubscriptionContext);
const handlerRef = useRef(handler);
useEffect(() => {
handlerRef.current = handler;
});
useEffect(() => {
if (!subscription) return;
return subscription.subscribe((event) => {
handlerRef.current(event);
});
}, [subscription]);
}
function readString(value: unknown): string | null {
return typeof value === "string" && value.length > 0 ? value : null;
}
function readRecord(value: unknown): Record<string, unknown> | null {
if (typeof value !== "object" || value === null || Array.isArray(value))
return null;
return value as Record<string, unknown>;
}
function shortId(value: string) {
return value.slice(0, 8);
}
function resolveAgentName(
queryClient: QueryClient,
companyId: string,
agentId: string,
): string | null {
const agents = queryClient.getQueryData<Agent[]>(
queryKeys.agents.list(companyId),
);
const listedAgent = agents?.find((a) => a.id === agentId);
if (listedAgent?.name) return listedAgent.name;
return (
queryClient.getQueryData<Agent>(queryKeys.agents.detail(agentId))?.name ??
null
);
}
function resolveUserName(
queryClient: QueryClient,
companyId: string,
userId: string,
): string | null {
const directory = queryClient.getQueryData<CompanyUserDirectoryResponse>(
queryKeys.access.companyUserDirectory(companyId),
);
if (!directory) return null;
const entry = directory.users.find((u) => u.principalId === userId);
return entry?.user?.name?.trim() || entry?.user?.email?.trim() || null;
}
function truncate(text: string, max: number): string {
if (text.length <= max) return text;
return text.slice(0, max - 1) + "\u2026";
}
function resolveActorLabel(
queryClient: QueryClient,
companyId: string,
actorType: string | null,
actorId: string | null,
): string {
if (actorType === "agent" && actorId) {
return (
resolveAgentName(queryClient, companyId, actorId) ??
`Agent ${shortId(actorId)}`
);
}
if (actorType === "system") return "System";
if (actorType === "user" && actorId) {
return resolveUserName(queryClient, companyId, actorId) ??
(actorId === "board" || actorId === "local-board" ? "Board" : `User ${shortId(actorId)}`);
}
return actorType === "plugin" ? "Plugin" : "Unknown actor";
}
interface IssueToastContext {
ref: string;
title: string | null;
label: string;
href: string;
}
interface VisibleRouteOptions {
isForegrounded?: boolean;
}
interface VisibleIssueRouteContext {
routeIssueRef: string;
issueRefs: Set<string>;
assigneeAgentId: string | null;
runIds: Set<string>;
subtreeIssueRefs: Set<string>;
subtreeAgentIds: Set<string>;
subtreeRunIds: Set<string>;
}
function resolveIssueQueryRefs(
queryClient: QueryClient,
companyId: string,
issueId: string,
details: Record<string, unknown> | null,
): string[] {
const refs = new Set<string>([issueId]);
const detailIssue = queryClient.getQueryData<Issue>(
queryKeys.issues.detail(issueId),
);
const listIssues = queryClient.getQueryData<Issue[]>(
queryKeys.issues.list(companyId),
);
const detailsIdentifier =
readString(details?.identifier) ?? readString(details?.issueIdentifier);
if (detailsIdentifier) refs.add(detailsIdentifier);
if (detailIssue?.id) refs.add(detailIssue.id);
if (detailIssue?.identifier) refs.add(detailIssue.identifier);
const listIssue = listIssues?.find((issue) => {
if (issue.id === issueId) return true;
if (issue.identifier && issue.identifier === issueId) return true;
if (detailsIdentifier && issue.identifier === detailsIdentifier)
return true;
return false;
});
if (listIssue?.id) refs.add(listIssue.id);
if (listIssue?.identifier) refs.add(listIssue.identifier);
return Array.from(refs);
}
function resolveIssueToastContext(
queryClient: QueryClient,
companyId: string,
issueId: string,
details: Record<string, unknown> | null,
): IssueToastContext {
const issueRefs = resolveIssueQueryRefs(
queryClient,
companyId,
issueId,
details,
);
const detailIssue = issueRefs
.map((ref) => queryClient.getQueryData<Issue>(queryKeys.issues.detail(ref)))
.find((issue): issue is Issue => !!issue);
const listIssue = queryClient
.getQueryData<Issue[]>(queryKeys.issues.list(companyId))
?.find((issue) =>
issueRefs.some((ref) => issue.id === ref || issue.identifier === ref),
);
const cachedIssue = detailIssue ?? listIssue ?? null;
const ref =
readString(details?.identifier) ??
readString(details?.issueIdentifier) ??
cachedIssue?.identifier ??
`Task ${shortId(issueId)}`;
const title =
readString(details?.title) ??
readString(details?.issueTitle) ??
cachedIssue?.title ??
null;
return {
ref,
title,
label: title ? `${ref} - ${truncate(title, 72)}` : ref,
href: `/issues/${cachedIssue?.identifier ?? issueId}`,
};
}
function isPageForegrounded(): boolean {
if (typeof document === "undefined") return false;
if (document.visibilityState !== "visible") return false;
if (typeof document.hasFocus === "function" && !document.hasFocus())
return false;
return true;
}
function resolveVisibleIssueRouteContext(
queryClient: QueryClient,
pathname: string,
options?: VisibleRouteOptions,
): VisibleIssueRouteContext | null {
const isForegrounded = options?.isForegrounded ?? isPageForegrounded();
if (!isForegrounded) return null;
const relativePath = toCompanyRelativePath(pathname);
const segments = relativePath.split("/").filter(Boolean);
if (!["issues", "chats"].includes(segments[0]) || !segments[1]) return null;
let issueRef = decodeURIComponent(segments[1]);
if (segments[0] === "chats") {
const session = queryClient.getQueryData<Awaited<ReturnType<typeof authApi.getSession>>>(queryKeys.auth.session);
const userId = session?.user?.id ?? session?.session?.userId ?? null;
const companyPrefix = extractCompanyPrefixFromPath(pathname);
const company = queryClient.getQueryData<CompanyListResult>(queryKeys.companies.list(userId))
?.companies.find(item => item.issuePrefix.toUpperCase() === companyPrefix?.toUpperCase());
if (!company) return null;
const agent = queryClient.getQueryData<Agent[]>(queryKeys.agents.list(company.id))
?.find(item => item.id === issueRef || agentRouteRef(item) === issueRef);
if (!agent) return null;
const conversation = queryClient.getQueryData<Issue | null>(queryKeys.agentChats.detail(company.id, userId, agent.id));
if (!conversation) return null;
issueRef = conversation.id;
}
const issue = queryClient.getQueryData<Issue>(queryKeys.issues.detail(issueRef)) ?? null;
const issueRefs = new Set<string>([issueRef]);
if (issue?.id) issueRefs.add(issue.id);
if (issue?.identifier) issueRefs.add(issue.identifier);
function collectCachedIssueRunIds(ref: string, target: Set<string>) {
const activeRun = queryClient.getQueryData<ActiveRunForIssue | null>(
queryKeys.issues.activeRun(ref),
);
const liveRuns =
queryClient.getQueryData<LiveRunForIssue[]>(
queryKeys.issues.liveRuns(ref),
) ?? [];
const linkedRuns =
queryClient.getQueryData<RunForIssue[]>(queryKeys.issues.runs(ref)) ?? [];
if (activeRun?.id) target.add(activeRun.id);
for (const run of liveRuns) {
if (run.id) target.add(run.id);
}
for (const run of linkedRuns) {
if (run.runId) target.add(run.runId);
}
}
// The route may use an identifier while the conversation loads canonical
// UUID-keyed history. A retryable terminal event can omit issueId after its
// first delivery has already removed the run from the company live cache.
const runIds = new Set<string>();
for (const ref of issueRefs) collectCachedIssueRunIds(ref, runIds);
// Notifications about descendants belong to the subtree already on screen.
// Keep these separate from issueRefs, which also drives cache invalidation.
const subtreeIssueRefs = new Set(issueRefs);
const subtreeAgentIds = new Set<string>();
const subtreeRunIds = new Set(runIds);
if (issue?.companyId && issue.id) {
const descendants =
queryClient.getQueryData<Issue[]>(
queryKeys.issues.listByDescendantRoot(issue.companyId, issue.id),
) ?? [];
for (const descendant of descendants) {
subtreeIssueRefs.add(descendant.id);
if (descendant.identifier) subtreeIssueRefs.add(descendant.identifier);
if (descendant.assigneeAgentId)
subtreeAgentIds.add(descendant.assigneeAgentId);
if (descendant.executionRunId)
subtreeRunIds.add(descendant.executionRunId);
// Only current descendants contribute history. Keep their run IDs out
// of runIds, which also drives invalidation of the root conversation.
collectCachedIssueRunIds(descendant.id, subtreeRunIds);
if (descendant.identifier)
collectCachedIssueRunIds(descendant.identifier, subtreeRunIds);
}
}
if (issue?.companyId) {
const companyRuns =
queryClient.getQueryData<LiveRunForIssue[]>(
queryKeys.liveRuns(issue.companyId),
) ?? [];
for (const run of companyRuns) {
if (run.issueId && subtreeIssueRefs.has(run.issueId))
subtreeRunIds.add(run.id);
}
}
return {
routeIssueRef: issueRef,
issueRefs,
assigneeAgentId: issue?.assigneeAgentId ?? null,
runIds,
subtreeIssueRefs,
subtreeAgentIds,
subtreeRunIds,
};
}
function buildIssueRefsForPayload(
entityId: string,
details: Record<string, unknown> | null,
): Set<string> {
const refs = new Set<string>([entityId]);
const identifier =
readString(details?.identifier) ?? readString(details?.issueIdentifier);
if (identifier) refs.add(identifier);
return refs;
}
function overlaps(a: Set<string>, b: Set<string>): boolean {
for (const value of a) {
if (b.has(value)) return true;
}
return false;
}
function shouldSuppressActivityToastForVisibleIssue(
queryClient: QueryClient,
pathname: string,
payload: Record<string, unknown>,
options?: VisibleRouteOptions,
): boolean {
const entityType = readString(payload.entityType);
const entityId = readString(payload.entityId);
if (entityType !== "issue" || !entityId) return false;
const context = resolveVisibleIssueRouteContext(
queryClient,
pathname,
options,
);
if (!context) return false;
return overlaps(
context.subtreeIssueRefs,
buildIssueRefsForPayload(entityId, readRecord(payload.details)),
);
}
function shouldSuppressRunStatusToastForVisibleIssue(
queryClient: QueryClient,
pathname: string,
payload: Record<string, unknown>,
options?: VisibleRouteOptions,
): boolean {
if (!(options?.isForegrounded ?? isPageForegrounded())) return false;
const visibleRunId = toCompanyRelativePath(pathname).match(
/^\/agents\/[^/]+\/runs\/([^/]+)/,
)?.[1];
if (
visibleRunId &&
decodeURIComponent(visibleRunId) === readString(payload.runId)
)
return true;
const context = resolveVisibleIssueRouteContext(
queryClient,
pathname,
options,
);
if (!context) return false;
const issueId = readString(payload.issueId);
if (issueId) return context.subtreeIssueRefs.has(issueId);
const runId = readString(payload.runId);
if (runId) return context.subtreeRunIds.has(runId);
const agentId = readString(payload.agentId);
return (
!!agentId &&
(agentId === context.assigneeAgentId ||
context.subtreeAgentIds.has(agentId))
);
}
function invalidateVisibleIssueRunQueries(
queryClient: QueryClient,
pathname: string,
payload: Record<string, unknown>,
options?: VisibleRouteOptions,
): boolean {
const context = resolveVisibleIssueRouteContext(
queryClient,
pathname,
options,
);
if (!context) return false;
const runId = readString(payload.runId);
const agentId = readString(payload.agentId);
const matchesVisibleIssue =
(runId !== null && context.runIds.has(runId)) ||
(!!agentId &&
!!context.assigneeAgentId &&
agentId === context.assigneeAgentId);
if (!matchesVisibleIssue) return false;
const status = readString(payload.status);
if (runId && status && TERMINAL_RUN_STATUSES.has(status)) {
for (const issueRef of context.issueRefs) {
// Finalization may commit the selected reply without comment_added.
// Fetch the canonical comment before handing off the live transcript;
// progress/queued events must not refetch the whole conversation.
queryClient.invalidateQueries({
queryKey: queryKeys.issues.comments(issueRef),
});
queryClient.setQueryData(
queryKeys.issues.liveRuns(issueRef),
(current: LiveRunForIssue[] | undefined) =>
removeLiveRunById(current, runId),
);
queryClient.setQueryData(
queryKeys.issues.activeRun(issueRef),
(current: ActiveRunForIssue | null | undefined) =>
current?.id === runId ? null : current,
);
queryClient.setQueryData(
queryKeys.issues.detail(issueRef),
(current: Issue | undefined) => clearIssueExecutionRun(current, runId),
);
}
}
for (const issueRef of context.issueRefs) {
queryClient.invalidateQueries({ queryKey: queryKeys.issues.detail(issueRef) });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.activity(issueRef) });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.runs(issueRef) });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.liveRuns(issueRef) });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.activeRun(issueRef) });
if (status && TERMINAL_RUN_STATUSES.has(status)) {
// A final comment can race the last in-flight history fetch. Reconcile
// persisted messages after the turn settles.
queryClient.invalidateQueries({ queryKey: queryKeys.issues.comments(issueRef) });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.attachments(issueRef) });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.workProducts(issueRef) });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.workProductPullRequestRefresh(issueRef) });
queryClient.invalidateQueries({ queryKey: ["issues", "tree-control-state", issueRef] });
}
}
return true;
}
interface RunLiveStatusPatch {
runId: string;
agentId: string | null;
issueId: string | null;
message?: string | null;
updatedAt?: string | null;
currentToolName?: string | null;
lastAssistantSnippet?: string | null;
lastEventAt?: string | null;
}
function hasPatchKey<K extends keyof RunLiveStatusPatch>(
patch: RunLiveStatusPatch,
key: K,
): patch is RunLiveStatusPatch & Required<Pick<RunLiveStatusPatch, K>> {
return Object.prototype.hasOwnProperty.call(patch, key);
}
function applyRunLiveStatusPatch<T extends ActiveRunForIssue | LiveRunForIssue>(
run: T,
patch: RunLiveStatusPatch,
): T {
if (run.id !== patch.runId) return run;
return {
...run,
...(hasPatchKey(patch, "message")
? { currentStatusMessage: patch.message }
: {}),
...(hasPatchKey(patch, "updatedAt")
? { currentStatusUpdatedAt: patch.updatedAt }
: {}),
...(hasPatchKey(patch, "currentToolName")
? { currentToolName: patch.currentToolName }
: {}),
...(hasPatchKey(patch, "lastAssistantSnippet")
? { lastAssistantSnippet: patch.lastAssistantSnippet }
: {}),
...(hasPatchKey(patch, "lastEventAt")
? { lastEventAt: patch.lastEventAt }
: {}),
};
}
function applyRunLiveStatusPatchToArray<T extends LiveRunForIssue>(
runs: T[] | undefined,
patch: RunLiveStatusPatch,
): T[] | undefined {
if (!runs) return runs;
let changed = false;
const nextRuns = runs.map((run) => {
if (run.id !== patch.runId) return run;
changed = true;
return applyRunLiveStatusPatch(run, patch);
});
return changed ? nextRuns : runs;
}
function readRunLiveStatusPatchFromPayload(
payload: Record<string, unknown>,
eventCreatedAt: string,
eventType: string,
): RunLiveStatusPatch | null {
const runId = readString(payload.runId);
if (!runId) return null;
const patch: RunLiveStatusPatch = {
runId,
agentId: readString(payload.agentId),
issueId: readString(payload.issueId),
};
if (eventType === "heartbeat.run.progress") {
patch.message = readString(payload.message);
patch.updatedAt = readString(payload.updatedAt) ?? eventCreatedAt;
patch.currentToolName = readString(payload.currentToolName);
patch.lastAssistantSnippet = readString(payload.lastAssistantSnippet);
patch.lastEventAt = readString(payload.lastEventAt) ?? patch.updatedAt;
return patch;
}
if (eventType === "heartbeat.run.log") {
patch.lastEventAt = readString(payload.ts) ?? eventCreatedAt;
return patch;
}
if (eventType === "heartbeat.run.event") {
const message = readString(payload.message);
if (message) patch.message = message;
patch.updatedAt = readString(payload.updatedAt) ?? eventCreatedAt;
const currentToolName = readString(payload.currentToolName);
if (currentToolName) patch.currentToolName = currentToolName;
const lastAssistantSnippet = readString(payload.lastAssistantSnippet);
if (lastAssistantSnippet) patch.lastAssistantSnippet = lastAssistantSnippet;
patch.lastEventAt = readString(payload.lastEventAt) ?? eventCreatedAt;
return patch;
}
return null;
}
function applyRunLiveStatusPatchToCaches(
queryClient: QueryClient,
companyId: string,
pathname: string,
patch: RunLiveStatusPatch,
options?: VisibleRouteOptions,
): boolean {
let changed = false;
queryClient.setQueryData(
queryKeys.liveRuns(companyId),
(current: LiveRunForIssue[] | undefined) => {
const nextRuns = applyRunLiveStatusPatchToArray(current, patch);
if (nextRuns !== current) changed = true;
return nextRuns;
},
);
const issueRefs = new Set<string>();
if (patch.issueId) {
for (const ref of resolveIssueQueryRefs(
queryClient,
companyId,
patch.issueId,
null,
)) {
issueRefs.add(ref);
}
}
const context = resolveVisibleIssueRouteContext(
queryClient,
pathname,
options,
);
if (
context &&
(context.runIds.has(patch.runId) ||
(!!patch.issueId && context.issueRefs.has(patch.issueId)) ||
(!!patch.agentId &&
!!context.assigneeAgentId &&
patch.agentId === context.assigneeAgentId))
) {
for (const ref of context.issueRefs) issueRefs.add(ref);
}
for (const issueRef of issueRefs) {
queryClient.setQueryData(
queryKeys.issues.activeRun(issueRef),
(current: ActiveRunForIssue | null | undefined) => {
if (!current || current.id !== patch.runId) return current;
changed = true;
return applyRunLiveStatusPatch(current, patch);
},
);
queryClient.setQueryData(
queryKeys.issues.liveRuns(issueRef),
(current: LiveRunForIssue[] | undefined) => {
const nextRuns = applyRunLiveStatusPatchToArray(current, patch);
if (nextRuns !== current) changed = true;
return nextRuns;
},
);
}
return changed;
}
function shouldSuppressAgentStatusToastForVisibleIssue(
queryClient: QueryClient,
pathname: string,
payload: Record<string, unknown>,
options?: VisibleRouteOptions,
): boolean {
const context = resolveVisibleIssueRouteContext(
queryClient,
pathname,
options,
);
if (!context) return false;
const agentId = readString(payload.agentId);
return (
!!agentId &&
(agentId === context.assigneeAgentId ||
context.subtreeAgentIds.has(agentId))
);
}
function shouldDeferIssueRefetchForVisibleAgentActivity(
queryClient: QueryClient,
pathname: string,
payload: Record<string, unknown>,
options?: VisibleRouteOptions,
): boolean {
const entityType = readString(payload.entityType);
const entityId = readString(payload.entityId);
const actorType = readString(payload.actorType);
const action = readString(payload.action);
const details = readRecord(payload.details);
if (entityType !== "issue" || !entityId) return false;
if (actorType !== "agent" && actorType !== "system") return false;
if (action !== "issue.updated") return false;
if (readString(details?.source) === "comment") return false;
const context = resolveVisibleIssueRouteContext(
queryClient,
pathname,
options,
);
if (!context) return false;
return overlaps(
context.issueRefs,
buildIssueRefsForPayload(entityId, details),
);
}
function shouldDeferVisibleIssueCommentActivity(
queryClient: QueryClient,
pathname: string,
payload: Record<string, unknown>,
options?: VisibleRouteOptions,
): boolean {
const entityType = readString(payload.entityType);
const entityId = readString(payload.entityId);
const action = readString(payload.action);
const details = readRecord(payload.details);
if (entityType !== "issue" || !entityId) return false;
if (action !== "issue.comment_added") return false;
// Authenticated events omit comment details. Only defer a refetch when
// the event contains the ID needed to hydrate the new comment separately.
if (!readString(details?.commentId)) return false;
const context = resolveVisibleIssueRouteContext(
queryClient,
pathname,
options,
);
if (!context) return false;
return overlaps(
context.issueRefs,
buildIssueRefsForPayload(entityId, details),
);
}
async function hydrateVisibleIssueComment(
queryClient: QueryClient,
pathname: string,
payload: Record<string, unknown>,
options?: VisibleRouteOptions,
) {
const entityType = readString(payload.entityType);
const action = readString(payload.action);
const details = readRecord(payload.details);
const commentId = readString(details?.commentId);
if (entityType !== "issue" || action !== "issue.comment_added" || !commentId)
return false;
const context = resolveVisibleIssueRouteContext(
queryClient,
pathname,
options,
);
if (!context) return false;
const entityId = readString(payload.entityId);
if (
!entityId ||
!overlaps(context.issueRefs, buildIssueRefsForPayload(entityId, details))
) {
return false;
}
try {
const comment = await issuesApi.getComment(
context.routeIssueRef,
commentId,
);
queryClient.setQueryData<
InfiniteData<IssueComment[], string | null> | undefined
>(queryKeys.issues.comments(context.routeIssueRef), (current) => {
if (!current) {
return {
pages: [[comment]],
pageParams: [null],
};
}
return {
...current,
pages: upsertIssueCommentInPages(current.pages, comment),
};
});
return true;
} catch {
queryClient.invalidateQueries({
queryKey: queryKeys.issues.comments(context.routeIssueRef),
});
return false;
}
}
const ISSUE_TOAST_ACTIONS = new Set([
"issue.created",
"issue.updated",
"issue.comment_added",
]);
const ISSUE_DOCUMENT_ACTIVITY_ACTIONS = new Set([
"issue.document_created",
"issue.document_updated",
"issue.document_restored",
"issue.document_deleted",
]);
const ISSUE_CASE_WORK_ACTIVITY_ACTIONS = new Set([
"issue.updated",
"issue.checked_out",
"issue.released",
]);
const ISSUE_DOCUMENT_ANNOTATION_ACTIVITY_ACTIONS = new Set([
"issue.document_annotation_thread_created",
"issue.document_annotation_comment_added",
"issue.document_annotation_thread_resolved",
"issue.document_annotation_thread_reopened",
"issue.document_annotation_remapped",
]);
const ROUTINE_DOCUMENT_ANNOTATION_ACTIVITY_ACTIONS = new Set([
"routine.document_annotation_thread_created",
"routine.document_annotation_comment_added",
"routine.document_annotation_thread_resolved",
"routine.document_annotation_thread_reopened",
"routine.document_annotation_remapped",
]);
const CASE_DOCUMENT_ANNOTATION_ACTIVITY_ACTIONS = new Set([
"case.document_annotation_thread_created",
"case.document_annotation_comment_added",
"case.document_annotation_thread_resolved",
"case.document_annotation_thread_reopened",
"case.document_annotation_remapped",
]);
const AGENT_TOAST_STATUSES = new Set(["error"]);
const RUN_TOAST_STATUSES = new Set(["failed", "timed_out", "cancelled"]);
function describeIssueUpdate(
details: Record<string, unknown> | null,
): string | null {
if (!details) return null;
const changes: string[] = [];
if (typeof details.status === "string")
changes.push(`status -> ${details.status.replace(/_/g, " ")}`);
if (typeof details.priority === "string")
changes.push(`priority -> ${details.priority}`);
if (
typeof details.assigneeAgentId === "string" ||
typeof details.assigneeUserId === "string"
) {
changes.push("reassigned");
} else if (
details.assigneeAgentId === null ||
details.assigneeUserId === null
) {
changes.push("unassigned");
}
if (details.reopened === true) {
const from = readString(details.reopenedFrom);
changes.push(
from ? `reopened from ${from.replace(/_/g, " ")}` : "reopened",
);
}
if (typeof details.title === "string") changes.push("title changed");
if (typeof details.description === "string")
changes.push("description changed");
if (changes.length > 0) return changes.join(", ");
return null;
}
function buildActivityToast(
queryClient: QueryClient,
companyId: string,
payload: Record<string, unknown>,
currentActor: { userId: string | null; agentId: string | null },
): ToastInput | null {
const entityType = readString(payload.entityType);
const entityId = readString(payload.entityId);
const action = readString(payload.action);
const details = readRecord(payload.details);
const actorId = readString(payload.actorId);
const actorType = readString(payload.actorType);
if (
entityType !== "issue" ||
!entityId ||
!action ||
!ISSUE_TOAST_ACTIONS.has(action)
) {
return null;
}
const issue = resolveIssueToastContext(
queryClient,
companyId,
entityId,
details,
);
const actor = readString(readString(payload.actorName)?.trim()) ?? resolveActorLabel(queryClient, companyId, actorType, actorId);
const cachedAgent = actorType === "agent" && actorId
? queryClient.getQueryData<Agent[]>(queryKeys.agents.list(companyId))?.find((a) => a.id === actorId)
?? queryClient.getQueryData<Agent>(queryKeys.agents.detail(actorId))
: undefined;
const cachedUser = actorType === "user" && actorId
? queryClient.getQueryData<CompanyUserDirectoryResponse>(queryKeys.access.companyUserDirectory(companyId))
?.users.find((u) => u.principalId === actorId)?.user
: undefined;
const toastActor: ToastActor | undefined = actorId && (actorType === "user" || actorType === "agent")
? { type: actorType, id: actorId, name: actor,
image: payload.actorImage !== undefined ? readString(payload.actorImage) : cachedUser?.image,
appearance: actorType === "agent"
? resolveAgentAppearance(payload.actorAppearance !== undefined ? payload.actorAppearance : cachedAgent?.appearance, actorId) : undefined }
: undefined;
const isSelfActivity =
(actorType === "user" &&
!!currentActor.userId &&
actorId === currentActor.userId) ||
(actorType === "agent" &&
!!currentActor.agentId &&
actorId === currentActor.agentId);
if (isSelfActivity) return null;
if (action === "issue.created") {
return {
actor: toastActor,
title: `${actor} created ${issue.ref}`,
body: issue.title ? truncate(issue.title, 96) : undefined,
tone: "success",
action: { label: `View ${issue.ref}`, href: issue.href },
dedupeKey: `activity:${action}:${entityId}`,
};
}
if (action === "issue.updated") {
if (readString(details?.source) === "comment") {
// Comment-driven updates emit a paired comment event; show one combined toast on the comment event.
return null;
}
const changeDesc = describeIssueUpdate(details);
const body = changeDesc
? issue.title
? `${truncate(issue.title, 64)} - ${changeDesc}`
: changeDesc
: issue.title
? truncate(issue.title, 96)
: issue.label;
return {
actor: toastActor,
title: `${actor} updated ${issue.ref}`,
body: truncate(body, 100),
tone: "info",
action: { label: `View ${issue.ref}`, href: issue.href },
dedupeKey: `activity:${action}:${entityId}`,
};
}
const commentId = readString(details?.commentId);
const bodySnippet = readString(details?.bodySnippet);
const reopened = details?.reopened === true;
const updated = details?.updated === true;
const reopenedFrom = readString(details?.reopenedFrom);
const reopenedLabel = reopened
? reopenedFrom
? `reopened from ${reopenedFrom.replace(/_/g, " ")}`
: "reopened"
: null;
const title = reopened
? `${actor} reopened and commented on ${issue.ref}`
: updated
? `${actor} commented and updated ${issue.ref}`
: `${actor} commented on ${issue.ref}`;
const body = bodySnippet
? reopenedLabel
? `${reopenedLabel} - ${bodySnippet.replace(/^#+\s*/m, "").replace(/\n/g, " ")}`
: bodySnippet.replace(/^#+\s*/m, "").replace(/\n/g, " ")
: reopenedLabel
? issue.title
? `${reopenedLabel} - ${issue.title}`
: reopenedLabel
: (issue.title ?? undefined);
return {
actor: toastActor,
title,
body: body ? truncate(body, 96) : undefined,
tone: "info",
action: { label: `View ${issue.ref}`, href: issue.href },
dedupeKey: `activity:${action}:${entityId}:${commentId ?? "na"}`,
};
}
function buildJoinRequestToast(
payload: Record<string, unknown>,
): ToastInput | null {
const entityType = readString(payload.entityType);
const action = readString(payload.action);
const entityId = readString(payload.entityId);
const details = readRecord(payload.details);
if (entityType !== "join_request" || !action || !entityId) return null;
if (action !== "join.requested" && action !== "join.request_replayed")
return null;
const requestType = readString(details?.requestType);
const label = requestType === "agent" ? "Agent" : "Someone";
return {
title: `${label} wants to join`,
body: "A new join request is waiting for approval.",
tone: "info",
action: { label: "View inbox", href: "/inbox/mine" },
dedupeKey: `join-request:${entityId}`,
};
}
function buildAgentStatusToast(
payload: Record<string, unknown>,
nameOf: (id: string) => string | null,
queryClient: QueryClient,
companyId: string,
): ToastInput | null {
const agentId = readString(payload.agentId);
const status = readString(payload.status);
if (!agentId || !status || !AGENT_TOAST_STATUSES.has(status)) return null;
const tone = status === "error" ? "error" : "info";
const name = nameOf(agentId) ?? `Agent ${shortId(agentId)}`;
const title = status === "running" ? `${name} started` : `${name} errored`;
const agents = queryClient.getQueryData<Agent[]>(
queryKeys.agents.list(companyId),
);
const agent = agents?.find((a) => a.id === agentId);
const body = agent?.title ?? undefined;
return {
title,
body,
tone,
action: { label: "View agent", href: `/agents/${agentId}` },
dedupeKey: `agent-status:${agentId}:${status}`,
};
}
function buildRunStatusToast(
payload: Record<string, unknown>,
nameOf: (id: string) => string | null,
): ToastInput | null {
const runId = readString(payload.runId);
const agentId = readString(payload.agentId);
const status = readString(payload.status);
if (!runId || !agentId || !status || !RUN_TOAST_STATUSES.has(status))
return null;
const error = readString(payload.error);
const errorCode = readString(payload.errorCode);
// Interrupt is an intentional conversation control. Its caller gives
// feedback; the terminal event must not announce a cancelled/failed run.
if (errorCode === "operator_interrupted") return null;
// Workspace contention is ordinary scheduling, not a failed user action.
if (errorCode === "workspace_busy") return null;
const contextSource = readString(payload.contextSource);
const triggerDetail = readString(payload.triggerDetail);
const name = nameOf(agentId) ?? "Agent";
if (
status === "failed" &&
errorCode === "low_trust_isolation_unavailable" &&
contextSource?.startsWith("chat:")
) {
return {
title: `${name} couldn't start this chat`,
body: "This external chat identity isn't linked, and isolated guest workspaces are disabled. Link the identity in Connectors or enable isolated workspaces, then start a new task.",
tone: "warn",
ttlMs: 10_000,
action: { label: "Open chat connections", href: "/apps" },
dedupeKey: `run-status:${runId}:${status}`,
};
}
const tone =
status === "succeeded"
? "success"
: status === "cancelled"
? "info"
: "error";
const statusLabel =
status === "succeeded"
? "succeeded"
: status === "failed"
? "failed"
: status === "timed_out"
? "timed out"
: "cancelled";
const title = `${name} run ${statusLabel}`;
let body: string | undefined;
if (error) {
body = truncate(error, 100);
} else if (triggerDetail) {
body = `Trigger: ${triggerDetail}`;
}
return {
title,
body,
tone,
ttlMs: status === "succeeded" ? 5000 : 7000,
action: { label: "View run", href: `/agents/${agentId}/runs/${runId}` },
dedupeKey: `run-status:${runId}:${status}`,
};
}
/**
* Event-source the company live-runs list from run-lifecycle events instead of
* invalidating + refetching it. Returns true when the cache was fully patched;
* false means a genuinely new run appeared that can't be reconstructed from the
* event, so the caller should refetch once to pick it up.
*/
function applyRunLifecycleToCompanyLiveRuns(
queryClient: QueryClient,
companyId: string,
payload: Record<string, unknown>,
): boolean {
const runId = readString(payload.runId);
const status = readString(payload.status);
if (!runId || !status) return false;
queryClient.setQueryData(
queryKeys.runDetail(runId),
(current: HeartbeatRun | undefined) => {
if (!current) return current;
const has = (key: string) =>
Object.prototype.hasOwnProperty.call(payload, key);
return {
...current,
status: status as HeartbeatRun["status"],
...(has("invocationSource")
? {
invocationSource:
readString(payload.invocationSource) ??
current.invocationSource,
}
: {}),
...(has("triggerDetail")
? { triggerDetail: readString(payload.triggerDetail) }
: {}),
...(has("error") ? { error: readString(payload.error) } : {}),
...(has("errorCode")
? { errorCode: readString(payload.errorCode) }
: {}),
...(has("startedAt")
? { startedAt: readString(payload.startedAt) }
: {}),
...(has("finishedAt")
? { finishedAt: readString(payload.finishedAt) }
: {}),
};
},
);
if (TERMINAL_RUN_STATUSES.has(status)) {
queryClient.setQueryData(
queryKeys.liveRuns(companyId),
(current: LiveRunForIssue[] | undefined) =>
removeRunFromList(current, runId),
);
// Always "handled": a terminal run must never be in the live list, so if it
// wasn't present there is deliberately nothing to refetch (removeRunFromList
// was a no-op and we must not re-add it).
return true;
}
let present = false;
queryClient.setQueryData(
queryKeys.liveRuns(companyId),
(current: LiveRunForIssue[] | undefined) => {
const result = patchRunStatusInList(current, runId, status);
present = result.present;
return result.next;
},
);
return present;
}
function invalidateHeartbeatQueries(
queryClient: ReturnType<typeof useQueryClient>,
companyId: string,
payload: Record<string, unknown>,
) {
queryClient.invalidateQueries({ queryKey: queryKeys.issues.summaries(companyId) });
// Note: liveRuns(companyId) is intentionally NOT invalidated here — it is
// event-sourced via applyRunLifecycleToCompanyLiveRuns in the caller.
queryClient.invalidateQueries({ queryKey: queryKeys.heartbeats(companyId) });
queryClient.invalidateQueries({ queryKey: queryKeys.agents.list(companyId) });
queryClient.invalidateQueries({ queryKey: queryKeys.dashboard(companyId) });
queryClient.invalidateQueries({ queryKey: ["costs", companyId] });
queryClient.invalidateQueries({
queryKey: queryKeys.sidebarBadges(companyId),
});
const agentId = readString(payload.agentId);
if (agentId) {
queryClient.invalidateQueries({
queryKey: queryKeys.agents.detail(agentId),
});
queryClient.invalidateQueries({
queryKey: queryKeys.heartbeats(companyId, agentId),
});
}
const runId = readString(payload.runId);
if (runId) {
queryClient.invalidateQueries({ queryKey: queryKeys.runDetail(runId) });
}
}
function invalidateHeartbeatProgressQueries(
queryClient: ReturnType<typeof useQueryClient>,
_companyId: string,
payload: Record<string, unknown>,
) {
const agentId = readString(payload.agentId);
if (agentId) {
queryClient.invalidateQueries({
queryKey: queryKeys.agents.detail(agentId),
});
}
}
function invalidateActivityQueries(
queryClient: ReturnType<typeof useQueryClient>,
companyId: string,
payload: Record<string, unknown>,
currentActor: { userId: string | null; agentId: string | null },
options?: { pathname?: string; isForegrounded?: boolean },
) {
queryClient.invalidateQueries({ queryKey: queryKeys.activity(companyId) });
queryClient.invalidateQueries({ queryKey: queryKeys.dashboard(companyId) });
queryClient.invalidateQueries({
queryKey: queryKeys.sidebarBadges(companyId),
});
const entityType = readString(payload.entityType);
const entityId = readString(payload.entityId);
const action = readString(payload.action);
const actorType = readString(payload.actorType);
const actorId = readString(payload.actorId);
const details = readRecord(payload.details);
const ownActorActivity =
(actorType === "user" &&
!!currentActor.userId &&
actorId === currentActor.userId) ||
(actorType === "agent" &&
!!currentActor.agentId &&
actorId === currentActor.agentId);
if (action?.startsWith("ai_connection.") || action?.startsWith("connection_grant.")) {
queryClient.invalidateQueries({ queryKey: ["ai-connections", companyId] });
}
if (action?.startsWith("primary_agent.") || action?.startsWith("agent.") || action?.startsWith("resource_membership.")) {
queryClient.invalidateQueries({ queryKey: ["primary-agent", companyId] });
}
if (action?.startsWith("resource_membership.") || action?.startsWith("primary_agent.")) {
const targetUserId = readString(details?.userId);
if (!targetUserId || targetUserId === currentActor.userId) {
queryClient.invalidateQueries({
queryKey: queryKeys.resourceMemberships.mine(companyId),
});
}
}
if (entityType === "issue") {
const chatListKey = queryKeys.agentChats.list(companyId, currentActor.userId);
const knownChat = entityId && queryClient.getQueryData<Issue[]>(chatListKey)?.some(chat => chat.id === entityId);
if (knownChat || action === "issue.conversation_opened" && ownActorActivity) {
queryClient.invalidateQueries({ queryKey: chatListKey });
}
if (action === "issue.tree_hold_created" || action === "issue.tree_hold_released" || action === "issue.updated") {
// An ancestor hold or reparenting changes descendants' effective pause.
queryClient.invalidateQueries({ queryKey: ["issues", "tree-control-state"] });
}
queryClient.invalidateQueries({
queryKey: queryKeys.issues.list(companyId),
});
queryClient.invalidateQueries({
queryKey: queryKeys.issues.listMineByMe(companyId),
});
queryClient.invalidateQueries({
queryKey: queryKeys.issues.listTouchedByMe(companyId),
});
queryClient.invalidateQueries({
queryKey: queryKeys.issues.listUnreadTouchedByMe(companyId),
});
if (entityId) {
const selfCommentActivity =
(action === "issue.comment_added" ||
(action === "issue.updated" &&
readString(details?.source) === "comment")) &&
((actorType === "user" &&
!!currentActor.userId &&
actorId === currentActor.userId) ||
(actorType === "agent" &&
!!currentActor.agentId &&
actorId === currentActor.agentId));
const visibleIssueAgentActivity =
!!options?.pathname &&
shouldDeferIssueRefetchForVisibleAgentActivity(
queryClient,
options.pathname,
payload,
{ isForegrounded: options.isForegrounded },
);
const visibleIssueCommentActivity =
!!options?.pathname &&
shouldDeferVisibleIssueCommentActivity(
queryClient,
options.pathname,
payload,
{ isForegrounded: options.isForegrounded },
);
const issueRefs = resolveIssueQueryRefs(
queryClient,
companyId,
entityId,
details,
);
const taskDocumentsChanged = !!action && ISSUE_DOCUMENT_ACTIVITY_ACTIONS.has(action);
const caseWorkChanged = !!action && ISSUE_CASE_WORK_ACTIVITY_ACTIONS.has(action);
if (caseWorkChanged) {
// A changed task can bring a case into a filtered list that did not
// contain it, even when no case detail has been loaded yet.
queryClient.invalidateQueries({ queryKey: queryKeys.cases.list(companyId) });
}
if (taskDocumentsChanged || caseWorkChanged) {
const changedTaskRefs = new Set(issueRefs);
// Both detail aliases embed primary-task state. Linked tasks affect the
// document rollup, including creation of their first document.
queryClient.invalidateQueries({
queryKey: ["cases", "detail"],
predicate: (query) => {
const cached = query.state.data as CaseDetail | undefined;
return cached?.companyId === companyId && (
changedTaskRefs.has(cached.workIssueId ?? "") ||
(taskDocumentsChanged && cached.issueLinks?.some((link) => changedTaskRefs.has(link.issueId)) === true)
);
},
});
}
const visibleContext = options?.pathname
? resolveVisibleIssueRouteContext(queryClient, options.pathname, {
isForegrounded: options.isForegrounded,
})
: null;
if (
visibleContext &&
overlaps(visibleContext.issueRefs, buildIssueRefsForPayload(entityId, details)) &&
!issueRefs.includes(visibleContext.routeIssueRef)
) {
// Authenticated payloads omit the identifier. Include the visible
// route's cache key when its loaded task matches the event's UUID.
issueRefs.push(visibleContext.routeIssueRef);
}
for (const ref of issueRefs) {
const invalidationOptions =
selfCommentActivity ||
visibleIssueAgentActivity ||
visibleIssueCommentActivity
? { refetchType: "inactive" as const }
: undefined;
queryClient.invalidateQueries({ queryKey: queryKeys.issues.detail(ref), ...invalidationOptions });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.activity(ref), ...invalidationOptions });
if (action === "issue.comment_added" || action === "issue.conversation_session_started") {
queryClient.invalidateQueries({ queryKey: queryKeys.issues.comments(ref), ...invalidationOptions });
}
if (action?.startsWith("issue.attachment_") || action?.startsWith("issue.work_product_")) {
// These cards are durable API objects, not streamed text. Refresh the
// visible task too, including attachments bound to an existing comment.
queryClient.invalidateQueries({ queryKey: queryKeys.issues.attachments(ref) });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.workProducts(ref) });
}
if (action === "issue.conversation_session_started") {
queryClient.invalidateQueries({ queryKey: ["issues", "tree-control-state", ref] });
queryClient.invalidateQueries({ queryKey: queryKeys.issues.interactions(ref) });
}
if (action && ISSUE_DOCUMENT_ACTIVITY_ACTIONS.has(action)) {
const documentKey = readString(details?.key);
queryClient.invalidateQueries({
queryKey: queryKeys.issues.documents(ref),
...invalidationOptions,
});
if (documentKey) {
queryClient.invalidateQueries({
queryKey: queryKeys.issues.document(ref, documentKey),
...invalidationOptions,
});
queryClient.invalidateQueries({
queryKey: queryKeys.issues.documentRevisions(ref, documentKey),
...invalidationOptions,
});
} else {
queryClient.invalidateQueries({
queryKey: ["issues", "document", ref],
...invalidationOptions,
});
queryClient.invalidateQueries({
queryKey: ["issues", "document-revisions", ref],
...invalidationOptions,
});
}
}
if (
action &&
(ISSUE_DOCUMENT_ACTIVITY_ACTIONS.has(action) ||
ISSUE_DOCUMENT_ANNOTATION_ACTIVITY_ACTIONS.has(action))
) {
const documentKey =
readString(details?.key) ?? readString(details?.documentKey);
queryClient.invalidateQueries({
queryKey: documentKey
? ["issues", "document-annotations", ref, documentKey]
: ["issues", "document-annotations", ref],
...invalidationOptions,
});
}
if (
action?.startsWith("issue.thread_interaction_") ||
((action === "issue.updated" ||
action === "issue.status_decision_recorded") &&
readString(details?.source) === "native_status_decision")
) {
// Native status decisions may materialize a review interaction in the
// same transaction. Unlike run/activity rows, that card is not present
// in the streamed PRP projection, so inactive-only invalidation leaves
// the visible task stale until a full reload.
queryClient.invalidateQueries({
queryKey: queryKeys.issues.interactions(ref),
});
queryClient.invalidateQueries({
queryKey: queryKeys.tools.actionRequests(companyId, "pending"),
});
queryClient.invalidateQueries({
queryKey: queryKeys.apps.attention(companyId),
});
queryClient.invalidateQueries({
queryKey: queryKeys.tools.trustRules(companyId),
});
}
}
}
return;
}
if (entityType === "agent") {
queryClient.invalidateQueries({
queryKey: queryKeys.agents.list(companyId),
});
queryClient.invalidateQueries({ queryKey: queryKeys.org(companyId) });
if (entityId) {
queryClient.invalidateQueries({
queryKey: queryKeys.agents.detail(entityId),
});
queryClient.invalidateQueries({
queryKey: queryKeys.heartbeats(companyId, entityId),
});
}
return;
}
if (entityType === "project") {
const sourceIssueId = readString((payload.details as Record<string, unknown> | undefined)?.sourceIssueId);
if (sourceIssueId) queryClient.invalidateQueries({ queryKey: queryKeys.issues.activity(sourceIssueId) });
queryClient.invalidateQueries({ queryKey: queryKeys.projects.all(companyId) });
if (entityId) queryClient.invalidateQueries({ queryKey: queryKeys.projects.detail(entityId) });
return;
}
if (entityType === "company_skill_source") {
queryClient.invalidateQueries({ queryKey: queryKeys.skillSources.all(companyId) });
queryClient.invalidateQueries({ queryKey: queryKeys.companySkills.list(companyId) });
queryClient.invalidateQueries({ queryKey: queryKeys.skillSources.projects(companyId) });
return;
}
if (entityType === "company_skill") {
const sourceIssueId = readString((payload.details as Record<string, unknown> | undefined)?.sourceIssueId);
if (sourceIssueId) queryClient.invalidateQueries({ queryKey: queryKeys.issues.activity(sourceIssueId) });
if (entityId) queryClient.invalidateQueries({ queryKey: queryKeys.companySkills.detail(companyId, entityId) });
queryClient.invalidateQueries({ queryKey: queryKeys.companySkills.list(companyId) });
return;
}
if (entityType === "goal") {
queryClient.invalidateQueries({
queryKey: queryKeys.goals.list(companyId),
});
if (entityId)
queryClient.invalidateQueries({
queryKey: queryKeys.goals.detail(entityId),
});
return;
}
if (entityType === "approval") {
queryClient.invalidateQueries({
queryKey: queryKeys.approvals.list(companyId),
});
return;
}
if (entityType === "join_request") {
queryClient.invalidateQueries({
queryKey: queryKeys.access.joinRequests(companyId),
});
return;
}
if (entityType === "cost_event") {
queryClient.invalidateQueries({ queryKey: ["costs", companyId] });
queryClient.invalidateQueries({
queryKey: ["usage-by-provider", companyId],
});
queryClient.invalidateQueries({ queryKey: ["usage-by-biller", companyId] });
queryClient.invalidateQueries({
queryKey: queryKeys.usageWindowSpend(companyId),
});
// usageQuotaWindows is intentionally excluded: quota windows come from external provider
// apis on a 5-minute poll and do not change in response to cost events logged by agents
return;
}
if (
entityType === "routine" ||
entityType === "routine_trigger" ||
entityType === "routine_run"
) {
queryClient.invalidateQueries({ queryKey: ["routines"] });
if (
entityType === "routine" &&
action &&
ROUTINE_DOCUMENT_ANNOTATION_ACTIVITY_ACTIONS.has(action) &&
entityId
) {
const documentKey =
readString(details?.key) ??
readString(details?.documentKey) ??
"description";
const routineInvalidationOptions = ownActorActivity
? { refetchType: "inactive" as const }
: undefined;
queryClient.invalidateQueries({
queryKey: ["routines", "document-annotations", entityId, documentKey],
...routineInvalidationOptions,
});
}
return;
}
if (entityType === "case") {
queryClient.invalidateQueries({
queryKey: queryKeys.cases.list(companyId),
});
if (entityId) {
const caseRefs = new Set([entityId]);
const issueRefs = new Set<string>();
const handoffIssueId = action?.startsWith("case.handoff_")
? readString(payload.issueId) ?? readString(details?.issueId)
: null;
if (handoffIssueId) issueRefs.add(handoffIssueId);
// Detail routes normally use the identifier, while activity carries the UUID.
for (const [key, cached] of queryClient.getQueriesData<{ id: string; identifier?: string; workIssueId?: string | null }>({
queryKey: ["cases", "detail"],
})) {
if (cached?.id !== entityId) continue;
queryClient.invalidateQueries({ queryKey: key });
if (cached.identifier) caseRefs.add(cached.identifier);
if (cached.workIssueId) issueRefs.add(cached.workIssueId);
}
// A task's cases panel can be mounted without ever loading case detail.
for (const [key, links] of queryClient.getQueriesData<IssueCaseLink[]>({
queryKey: ["cases", "for-issue"],
})) {
const link = links?.find((candidate) => candidate.case.id === entityId);
if (!link) continue;
queryClient.invalidateQueries({ queryKey: key });
caseRefs.add(link.case.identifier);
}
const documentChanged = action !== null && [
"case.document_revised", "case.document_created", "case.document_updated",
"case.document_restored", "case.document_deleted",
].includes(action);
const documentKey = readString(details?.key) ?? readString(details?.documentKey);
for (const ref of caseRefs) {
queryClient.invalidateQueries({ queryKey: queryKeys.cases.detail(ref) });
queryClient.invalidateQueries({ queryKey: queryKeys.cases.events(ref) });
queryClient.invalidateQueries({ queryKey: queryKeys.cases.handoffs(ref) });
if (documentChanged) {
queryClient.invalidateQueries({ queryKey: queryKeys.cases.documents(ref) });
queryClient.invalidateQueries({
queryKey: documentKey ? ["cases", "document", ref, documentKey] : ["cases", "document", ref],
});
queryClient.invalidateQueries({
queryKey: documentKey ? queryKeys.cases.revisions(ref, documentKey) : ["cases", "revisions", ref],
});
}
}
for (const ref of issueRefs) {
queryClient.invalidateQueries({ queryKey: queryKeys.cases.forIssue(ref) });
}
if (action && CASE_DOCUMENT_ANNOTATION_ACTIVITY_ACTIONS.has(action)) {
const documentKey =
readString(details?.key) ?? readString(details?.documentKey);
const caseInvalidationOptions = ownActorActivity
? { refetchType: "inactive" as const }
: undefined;
queryClient.invalidateQueries({
queryKey: documentKey
? ["cases", "document-annotations", entityId, documentKey]
: ["cases", "document-annotations", entityId],
...caseInvalidationOptions,
});
}
}
return;
}
if (entityType === "summary_slot") {
// The Summarizer's authoritative PUT logs `summary_slot.write`; refresh the
// affected slot + its revisions so the finished summary lands instantly
// instead of waiting for the card's 3s generating poll tick.
const scopeKind = readString(details?.scopeKind);
const slotKey = readString(details?.slotKey);
const scopeId = readString(details?.scopeId);
if (scopeKind && slotKey) {
queryClient.invalidateQueries({
queryKey: queryKeys.summarySlots.detail(
companyId,
scopeKind,
slotKey,
scopeId,
),
});
queryClient.invalidateQueries({
queryKey: queryKeys.summarySlots.revisions(
companyId,
scopeKind,
slotKey,
scopeId,
),
});
}
return;
}
if (entityType === "company") {
queryClient.invalidateQueries({ queryKey: queryKeys.companies.all });
}
}
interface ToastGate {
cooldownHits: Map<string, number[]>;
suppressUntil: number;
observedRunOutcomes: Set<string>;
}
function observeRunOutcome(gate: ToastGate, event: LiveEvent): boolean {
const payload = event.payload ?? {};
const runId = readString(payload.runId);
const status = readString(payload.status);
if (!runId || !status || !TERMINAL_RUN_STATUSES.has(status)) return false;
const key = `${event.companyId}:${runId}:${status}`;
const observed = gate.observedRunOutcomes.has(key);
// Refresh insertion order so repeated deliveries stay remembered even in a
// busy company. Remember suppressed outcomes too (visible task/reconnect).
gate.observedRunOutcomes.delete(key);
gate.observedRunOutcomes.add(key);
if (gate.observedRunOutcomes.size > MAX_OBSERVED_RUN_OUTCOMES) {
gate.observedRunOutcomes.delete(gate.observedRunOutcomes.values().next().value!);
}
// Use the two server timestamps: delivery time is not failure time. This
// also keeps historical failures quiet after a reload, without client clock
// skew hiding fresh failures. Missing timestamps retain legacy behavior.
const finishedAt = Date.parse(readString(payload.finishedAt) ?? "");
const deliveredAt = Date.parse(event.createdAt);
return observed || deliveredAt - finishedAt > RUN_TOAST_MAX_AGE_MS;
}
function shouldSuppressToast(gate: ToastGate, category: string): boolean {
const now = Date.now();
if (now < gate.suppressUntil) return true;
const hits = gate.cooldownHits.get(category);
if (!hits) return false;
const recent = hits.filter((t) => now - t < TOAST_COOLDOWN_WINDOW_MS);
gate.cooldownHits.set(category, recent);
return recent.length >= TOAST_COOLDOWN_MAX;
}
function recordToastHit(gate: ToastGate, category: string) {
const now = Date.now();
const hits = gate.cooldownHits.get(category) ?? [];
hits.push(now);
gate.cooldownHits.set(category, hits);
}
function gatedPushToast(
gate: ToastGate,
pushToast: (toast: ToastInput) => string | null,
category: string,
toast: ToastInput,
) {
if (shouldSuppressToast(gate, category)) return;
const id = pushToast(toast);
if (id !== null) recordToastHit(gate, category);
}
function handleLiveEvent(
queryClient: QueryClient,
expectedCompanyId: string,
pathname: string,
event: LiveEvent,
pushToast: (toast: ToastInput) => string | null,
gate: ToastGate,
currentActor: { userId: string | null; agentId: string | null },
) {
if (event.companyId !== expectedCompanyId) return;
const nameOf = (id: string) =>
resolveAgentName(queryClient, expectedCompanyId, id);
const payload = event.payload ?? {};
// Resolve membership before terminal lifecycle patches remove live-run rows.
const suppressRunToast =
event.type === "heartbeat.run.status" &&
(observeRunOutcome(gate, event) ||
shouldSuppressRunStatusToastForVisibleIssue(queryClient, pathname, payload));
const liveStatusPatch = readRunLiveStatusPatchFromPayload(
payload,
event.createdAt,
event.type,
);
if (liveStatusPatch) {
applyRunLiveStatusPatchToCaches(
queryClient,
expectedCompanyId,
pathname,
liveStatusPatch,
);
}
if (event.type === "heartbeat.run.log") {
return;
}
if (
event.type === "heartbeat.run.queued" ||
event.type === "heartbeat.run.status"
) {
const liveRunsPatched = applyRunLifecycleToCompanyLiveRuns(
queryClient,
expectedCompanyId,
payload,
);
invalidateHeartbeatQueries(queryClient, expectedCompanyId, payload);
if (!liveRunsPatched) {
// A new run we couldn't reconstruct from the event — refetch once to add it.
queryClient.invalidateQueries({
queryKey: queryKeys.liveRuns(expectedCompanyId),
});
}
invalidateVisibleIssueRunQueries(queryClient, pathname, payload);
if (event.type === "heartbeat.run.status") {
const toast = buildRunStatusToast(payload, nameOf);
if (toast && event.toastEligible === true && !suppressRunToast) {
gatedPushToast(gate, pushToast, "run-status", toast);
}
}
return;
}
if (event.type === "heartbeat.run.progress") {
invalidateHeartbeatProgressQueries(queryClient, expectedCompanyId, payload);
invalidateVisibleIssueRunQueries(queryClient, pathname, payload);
return;
}
if (event.type === "heartbeat.run.event") {
return;
}
if (event.type === "agent.status") {
queryClient.invalidateQueries({
queryKey: queryKeys.agents.list(expectedCompanyId),
});
queryClient.invalidateQueries({
queryKey: queryKeys.dashboard(expectedCompanyId),
});
queryClient.invalidateQueries({
queryKey: queryKeys.org(expectedCompanyId),
});
const agentId = readString(payload.agentId);
if (agentId)
queryClient.invalidateQueries({
queryKey: queryKeys.agents.detail(agentId),
});
// Agent-wide errors have no task participation context. Task-linked
// failures are announced through the personally scoped run-status event.
return;
}
if (event.type === "activity.logged") {
invalidateActivityQueries(
queryClient,
expectedCompanyId,
payload,
currentActor,
{ pathname },
);
if (
shouldDeferVisibleIssueCommentActivity(queryClient, pathname, payload)
) {
void hydrateVisibleIssueComment(queryClient, pathname, payload);
}
const action = readString(payload.action);
const toast =
(event.toastEligible === true ? buildActivityToast(
queryClient,
expectedCompanyId,
payload,
currentActor,
) : null) ?? buildJoinRequestToast(payload);
if (
toast &&
!shouldSuppressActivityToastForVisibleIssue(
queryClient,
pathname,
payload,
)
) {
gatedPushToast(gate, pushToast, `activity:${action ?? "unknown"}`, toast);
}
}
}
function resolveLiveCompanyId(
selectedCompanyId: string | null,
selectedCompanyLiveId: string | null,
): string | null {
return selectedCompanyId && selectedCompanyId === selectedCompanyLiveId
? selectedCompanyId
: null;
}
function resetSocketHandlers(target: LiveUpdatesSocketLike) {
target.onopen = null;
target.onmessage = null;
target.onerror = null;
target.onclose = null;
}
function closeSocketQuietly(
target: LiveUpdatesSocketLike | null,
reason: string,
) {
if (!target) return;
if (target.readyState === SOCKET_CONNECTING) {
// Let the handshake complete and then close. Calling close() while the
// socket is still CONNECTING is what triggers the noisy browser error.
target.onopen = () => {
resetSocketHandlers(target);
target.close(1000, reason);
};
target.onmessage = null;
target.onerror = () => undefined;
target.onclose = null;
return;
}
resetSocketHandlers(target);
if (target.readyState === SOCKET_OPEN) {
target.close(1000, reason);
}
}
export const __liveUpdatesTestUtils = {
buildActivityToast,
applyRunLifecycleToCompanyLiveRuns,
buildAgentStatusToast,
buildRunStatusToast,
closeSocketQuietly,
dispatchLiveEventToSubscribers,
LiveEventSubscriptionContext,
applyRunLiveStatusPatchToCaches,
hydrateVisibleIssueComment,
invalidateActivityQueries,
invalidateHeartbeatQueries,
invalidateHeartbeatProgressQueries,
invalidateVisibleIssueRunQueries,
readRunLiveStatusPatchFromPayload,
resolveLiveCompanyId,
canUseLiveSession,
shouldDeferIssueRefetchForVisibleAgentActivity,
shouldDeferVisibleIssueCommentActivity,
shouldSuppressActivityToastForVisibleIssue,
shouldSuppressRunStatusToastForVisibleIssue,
shouldSuppressAgentStatusToastForVisibleIssue,
};
function canUseLiveSession(sessionStatus: string, hasSession: boolean, deploymentMode?: string) {
return sessionStatus === "success" && (hasSession || deploymentMode === "local_trusted");
}
export function LiveUpdatesProvider({ children }: { children: ReactNode }) {
const { visible } = usePageVisibility();
const wasHidden = useRef(!visible);
const hasOpenedSocket = useRef(false);
const hasNeededSocketRecovery = useRef(!visible);
const { selectedCompanyId, selectedCompany } = useCompany();
const queryClient = useQueryClient();
const { pushToast } = useToastActions();
const location = useLocation();
const gateRef = useRef<ToastGate>({
cooldownHits: new Map(),
suppressUntil: 0,
observedRunOutcomes: new Set(),
});
const pathnameRef = useRef(location.pathname);
const { data: session, status: sessionStatus } = useQuery({
queryKey: queryKeys.auth.session,
queryFn: () => authApi.getSession(),
retry: false,
});
const { data: health } = useQuery({ queryKey: queryKeys.health, queryFn: healthApi.get });
const currentUserId = session?.user?.id ?? session?.session?.userId ??
(health?.deploymentMode === "local_trusted" ? "local-board" : null);
const socketAuthKey = session?.session?.id ?? currentUserId ?? "signed_out";
const liveCompanyId = resolveLiveCompanyId(selectedCompanyId, selectedCompany?.id ?? null);
const canConnectSocket = canUseLiveSession(sessionStatus, session != null, health?.deploymentMode) && liveCompanyId !== null;
const currentActorRef = useRef<{ userId: string | null; agentId: string | null }>({
userId: currentUserId,
agentId: null,
});
const subscribersRef = useRef<Set<CompanyLiveEventHandler>>(new Set());
const subscribe = useCallback((handler: CompanyLiveEventHandler) => {
subscribersRef.current.add(handler);
return () => {
subscribersRef.current.delete(handler);
};
}, []);
const subscriptionValue = useMemo<LiveEventSubscription>(
() => ({ subscribe }),
[subscribe],
);
// Coalesce the per-event invalidation storm. Optimistic setQueryData writes
// still pass straight through (immediate); only invalidateQueries is batched
// and flushed at most a few times per second.
const invalidationBatcher = useMemo(
() => createInvalidationBatcher(queryClient),
[queryClient],
);
const coalescingClient = useMemo(
() => createCoalescingQueryClient(queryClient, invalidationBatcher),
[queryClient, invalidationBatcher],
);
useEffect(() => () => invalidationBatcher.dispose(), [invalidationBatcher]);
useEffect(() => {
pathnameRef.current = location.pathname;
}, [location.pathname]);
useEffect(() => {
currentActorRef.current = {
userId: currentUserId,
agentId: null,
};
}, [currentUserId]);
useEffect(() => {
if (!visible) {
wasHidden.current = true;
hasNeededSocketRecovery.current = true;
invalidationBatcher.dispose();
return;
}
if (!canConnectSocket || !liveCompanyId) return;
if (wasHidden.current) {
wasHidden.current = false;
// Reconcile events missed while hidden, including completed runs/issues.
void queryClient.invalidateQueries({ type: "active" }, { cancelRefetch: false });
}
let closed = false;
let reconnectAttempt = 0;
let reconnectTimer: number | null = null;
let pollTimer: number | null = null;
let socket: WebSocket | null = null;
const clearReconnect = () => {
if (reconnectTimer !== null) {
window.clearTimeout(reconnectTimer);
reconnectTimer = null;
}
};
const stopPolling = () => {
if (pollTimer !== null) {
window.clearInterval(pollTimer);
pollTimer = null;
}
};
const startPolling = () => {
if (closed || pollTimer !== null) return;
// Visible queries still need fresh state when realtime is unavailable.
pollTimer = window.setInterval(() => {
void queryClient.invalidateQueries({ type: "active" }, { cancelRefetch: false });
}, DISCONNECTED_POLL_INTERVAL_MS);
};
const scheduleReconnect = () => {
if (closed || reconnectTimer !== null) return;
reconnectAttempt += 1;
hasNeededSocketRecovery.current = true;
const delayMs = Math.min(
15000,
1000 * 2 ** Math.min(reconnectAttempt - 1, 4),
);
reconnectTimer = window.setTimeout(() => {
reconnectTimer = null;
connect();
}, delayMs);
};
const connect = () => {
if (closed) return;
const url = buildSameOriginWebSocketUrl(
`/api/companies/${encodeURIComponent(liveCompanyId)}/events/ws`,
);
const nextSocket = tryCreateWebSocket(url);
if (!nextSocket) {
startPolling();
scheduleReconnect();
return;
}
socket = nextSocket;
nextSocket.onopen = () => {
if (closed || socket !== nextSocket) {
closeSocketQuietly(nextSocket, "stale_connection");
return;
}
stopPolling();
if (reconnectAttempt > 0) {
gateRef.current.suppressUntil = Date.now() + RECONNECT_SUPPRESS_MS;
}
// The initial page queries can finish before the first subscription,
// too. Reconcile that gap as well as reconnects: events are not replayed.
const initialHealthyOpen = !hasOpenedSocket.current && !hasNeededSocketRecovery.current;
hasOpenedSocket.current = true;
void queryClient.invalidateQueries({
type: "active",
...(initialHealthyOpen ? { predicate: (query) => {
const key = JSON.stringify(query.queryKey);
return key !== JSON.stringify(queryKeys.auth.session) && key !== JSON.stringify(queryKeys.health);
} } : {}),
}, { cancelRefetch: false });
reconnectAttempt = 0;
};
nextSocket.onmessage = (message) => {
if (closed || socket !== nextSocket) return;
if (!getPageVisibility().visible) return;
const raw = typeof message.data === "string" ? message.data : "";
if (!raw) return;
try {
const parsed = JSON.parse(raw) as LiveEvent;
handleLiveEvent(
coalescingClient,
liveCompanyId,
pathnameRef.current,
parsed,
pushToast,
gateRef.current,
{
userId: currentActorRef.current.userId,
agentId: currentActorRef.current.agentId,
},
);
// Fan the raw event out to component subscribers after cache
// handling so any reader sees fresh query data.
dispatchLiveEventToSubscribers(
subscribersRef.current,
liveCompanyId,
parsed,
);
} catch {
// Ignore non-JSON payloads.
}
};
nextSocket.onerror = () => {
// Wait for onclose to drive the reconnect. Self-closing here is what
// produces the "closed before connection established" browser noise.
};
nextSocket.onclose = () => {
if (socket !== nextSocket) return;
socket = null;
if (closed) return;
startPolling();
scheduleReconnect();
};
};
// Delay initial connect slightly so React StrictMode's double-invoke
// cleanup fires before the WebSocket is created, avoiding the
// "WebSocket closed before connection established" dev-mode error.
const connectTimer = window.setTimeout(connect, 0);
return () => {
closed = true;
window.clearTimeout(connectTimer);
clearReconnect();
stopPolling();
const activeSocket = socket;
socket = null;
closeSocketQuietly(activeSocket, "provider_unmount");
};
}, [
visible,
invalidationBatcher,
queryClient,
coalescingClient,
liveCompanyId,
pushToast,
canConnectSocket,
socketAuthKey,
]);
return (
<LiveEventSubscriptionContext.Provider value={subscriptionValue}>
{children}
</LiveEventSubscriptionContext.Provider>
);
}