Realtime gateway: sharee access-check endpoint + SSE reconnect cursor

.changeset/realtime-gateway-access-and-reconnect.md

@@ -0,0 +1,10 @@
+---
+"@agent-native/core": minor
+---
+
+Realtime sync: two hosted-gateway follow-ups. Both are opt-in and leave apps without hosted-realtime config unchanged.
+
+- Hosted SSE reconnect now sends the client's cursor (`&since=`) on the gateway stream URL, so the gateway's connect-time catch-up replays events written during the reconnect gap immediately instead of deferring them to the next poll. First connect (cursor 0) is unaffected. Because a `since=` only takes effect when a new `EventSource` is constructed (the browser's own auto-reconnect reuses the URL frozen at construction), the hosted transport now owns reconnects: on a stream error it closes the stream and schedules its own reconnect so the next connect rebuilds the URL from the current cursor, keeping the token on transient errors and reminting on a closed stream. A late error from a replaced (stale) stream is ignored so it cannot tear down the current one. Local mode keeps native EventSource reconnect.
+- New gateway access-check tokens (`signGatewayAccessToken`/`verifyGatewayAccessToken`, exported from `./server/short-lived-token`): per-project HMAC key, a `typ` discriminator (not interchangeable with subscribe or media tokens), and the full access query (`resourceType`/`resourceId`/`userEmail`/`orgId`) bound into the signature so the app authenticates the params, not just the caller. `verifyGatewayAccessToken` also accepts an optional expected `projectId` to bind the channel (mirroring `verifyRealtimeSubscribeToken`); the `can-see` endpoint passes its own project id when known.
+- New endpoint `GET /_agent-native/can-see` mounted by core-routes. The hosted Realtime Gateway has no copy of an app's shareable-resource registry, so it calls this to resolve sharee visibility: the endpoint verifies a gateway access-check token against the app's per-project secret, runs the app's registry-based `resolveAccess`, and returns `{ allowed }`. Fails closed (`allowed: false`) on an unknown resource type or lookup error; 404 when the app has no realtime secret; `Cache-Control: private, no-store`.
+- New `AppSyncStateOptions.dbAssignedVersions` (default off): allocate durable-event versions from the app's Postgres DB — a one-row allocator advanced with `GREATEST(v + 1, epoch_ms_now)` inside the same autocommit insert statement — instead of the per-writer in-memory clock counter. Hosted realtime has multiple writers per app DB (the app's serverless instances plus gateway instances), where clock skew can assign a LOWER version to a LATER event and a client whose cursor passed the higher value filters the later event out permanently; DB allocation serializes on the allocator row's lock so version order equals commit order across all writers. Buffer/emit defer until the allocated version returns (no provisional version ever reaches clients); a deterministic-id dedupe loser adopts the winner's version via `ON CONFLICT DO UPDATE ... RETURNING`; on DB failure the writer falls back to clock allocation (logged) so the in-process fast path never stalls. Versions stay epoch-ms scale, so existing cursors, seeds, and lag metrics are unaffected. The default instance enables this automatically when `AGENT_NATIVE_REALTIME_TRANSPORT=hosted` with a gateway URL; self-hosted apps are byte-identical. Postgres only; ignored on SQLite.

packages/core/src/a2a/client.spec.ts

@@ -9,6 +9,28 @@ import {
   signA2AToken,
 } from "./client.js";
 
