feat: show your agent connecting and disconnecting at once on Your agent

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
 

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.

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"

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");

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(

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 {};

packages/api/src/core/agentAccess/http/index.ts

@@ -1 +1,3 @@
 export { agentDiscoveryRouter } from "./agentDiscoveryRouter";
+export { createLiveNudges } from "./liveNudges";
+export type { LiveNudges } from "./liveNudges";

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<string, Set<express.Response>>();
+  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();
+    },
+  };
+}

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";

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();

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: "[email protected]", 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();
+  }
+});

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"),
   });