diff --git a/docs/architecture.md b/docs/architecture.md index 81207891..26298e4c 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -66,6 +66,7 @@ For example, a domain rule decides whether an import source may retry. The appli - **Setup access stays separate.** The owner's coding agent during setup keeps its one-time code and dedicated HTTP routes; its session is limited to setup and never becomes a person's agent connection. - **Transport and authority.** Stateless Streamable HTTP at `/api/mcp`, authorized by OAuth 2.1 with PKCE. Better Auth's OAuth provider is the authorization server at `/api/auth`, for public clients registered dynamically or through a client metadata document. Tokens are opaque and checked in the database on every call, so an ended connection is refused at once; each call resolves the person's security context afresh, as for their browser, and a token must name this instance's MCP address as its resource. Two scopes, read and write, are intersected with the person's permissions; each tool declares the scope it needs. Resources and prompts remain outside the initial scope. - **One connection per person.** Allowing an agent ends the person's other agents' tokens and consents; a password reset ends them all. A token is issued within the browser session that allowed it, so the server detaches tokens from expired browser sessions before issuing new ones, and the connection outlives the browser's sign-in. +- **Live updates.** A signed-in browser keeps one server-sent events stream open at `/api/live`. It carries only a nudge that the person's agent connection changed, never data; the page reads it again through GraphQL, so record scope stays in one place. Nudges follow the commit through the in-process event bus, and streams live in the one API process an instance runs. ## 4. Data access diff --git a/docs/qa/connect-your-agent.md b/docs/qa/connect-your-agent.md index 606a9bd0..bbaedcb5 100644 --- a/docs/qa/connect-your-agent.md +++ b/docs/qa/connect-your-agent.md @@ -33,7 +33,10 @@ A manual check of connecting a person's own agent over MCP against a development - [ ] "Your agent" is in the menu for every signed-in person, and its page shows the address the browser is on plus `/api/mcp`, with Copy. - [ ] On a self-hosted instance, a line under the address says agents work only at the address they connected to, with a link to set `SELF_HOSTED_PUBLIC_URL`. - [ ] The Claude, ChatGPT and "Claude Code or Codex" tabs show their steps; each command's Copy copies it with this address. -- [ ] Following the Claude Code tab connects it; coming back to the window shows the agent, "Connected …· Last active …" and a dot on the menu item. +- [ ] Following the Claude Code tab connects it; while the agent's Allow is chosen, Your agent (left open, not focused) turns to the agent, "Connected …· Last active …" and a dot on the menu item within a second. +- [ ] With Your agent open in two tabs, Disconnect in one turns the other to "You disconnected Claude Code." within a second. +- [ ] With Your agent open, a password reset in another browser turns it to "Claude Code was disconnected when your password was reset." within a second. +- [ ] After the network drops and comes back (DevTools offline, then online), a Disconnect made meanwhile shows without a reload. - [ ] Within an hour of connecting, the card opens with "Claude Code is connected. Nice."; × hides it. Each "Try asking" prompt's Copy copies it, and pasting it into a new session of the agent gets an answer (Codex loads Broccoli's tools only in a session started after login). - [ ] "Disconnect", then "Disconnect" in the confirm, shows "You disconnected Claude Code."; the agent's next call is refused. - [ ] After a password reset by email and signing in again, the page says the agent was disconnected when the password was reset. diff --git a/packages/api/importRules.json b/packages/api/importRules.json index 836fc151..bdecddc9 100644 --- a/packages/api/importRules.json +++ b/packages/api/importRules.json @@ -104,6 +104,7 @@ "src/core/applicationError", "src/core/auth", "src/core/db", + "src/core/events", "src/core/instance", "src/core/logger", "src/core/security" diff --git a/packages/api/src/core/agentAccess/app/agentConnectionService.int.test.ts b/packages/api/src/core/agentAccess/app/agentConnectionService.int.test.ts index 4e0b1f59..2ce2044e 100644 --- a/packages/api/src/core/agentAccess/app/agentConnectionService.int.test.ts +++ b/packages/api/src/core/agentAccess/app/agentConnectionService.int.test.ts @@ -1,6 +1,73 @@ import { afterEach, beforeEach, vi } from "vitest"; -import { ADMIN, agentConnectionService, expect, OWNER, SYSTEM, test } from "#test/support"; +import { + ADMIN, + agentConnectionService, + domainEvents, + expect, + OWNER, + SYSTEM, + test, +} from "#test/support"; + +test("disconnect publishes one connection change for the caller", async ({ orm, seed }) => { + const userId = OWNER.user!.userId; + await seed.agentConnection({ userId }); + await seed.agentConnection({ userId }); + const { transact, events } = domainEvents(orm); + + await agentConnectionService(orm, { transact }).disconnectMyAgent(OWNER); + + expect(events).toEqual([{ type: "agentAccess.connection_changed", payload: { userId } }]); +}); + +test("disconnect without a connection publishes nothing", async ({ orm }) => { + const { transact, events } = domainEvents(orm); + + await agentConnectionService(orm, { transact }).disconnectMyAgent(OWNER); + + expect(events).toEqual([]); +}); + +test("allowing an agent publishes a connection change for that person", async ({ orm, seed }) => { + const userId = OWNER.user!.userId; + const { clientId } = await seed.agentConnection({ userId }); + const { transact, events } = domainEvents(orm); + + await agentConnectionService(orm, { transact }).keepOnlyConnection(SYSTEM, { userId, clientId }); + + expect(events).toEqual([{ type: "agentAccess.connection_changed", payload: { userId } }]); +}); + +test("a password reset publishes one connection change for that person", async ({ orm, seed }) => { + const userId = OWNER.user!.userId; + await seed.agentConnection({ userId }); + await seed.agentConnection({ userId }); + const { transact, events } = domainEvents(orm); + + await agentConnectionService(orm, { transact }).endConnections(SYSTEM, { + userId, + reason: "ended_by_password_reset", + }); + + expect(events).toEqual([{ type: "agentAccess.connection_changed", payload: { userId } }]); +}); + +test("using a connection unused for 31 days publishes a connection change", async ({ + orm, + seed, +}) => { + const userId = OWNER.user!.userId; + const { clientId } = await seed.agentConnection({ + userId, + connectedAt: new Date("2026-10-12T12:00:00Z"), + }); + const { transact, events } = domainEvents(orm); + + await agentConnectionService(orm, { transact }).useAgentConnection(SYSTEM, { userId, clientId }); + + expect(events).toEqual([{ type: "agentAccess.connection_changed", payload: { userId } }]); +}); test("disconnect ends the caller's connection and refuses its next use", async ({ orm, seed }) => { const connectedAt = new Date("2026-11-11T12:00:00Z"); diff --git a/packages/api/src/core/agentAccess/app/agentConnectionService.ts b/packages/api/src/core/agentAccess/app/agentConnectionService.ts index b14be52e..cc68a61d 100644 --- a/packages/api/src/core/agentAccess/app/agentConnectionService.ts +++ b/packages/api/src/core/agentAccess/app/agentConnectionService.ts @@ -71,6 +71,7 @@ export function createAgentConnectionService({ if (!(await connections.findConnectionOf(userId, clientId))) { return { ended: [], allowed: null }; } + tx.publish("agentAccess.connection_changed", { userId }); return { ended: await connections.endConnections(userId, clientId, { reason: "replaced_by_another_agent", @@ -159,7 +160,9 @@ async function endAllConnections( const ended = await transact(async (tx) => { const connections = agentConnectionRepository(tx); await connections.lockPerson(userId); - return connections.endConnections(userId, null, { reason, endedAt: new Date() }); + const ended = await connections.endConnections(userId, null, { reason, endedAt: new Date() }); + if (ended.length > 0) tx.publish("agentAccess.connection_changed", { userId }); + return ended; }); if (ended.length > 0) { logger.info( @@ -178,10 +181,12 @@ async function endUnusedConnection( const ended = await transact(async (tx) => { const connections = agentConnectionRepository(tx); await connections.lockPerson(userId); - return connections.endConnection(userId, clientId, { + const ended = await connections.endConnection(userId, clientId, { reason: "ended_after_30_days_unused", endedAt: new Date(), }); + if (ended) tx.publish("agentAccess.connection_changed", { userId }); + return ended; }); if (ended) { logger.info( diff --git a/packages/api/src/core/agentAccess/app/registrations.ts b/packages/api/src/core/agentAccess/app/registrations.ts new file mode 100644 index 00000000..350133b1 --- /dev/null +++ b/packages/api/src/core/agentAccess/app/registrations.ts @@ -0,0 +1,10 @@ +// What agent access adds to core's shared registries. + +declare module "../../events" { + interface DomainEvents { + /** A person's agent connection was allowed or ended; their open pages read it again. */ + "agentAccess.connection_changed": { userId: string }; + } +} + +export {}; diff --git a/packages/api/src/core/agentAccess/http/index.ts b/packages/api/src/core/agentAccess/http/index.ts index b8abca24..7471c08e 100644 --- a/packages/api/src/core/agentAccess/http/index.ts +++ b/packages/api/src/core/agentAccess/http/index.ts @@ -1 +1,3 @@ export { agentDiscoveryRouter } from "./agentDiscoveryRouter"; +export { createLiveNudges } from "./liveNudges"; +export type { LiveNudges } from "./liveNudges"; diff --git a/packages/api/src/core/agentAccess/http/liveNudges.ts b/packages/api/src/core/agentAccess/http/liveNudges.ts new file mode 100644 index 00000000..d439c7fb --- /dev/null +++ b/packages/api/src/core/agentAccess/http/liveNudges.ts @@ -0,0 +1,82 @@ +import express from "express"; + +import { sendApplicationError } from "../../applicationError"; +import type { EventHandlers } from "../../events"; +import { requireSignedIn, type SecurityResolver } from "../../security"; + +/** Often enough that proxies never close a quiet stream. */ +const HEARTBEAT_MS = 25_000; +/** The only message a stream sends: read the agent connection again. Never data. */ +const AGENT_CONNECTION_CHANGED = `data: ${JSON.stringify({ type: "agentConnection.changed" })}\n\n`; + +/** + * `GET /api/live`: a server-sent events stream per signed-in browser tab that tells the page when + * to read again, so the page shows a change at once without the stream carrying any data. Streams + * live in this process; one API process serves an instance. + */ +export interface LiveNudges { + router: express.Router; + /** Nudges a person's open streams after their agent connection changes. */ + events: EventHandlers; + /** Ends every open stream, so the server can stop without waiting for them. */ + close(): void; +} + +export function createLiveNudges(resolveSecurity: SecurityResolver): LiveNudges { + const streams = new Map>(); + let closed = false; + + const router = express.Router().get("/live", async (request, response) => { + let userId: string; + try { + const { security, renewedCookies } = await resolveSecurity(request.headers); + const user = requireSignedIn(security).user; + if (!user) { + response.sendStatus(401); + return; + } + userId = user.userId; + if (renewedCookies.length > 0) response.set("Set-Cookie", renewedCookies); + } catch (error) { + sendApplicationError(error, response); + return; + } + // The tab may have gone, or the server begun stopping, while the session was read. + if (closed || response.destroyed) { + response.end(); + return; + } + response + .status(200) + .set({ + "Content-Type": "text/event-stream", + "Cache-Control": "no-store", + // Tells a buffering proxy to pass each message on at once. + "X-Accel-Buffering": "no", + }) + // A proxy may hold the headers back until the first bytes, so the stream opens with a comment. + .write(": open\n\n"); + const own = streams.get(userId) ?? new Set(); + streams.set(userId, own.add(response)); + const heartbeat = setInterval(() => response.write(": heartbeat\n\n"), HEARTBEAT_MS); + response.on("close", () => { + clearInterval(heartbeat); + own.delete(response); + if (own.size === 0) streams.delete(userId); + }); + }); + + return { + router, + events: { + "agentAccess.connection_changed": ({ userId }) => { + for (const stream of streams.get(userId) ?? []) stream.write(AGENT_CONNECTION_CHANGED); + return Promise.resolve(); + }, + }, + close() { + closed = true; + for (const stream of [...streams.values()].flatMap((own) => [...own])) stream.end(); + }, + }; +} diff --git a/packages/api/src/core/agentAccess/index.ts b/packages/api/src/core/agentAccess/index.ts index 95077c3f..f551f212 100644 --- a/packages/api/src/core/agentAccess/index.ts +++ b/packages/api/src/core/agentAccess/index.ts @@ -2,10 +2,11 @@ export { createAgentConnectionService } from "./app"; export { agentAccessPlugins, agentSecurityResolver } from "./betterAuth"; export { agentResourceMetadataOf } from "./domain"; -export { agentDiscoveryRouter } from "./http"; +export { agentDiscoveryRouter, createLiveNudges } from "./http"; // Public types. export type { AgentConnectionService } from "./app"; export type { AgentSecurityResolver } from "./betterAuth"; export type { StoredEndedAgentConnection } from "./db"; export type { AgentScope } from "./domain"; +export type { LiveNudges } from "./http"; diff --git a/packages/api/src/core/server.ts b/packages/api/src/core/server.ts index a97102d5..10016ab1 100644 --- a/packages/api/src/core/server.ts +++ b/packages/api/src/core/server.ts @@ -12,6 +12,8 @@ import { agentSecurityResolver, type AgentSecurityResolver, createAgentConnectionService, + createLiveNudges, + type LiveNudges, } from "./agentAccess"; import { authHandler, type DeploymentServer } from "./auth"; import { connectDatabase, createBeforeCommitBus, type Database, transactor } from "./db"; @@ -71,6 +73,7 @@ export async function startServer( let resolveAgentSecurity: AgentSecurityResolver; const agentConnections = createAgentConnectionService({ orm: database.orm, transact, logger }); let graphql: GraphqlApi; + let live: LiveNudges; try { const { manifests, setupDemo, addApprovedPeople, employeeDirectory, passwordResetEmail } = createModules({ @@ -106,6 +109,8 @@ export async function startServer( const auth = deployment.auth; resolveSecurity = (headers) => resolveSecurityContext(auth, database.orm, employeeDirectory, headers); + live = createLiveNudges(resolveSecurity); + eventBus.subscribe(live.events); resolveAgentSecurity = agentSecurityResolver( auth, database.orm, @@ -143,6 +148,7 @@ export async function startServer( ); } } + app.use("/api", live.router); app.all("/api/auth/{*path}", authHandler(deployment.auth)); app.use(graphql.router); if (webApp) { @@ -166,6 +172,7 @@ export async function startServer( stop: () => { stopping ??= (async () => { stopped.abort(); + live.close(); // Drains in-flight requests, then closes the HTTP server. await graphql.stop(); await jobRunner.stop(); diff --git a/packages/api/test/e2e/liveNudges.test.ts b/packages/api/test/e2e/liveNudges.test.ts new file mode 100644 index 00000000..06f62efe --- /dev/null +++ b/packages/api/test/e2e/liveNudges.test.ts @@ -0,0 +1,75 @@ +import { + connectAgent, + expect, + registerAgentClient, + sseMessages, + test, + TEST_OWNER, +} from "#test/support"; + +test("connecting an agent and disconnecting each send the person's stream one nudge", async ({ + env, +}) => { + const { browser } = await env.signInAsOwner(); + const clientId = await registerAgentClient(env, "Claude Code"); + const response = await browser.get("/api/live"); + const messages = sseMessages(response); + try { + expect(response.status).toBe(200); + expect(response.headers.get("content-type")).toMatch(/^text\/event-stream/); + expect(response.headers.get("cache-control")).toBe("no-store"); + + await connectAgent(env, browser, { clientId }); + + expect(await messages.next()).toBe('{"type":"agentConnection.changed"}'); + + const disconnected = await browser.graphql( + "mutation { mcpAgents { disconnect { endedAgentConnection { reason } } } }", + ); + + expect(await disconnected.json()).toEqual({ + data: { + mcpAgents: { disconnect: { endedAgentConnection: { reason: "DISCONNECTED_BY_PERSON" } } }, + }, + }); + expect(await messages.next()).toBe('{"type":"agentConnection.changed"}'); + expect(await messages.next(100)).toBeNull(); + } finally { + await messages.cancel(); + } +}); + +test("refuses a live stream without a session", async ({ env }) => { + await env.signInAsOwner(); + + const response = await env.browser().get("/api/live"); + + try { + expect(response.status).toBe(401); + } finally { + await response.body?.cancel(); + } +}); + +test("a person's stream receives no nudge when another person's connection changes", async ({ + env, +}) => { + const { browser: owner } = await env.signInAsOwner(); + const credentials = { email: "another@example.com", password: TEST_OWNER.password }; + await env.seed.user(credentials); + const { browser } = await env.signIn(credentials); + const clientId = await registerAgentClient(env, "Claude Code"); + const response = await browser.get("/api/live"); + const messages = sseMessages(response); + try { + expect(response.status).toBe(200); + + await connectAgent(env, owner, { clientId }); + await connectAgent(env, browser, { clientId }); + + expect(await messages.next()).toBe('{"type":"agentConnection.changed"}'); + expect(await messages.next(100)).toBeNull(); + } finally { + await messages.cancel(); + } +}); diff --git a/packages/api/test/support/agentConnectionService.ts b/packages/api/test/support/agentConnectionService.ts index 7b0360a2..c68f4f0d 100644 --- a/packages/api/test/support/agentConnectionService.ts +++ b/packages/api/test/support/agentConnectionService.ts @@ -1,12 +1,15 @@ import { createAgentConnectionService } from "../../src/core/agentAccess"; -import type { Orm } from "../../src/core/db"; +import type { Orm, Transact } from "../../src/core/db"; import { createLogger } from "../../src/core/logger"; import { testTransactor } from "./testTransactor"; /** People's agent connections on `orm`, with a silent logger. */ -export const agentConnectionService = (orm: Orm) => +export const agentConnectionService = ( + orm: Orm, + { transact = testTransactor(orm) }: { transact?: Transact } = {}, +) => createAgentConnectionService({ orm, - transact: testTransactor(orm), + transact, logger: createLogger("silent"), }); diff --git a/packages/api/test/support/domainEvents.ts b/packages/api/test/support/domainEvents.ts index 2b9c9bc1..32802d1a 100644 --- a/packages/api/test/support/domainEvents.ts +++ b/packages/api/test/support/domainEvents.ts @@ -2,11 +2,15 @@ import { type Orm, transactor } from "../../src/core/db"; import { createEventBus, type DomainEvent } from "../../src/core/events"; import { createLogger } from "../../src/core/logger"; -/** Task events delivered after the transaction commits. */ +/** Task and agent connection events delivered after the transaction commits. */ export function domainEvents(orm: Orm) { const events: DomainEvent[] = []; const eventBus = createEventBus(createLogger("silent")); eventBus.subscribe({ + "agentAccess.connection_changed": (payload) => { + events.push({ type: "agentAccess.connection_changed", payload }); + return Promise.resolve(); + }, "tasks.task_created": (payload) => { events.push({ type: "tasks.task_created", payload }); return Promise.resolve(); diff --git a/packages/api/test/support/index.ts b/packages/api/test/support/index.ts index d7773f3a..906128dd 100644 --- a/packages/api/test/support/index.ts +++ b/packages/api/test/support/index.ts @@ -81,6 +81,7 @@ export type { EmployeeServices } from "./services"; export { processService, transactWithProcesses } from "./processService"; export { eventProcess, seedProcess } from "./seedProcess"; export { domainEvents } from "./domainEvents"; +export { sseMessages } from "./sseMessages"; export { employeeSetup, fileStorage, diff --git a/packages/api/test/support/sseMessages.ts b/packages/api/test/support/sseMessages.ts new file mode 100644 index 00000000..7c5367ee --- /dev/null +++ b/packages/api/test/support/sseMessages.ts @@ -0,0 +1,42 @@ +import assert from "node:assert/strict"; + +/** Reads SSE data, preserving messages across chunks; a timeout returns null and cancels the stream. */ +export function sseMessages(response: Response) { + assert(response.body, "The SSE response has no body"); + const reader: ReadableStreamDefaultReader = response.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + + return { + cancel: () => reader.cancel(), + async next(timeoutMs = 2_000): Promise { + let timer: ReturnType; + const timeout = new Promise((resolve) => { + timer = setTimeout(() => resolve(null), timeoutMs); + }); + try { + for (;;) { + const boundary = /\r?\n\r?\n/.exec(buffer); + if (boundary) { + const lines = buffer.slice(0, boundary.index).split(/\r?\n/); + buffer = buffer.slice(boundary.index + boundary[0].length); + const data = lines.filter((line) => line.startsWith("data:")); + if (data.length > 0) { + return data.map((line) => line.slice(5).replace(/^ /, "")).join("\n"); + } + continue; + } + const chunk = await Promise.race([reader.read(), timeout]); + if (chunk === null) { + await reader.cancel(); + return null; + } + assert(!chunk.done, "The SSE stream ended before the next message"); + buffer += decoder.decode(chunk.value, { stream: true }); + } + } finally { + clearTimeout(timer!); + } + }, + }; +} diff --git a/packages/web/src/features/mcpAgents/YourAgentPage/WaitingForAgent.tsx b/packages/web/src/features/mcpAgents/YourAgentPage/WaitingForAgent.tsx index 3373b065..e11e7fc2 100644 --- a/packages/web/src/features/mcpAgents/YourAgentPage/WaitingForAgent.tsx +++ b/packages/web/src/features/mcpAgents/YourAgentPage/WaitingForAgent.tsx @@ -2,7 +2,7 @@ import { t } from "@/i18n"; import "./waitingForAgent.css"; -/** Shown until the person's agent connects; the page checks again when they come back to it. */ +/** Shown until the person's agent connects, which the page shows the moment it happens. */ export function WaitingForAgent() { return (

