[posthog-monitor] broker: end the session when the grant is revoked, don't poll it forever

app/src/main/services/broker/connect.test.ts

@@ -1,5 +1,7 @@
 import { Database } from "bun:sqlite";
 import { afterEach, beforeEach, describe, expect, test } from "bun:test";
+import { UnauthorizedError } from "@modelcontextprotocol/sdk/client/auth.js";
+import { InvalidGrantError } from "@modelcontextprotocol/sdk/server/auth/errors.js";
 import { ErrorCode, McpError } from "@modelcontextprotocol/sdk/types.js";
 import type { Account, OrderStatus, Portfolio, Position, Quote } from "@shared/broker";
 import { drizzle } from "drizzle-orm/bun-sqlite";
@@ -49,12 +51,19 @@ class FakeAdapter {
       pending.reject(new ConnectSuperseded());
     }
   }
-  /** Land call #i as connected. */
+  /** Land call #i as connected (and store tokens, as a real connect does). */
   succeed(i: number) {
     this.connected = true;
+    this.tokens = true;
     Object.assign(this.calls[i], { done: true });
     this.calls[i].resolve();
   }
+  /** Stored tokens, so `svc.isAuthorized()` means something: true after a connect
+   *  lands, false after `reset()` drops them. */
+  tokens = false;
+  hasTokens() {
+    return this.tokens;
+  }
   fail(i: number, err: unknown) {
     Object.assign(this.calls[i], { done: true });
     this.calls[i].reject(err);
@@ -67,6 +76,7 @@ class FakeAdapter {
     this.resets++;
     this.cancelConnect();
     this.connected = false;
+    this.tokens = false;
   }
   /** Set to make the service pick an account and start polling. */
   account: Account | null = null;
@@ -416,6 +426,75 @@ describe("BrokerService poll loop — network outages are not app errors", () =>
     });
   });
 