+// ssrfSafeFetch does a REAL node:dns lookup before calling fetch. Under fake
+// timers that wall-clock work can take seconds on CI resolvers (agent.test is
+// not a real host), so fake time races past request timeouts and deadlines
+// before the stubbed fetch is ever reached. Keep the synchronous private-host
+// check (the blocking test relies on it; IP literals need no DNS) and skip
+// only the DNS phase — full SSRF behavior is covered by url-safety's own spec.
+vi.mock("../extensions/url-safety.js", async (importOriginal) => {
+  const original =
+    await importOriginal<typeof import("../extensions/url-safety.js")>();
+  return {
+    ...original,
+    ssrfSafeFetch: async (url: string, init?: RequestInit) => {
+      if (original.isBlockedExtensionUrl(url)) {
+        throw new Error(
+          `SSRF blocked: refusing to fetch private/internal address (${url})`,
+        );
+      }
+      return fetch(url, init);
+    },
+  };
+});
+
 describe("A2AClient", () => {
   const originalEnv = { ...process.env };
 
@@ -248,7 +270,17 @@ describe("A2AClient", () => {
       { role: "user", parts: [{ type: "text", text: "hello" }] },
       { timeoutMs: 60_000, pollIntervalMs: 1_000 },
     );
-    const assertion = expect(result).resolves.toMatchObject({
+    // Attach a handler before advancing timers so an unexpected rejection is
+    // never reported as unhandled while the fake clock is moving.
+    void result.catch(() => undefined);
+
+    // Same coarse-interval pacing rationale as the hung-poll test above.
+    await vi.waitFor(() => expect(taskReads).toBeGreaterThan(0), {
+      interval: 100,
+      timeout: 5_000,
+    });
+    await vi.runAllTimersAsync();
+    await expect(result).resolves.toMatchObject({
       status: {
         state: "completed",
         message: {
@@ -258,10 +290,6 @@ describe("A2AClient", () => {
         },
       },
     });
-
-    while (taskReads === 0) await vi.advanceTimersByTimeAsync(1);
-    await vi.advanceTimersByTimeAsync(16_000);
-    await assertion;
     expect(firstPollSignal?.aborted).toBe(true);
     expect(taskReads).toBe(2);
   }, 30_000);

packages/core/src/client/use-db-sync.gateway-reconnect.spec.tsx

@@ -0,0 +1,132 @@
+// @vitest-environment happy-dom
+
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+
+import {
+  _resetSyncTransportRegistryForTests,
+  subscribeSyncEvents,
+} from "./use-db-sync";
+
+/** Minimal EventSource stand-in that records every constructed instance so the
+ * test can inspect the URL each connect was built with. */
+class FakeEventSource {
+  static readonly CONNECTING = 0;
+  static readonly OPEN = 1;
+  static readonly CLOSED = 2;
+  static instances: FakeEventSource[] = [];
+  readyState = FakeEventSource.CONNECTING;
+  onopen: (() => void) | null = null;
+  onerror: (() => void) | null = null;
+  onmessage: ((message: { data: string }) => void) | null = null;
+  addEventListener(): void {}
+  constructor(readonly url: string) {
+    FakeEventSource.instances.push(this);
+  }
+  close(): void {
+    this.readyState = FakeEventSource.CLOSED;
+  }
+}
+
+describe("hosted SSE reconnect ownership", () => {
+  beforeEach(() => {
+    vi.useFakeTimers();
+    FakeEventSource.instances = [];
+    _resetSyncTransportRegistryForTests();
+    vi.stubGlobal("EventSource", FakeEventSource);
+    vi.stubGlobal(
+      "fetch",
+      vi.fn(async (input: RequestInfo | URL) => {
+        const url = String(input);
+        if (url.includes("/_agent-native/realtime-token")) {
+          return {
+            ok: true,
+            json: async () => ({ token: "tok-1", ttlSeconds: 600 }),
+          };
+        }
+        return { ok: true, json: async () => ({ version: 0, events: [] }) };
+      }),
+    );
+    (
+      window as unknown as { __AGENT_NATIVE_CONFIG__: unknown }
+    ).__AGENT_NATIVE_CONFIG__ = {
+      realtime: { transport: "hosted", gatewayBaseUrl: "https://gw.example" },
+    };
+  });
+
+  afterEach(() => {
+    _resetSyncTransportRegistryForTests();
+    delete (window as unknown as { __AGENT_NATIVE_CONFIG__?: unknown })
+      .__AGENT_NATIVE_CONFIG__;
+    vi.unstubAllGlobals();
+    vi.useRealTimers();
+  });
+
+  it("rebuilds the stream URL with the current cursor on a browser-managed reconnect", async () => {
+    const unsub = subscribeSyncEvents({ onEvents: () => {} });
+
+    // Mint resolves, then the first stream opens with the token and no cursor.
+    await vi.advanceTimersByTimeAsync(200);
+    const first = FakeEventSource.instances.at(-1)!;
+    expect(first.url).toContain("token=tok-1");
+    expect(first.url).not.toContain("since=");
+
+    first.readyState = FakeEventSource.OPEN;
+    first.onopen?.();
+
+    // A delivered batch advances the transport cursor to 100.
+    first.onmessage?.({
+      data: JSON.stringify({
+        type: "batch",
+        version: 100,
+        events: [
+          { version: 100, source: "app-state", type: "change", key: "*" },
+        ],
+      }),
+    });
+
+    // Transient (CONNECTING) error: the browser would auto-reconnect this same
+    // instance with its frozen URL. We must own it and close the stream.
+    first.readyState = FakeEventSource.CONNECTING;
+    first.onerror?.();
+    expect(first.readyState).toBe(FakeEventSource.CLOSED);
+
+    // The owned reconnect builds a NEW stream carrying the current cursor and
+    // the still-valid token (no re-mint on a transient error).
+    await vi.advanceTimersByTimeAsync(1500);
+    const second = FakeEventSource.instances.at(-1)!;
+    expect(second).not.toBe(first);
+    expect(second.url).toContain("since=100");
+    expect(second.url).toContain("token=tok-1");
+
+    unsub();
+  });
+
+  it("ignores a late error from a replaced (stale) stream", async () => {
+    const unsub = subscribeSyncEvents({ onEvents: () => {} });
+    await vi.advanceTimersByTimeAsync(200);
+    const first = FakeEventSource.instances.at(-1)!;
+    first.readyState = FakeEventSource.OPEN;
+    first.onopen?.();
+
+    // Force a reconnect so `first` is replaced by `second`.
+    first.readyState = FakeEventSource.CONNECTING;
+    first.onerror?.();
+    await vi.advanceTimersByTimeAsync(1500);
+    const second = FakeEventSource.instances.at(-1)!;
+    expect(second).not.toBe(first);
+    second.readyState = FakeEventSource.OPEN;
+    second.onopen?.();
+
+    const countBefore = FakeEventSource.instances.length;
+    // The stale `first` fires a late error: the guard must ignore it so the
+    // healthy `second` is neither closed nor triggers a spurious reconnect.
+    first.readyState = FakeEventSource.CONNECTING;
+    first.onerror?.();
+    await vi.advanceTimersByTimeAsync(1500);
+
+    expect(second.readyState).toBe(FakeEventSource.OPEN);
+    expect(FakeEventSource.instances.length).toBe(countBefore);
+
+    unsub();
+  });
+});

packages/core/src/client/use-db-sync.ts

@@ -330,9 +330,11 @@ class SyncTransport {
 
   private get activeSseUrl(): string | false {
     if (this.mode === "hosted" && this.gateway) {
-      return this.token
-        ? `${this.gateway.sseUrl}?token=${encodeURIComponent(this.token)}`
-        : this.gateway.sseUrl;
+      if (!this.token) return this.gateway.sseUrl;
+      const base = `${this.gateway.sseUrl}?token=${encodeURIComponent(this.token)}`;
+      // Cursor lets the gateway replay the reconnect gap on connect instead of
+      // deferring it to the next poll; 0 on first connect means nothing to replay.
+      return this.versionRef > 0 ? `${base}&since=${this.versionRef}` : base;
     }
     return this.sseUrl;
   }
@@ -656,27 +658,27 @@ class SyncTransport {
       this.schedulePoll();
     };
     source.onerror = () => {
+      // A replaced/closed source can still fire late; ignore it so it can't
+      // flip the connected state or tear down the current stream.
+      if (this.eventSource !== source) return;
       this.setSseConnected(false);
-      // When the browser gives up permanently (HTTP error → readyState
-      // CLOSED), it won't auto-reconnect. Drop the ref so a later
-      // connectEvents() (on focus/visibility) can establish a fresh stream;
-      // otherwise the non-null closed `eventSource` blocks reconnection and
-      // we'd be stuck on polling-only forever.
+      if (this.mode === "hosted" && this.gateway) {
+        // Browser auto-reconnect reuses the URL frozen at construction, so it
+        // would replay from a stale `since`. Own the reconnect so the next
+        // connect rebuilds activeSseUrl from the current versionRef. CLOSED also
+        // refreshes the token (expired/rotated/deploy); CONNECTING keeps it. A
+        // successful reconnect resets the count in onopen; a hard-down gateway
+        // trips the threshold and health-gates to local.
+        if (source.readyState === EventSource.CLOSED) this.token = null;
+        this.closeEvents();
+        this.onGatewayTransientFailure();
+        if (this.mode === "hosted") this.scheduleGatewayReconnect();
+        return;
+      }
+      // Local mode: native EventSource reconnect is fine. Drop a CLOSED ref so a
+      // later connectEvents() (focus/visibility) can establish a fresh stream.
       if (source.readyState === EventSource.CLOSED) {
         this.eventSource = null;
-        if (this.mode === "hosted" && this.gateway) {
-          // A closed gateway stream is most likely an expired token or a
-          // request-timeout/deploy cycle. Re-mint and reconnect with jitter;
-          // this is NOT the poll-401 cooldown path. Each closed stream counts
-          // toward the unhealthy threshold so a hard-down gateway (or one
-          // rejecting our tokens) health-gates to local instead of looping
-          // mint+connect forever; a successful reconnect resets the count in
-          // onopen above.
-          this.token = null;
-          this.onGatewayTransientFailure();
-          if (this.mode === "hosted") this.scheduleGatewayReconnect();
-          return;
-        }
       }
       this.schedulePoll();
     };

packages/core/src/server/core-routes-plugin.ts

@@ -159,6 +159,7 @@ import {
   markDefaultPluginProvided,
   trackPluginInit,
 } from "./framework-request-handler.js";
+import { createGatewayAccessCheckHandler } from "./gateway-access-check.js";
 import { getAppBasePath, getOrigin } from "./google-oauth.js";
 import { createGoogleRealtimeSessionHandler } from "./google-realtime-session.js";
 import {
@@ -1429,6 +1430,8 @@ export function createCoreRoutesPlugin(
         `${P}/realtime-token`,
         createRealtimeTokenHandler(),
       );
+      // Sharee visibility check for the hosted gateway
+      getH3App(nitroApp).use(`${P}/can-see`, createGatewayAccessCheckHandler());
 
       // SSE
       if (!options.disableSSE) {