diff --git a/packages/web/src/i18n/en/mcpAgents.ts b/packages/web/src/i18n/en/mcpAgents.ts index 1e6ab1da..a92b539c 100644 --- a/packages/web/src/i18n/en/mcpAgents.ts +++ b/packages/web/src/i18n/en/mcpAgents.ts @@ -51,7 +51,7 @@ export const mcpAgentsEn = { "Then run codex mcp login broccoli, choose Allow, and start a new Codex session.", "mcpAgents.cli.copyCommand": "Copy command", "mcpAgents.cli.other": "Other clients: add the address as a remote HTTP server with OAuth.", - "mcpAgents.waiting": "Waiting for your agent…", + "mcpAgents.waiting": "Waiting for your agent… This page updates the moment you choose Allow.", "mcpAgents.connected.badge": "Connected", "mcpAgents.connected.since": "Acts as you, {name} · Connected {date} · Last active {lastActive}", "mcpAgents.welcome.title": "{client} is connected. Nice.", diff --git a/packages/web/src/shell/AppShell.tsx b/packages/web/src/shell/AppShell.tsx index 782aa121..26769d81 100644 --- a/packages/web/src/shell/AppShell.tsx +++ b/packages/web/src/shell/AppShell.tsx @@ -8,6 +8,7 @@ import { cn } from "@/lib/class-names"; import { LeftNav } from "./LeftNav"; import { useFocusHeadingOnNavigation } from "./useFocusHeadingOnNavigation"; import { useInSettings } from "./useInSettings"; +import { useLiveAgentConnection } from "./useLiveAgentConnection"; /** * The window-sized app frame: navigation above or beside the Canvas, a card that @@ -17,6 +18,7 @@ export function AppShell({ children }: { children: ReactNode }) { const main = useRef(null); const inSettings = useInSettings(); useFocusHeadingOnNavigation(main); + useLiveAgentConnection(); return ( <> diff --git a/packages/web/src/shell/useLiveAgentConnection.ts b/packages/web/src/shell/useLiveAgentConnection.ts new file mode 100644 index 00000000..eef43057 --- /dev/null +++ b/packages/web/src/shell/useLiveAgentConnection.ts @@ -0,0 +1,35 @@ +// eslint-disable-next-line no-restricted-imports -- A nudge refetches open queries, which the cache alone cannot. +import { useApolloClient } from "@apollo/client/react"; +import { useEffect } from "react"; + +/** The open queries that show the person's agent connection. */ +const AGENT_CONNECTION_QUERIES = new Set(["YourAgent", "LeftNavAgentConnection"]); + +/** + * Keeps the person's agent connection current while they are signed in: the server nudges this + * tab when their agent connects or a connection ends, and the open queries that show it read again. + * They read again each time the stream opens too, for a change made while it was not yet or no + * longer open; the browser reopens a dropped stream by itself. + */ +export function useLiveAgentConnection(): void { + const apolloClient = useApolloClient(); + useEffect(() => { + const refetch = () => + void apolloClient.refetchQueries({ + include: "active", + onQueryUpdated: (query) => + AGENT_CONNECTION_QUERIES.has(query.queryName ?? "") && + // A new read: one already in flight may have read the state before the change. + apolloClient.query({ + query: query.query, + variables: query.variables, + fetchPolicy: "network-only", + context: { queryDeduplication: false }, + }), + }); + const stream = new EventSource("/api/live"); + stream.onmessage = refetch; + stream.onopen = refetch; + return () => stream.close(); + }, [apolloClient]); +}