+  test("a revoked grant ends the session instead of retrying it forever", async () => {
+    const { svc, adapter, client } = await connectedWithAccount();
+    // Production shape: the stored refresh token was revoked, so the first poll to use it
+    // throws and every later one throws identically. One install emitted 167 of these in
+    // 26 minutes — one per poll — while its UI still read `connected`.
+    adapter.pollScript = async () => {
+      throw new InvalidGrantError("token revoked");
+    };
+    expect(svc.isAuthorized()).toBe(true);
+    await poll(svc);
+    // Reported once, then the session is over: tokens dropped, so the panel shows the
+    // Connect CTA (a fresh consent) rather than pretending it can reconnect silently.
+    // The end itself is counted, so a stranded install is visible in telemetry.
+    expect(events(client)).toEqual(["app_error", "broker_session_ended"]);
+    expect(client.events[0].properties).toMatchObject({
+      subsystem: "broker",
+      error_name: "InvalidGrantError",
+      source: "caught",
+    });
+    expect(client.events[1].properties).toMatchObject({ reason: "invalid_grant" });
+    expect(svc.getStatus()).toBe("disconnected");
+    expect(adapter.resets).toBe(1);
+    expect(svc.isAuthorized()).toBe(false);
+
+    // And the loop is genuinely stopped — a further poll reports nothing new.
+    await poll(svc);
+    expect(events(client)).toEqual(["app_error", "broker_session_ended"]);
+  });
+
+  test("a revoked grant on the connect-time first poll ends the session without hanging connect()", async () => {
+    const { svc, adapter, client } = setup();
+    adapter.account = { accountNumber: "A1", agentic: true } as Account;
+    // `runConnect` awaits the first poll itself, while `inflight` still holds runConnect.
+    // Ending the session there via `disconnect()` — which awaits `inflight` — waited on
+    // itself: connect() never resolved, status stuck on `connected`, Reset hung too.
+    adapter.pollScript = async () => {
+      throw new InvalidGrantError("token revoked");
+    };
+    const c = svc.connect();
+    await tick();
+    adapter.succeed(0);
+    await c; // hung forever before the fix
+    expect(svc.getStatus()).toBe("disconnected");
+    expect(adapter.resets).toBe(1);
+    expect(events(client)).toEqual([
+      "broker_connect_started",
+      "broker_connected",
+      "app_error",
+      "broker_session_ended",
+    ]);
+    // No poller was started for the session that no longer exists…
+    expect((svc as unknown as { timer: unknown }).timer).toBeNull();
+    // …and the user's Reset still works.
+    await svc.disconnect();
+    expect(svc.getStatus()).toBe("disconnected");
+  });
+
+  test("a 401 mid-session is NOT treated as a dead grant — good tokens are kept", async () => {
+    const { svc, adapter, client } = await connectedWithAccount();
+    // `isReauthRequired` on the connect path also counts UnauthorizedError, but on the
+    // poll path a transient/endpoint-specific 401 must not cost the user their session.
+    adapter.pollScript = async () => {
+      throw new UnauthorizedError("nope");
+    };
+    await poll(svc);
+    expect(events(client)).toEqual(["app_error"]);
+    expect(svc.getStatus()).toBe("connected");
+  });
+
   test("disconnect closes the books: no broker_online spanning a deliberate disconnect", async () => {
     const { svc, adapter, client } = await connectedWithAccount();
     adapter.pollScript = async () => {

app/src/main/services/broker/index.ts

@@ -21,7 +21,7 @@ import { analytics } from "../analytics";
 import { bus } from "../event-bus";
 import type { SettingsService } from "../settings";
 import { type BrokerAdapter, ConnectSuperseded } from "./adapter";
-import { brokerErrorCode, isTransientNetworkError } from "./network-error";
+import { brokerErrorCode, isDeadGrantError, isTransientNetworkError } from "./network-error";
 import { orderNotification, terminalTransition } from "./order-notify";
 
 /**
@@ -208,7 +208,9 @@ export class BrokerService {
       this.setStatus("connected");
       analytics.track("broker_connected");
       await this.pollOnce();
-      this.startPolling();
+      // The first poll can end the session itself (a dead grant → `forgetSession`):
+      // don't start a poller for a session that no longer exists.
+      if (this.status === "connected") this.startPolling();
     } catch (err) {
       // A newer connect took this one over (the user clicked Connect again): not a
       // failure — the newer attempt owns the status now, so leave it and report nothing.
@@ -235,13 +237,27 @@ export class BrokerService {
     // Let the in-flight connect unwind (a superseded one resolves quietly and leaves the
     // status alone — it's ours to set below).
     await this.inflight?.catch(() => {});
+    this.forgetSession();
+  }
+
+  /**
+   * Forget the session: stop polling, clear per-session state, status `disconnected`
+   * (the Connect CTA comes back; the next Connect is a fresh consent), drop the tokens.
+   * Shared by Reset and the poll loop's dead-grant path, which must call this and **not**
+   * `disconnect()`: that awaits `inflight`, and a dead grant can surface on the *first*
+   * poll, which `runConnect` awaits while `inflight` still holds `runConnect` — the await
+   * would wait on itself forever (status stuck on `connected`, Reset hung on it too).
+   * The token drop goes last: it is the one step that touches the DB, so status is
+   * already settled if it throws.
+   */
+  private forgetSession(): void {
     this.stopPolling();
-    this.adapter.reset();
     this.account = null;
     this.ledger.clear();
     this.offlineSince = null; // an outage doesn't span a deliberate disconnect
     this.failedPolls = 0;
     this.setStatus("disconnected");
+    this.adapter.reset();
   }
 
   /** True if we already have tokens and can connect without a browser. */
@@ -373,7 +389,15 @@ export class BrokerService {
       if (this.status !== "connected") return;
       // Network outage (laptop sleep/wake, Wi‑Fi blip) → broker_offline; else app_error.
       if (isTransientNetworkError(err)) this.noteOffline(err);
-      else analytics.trackError("broker", err, "caught", brokerErrorCode(err));
+      else {
+        analytics.trackError("broker", err, "caught", brokerErrorCode(err));
+        // A revoked grant never comes back (`isDeadGrantError`): end the session the way
+        // Reset does, instead of throwing this same error every 5–10 s over stale data.
+        if (isDeadGrantError(err)) {
+          this.forgetSession();
+          analytics.track("broker_session_ended", { reason: "invalid_grant" });
+        }
+      }
     } finally {
       this.polling = false;
     }

app/src/main/services/broker/network-error.ts

@@ -1,3 +1,4 @@
+import { InvalidGrantError } from "@modelcontextprotocol/sdk/server/auth/errors.js";
 import { ErrorCode, McpError } from "@modelcontextprotocol/sdk/types.js";
 import { errorCodeOf } from "@shared/analytics";
 
@@ -47,6 +48,20 @@ export function isTransientNetworkError(err: unknown): boolean {
   return err instanceof TypeError && err.message === "fetch failed";
 }
 
+/**
+ * Has the authorization server rejected our refresh token for good? `invalid_grant` is
+ * the one auth failure that is *never* worth retrying: the grant is expired or revoked,
+ * and no amount of waiting brings it back — only a fresh consent does.
+ *
+ * Deliberately narrower than `isReauthRequired` (`robinhood/client.ts`), which also
+ * counts a resource `UnauthorizedError`. That breadth is right when deciding whether to
+ * start a consent the user just asked for; it is wrong on the poll path, where a
+ * transient or endpoint-specific 401 would throw away credentials that still work.
+ */
+export function isDeadGrantError(err: unknown): boolean {
+  return err instanceof InvalidGrantError;
+}
+
 /**
  * The bounded telemetry code for a broker error: `errorCodeOf` (err.code / err.cause.code)
  * for the transport cases, plus the MCP layer — an `McpError`'s `code` is a *number*

app/src/shared/analytics.ts

@@ -190,6 +190,13 @@ export const TELEMETRY_EVENTS = {
     offline_ms: z.number().int().nonnegative(),
     failed_polls: z.number().int().nonnegative(),
   }),
+  /**
+   * The poll loop found the stored grant revoked (`invalid_grant` on refresh) and ended
+   * the session itself: tokens dropped, status `disconnected`, Connect CTA back. A
+   * lifecycle event beside `broker_offline`/`broker_online`, so the broker funnel can
+   * count stranded installs without reading error dashboards for `InvalidGrantError`.
+   */
+  broker_session_ended: z.strictObject({ reason: z.enum(["invalid_grant"]) }),
 
   // autonomy
   schedule_created: z.strictObject({