feat: bind snapshots and refs to exact targets

.changeset/clean-tabs-snapshot.md

@@ -0,0 +1,22 @@
+---
+"@understudy/protocol": minor
+"@understudy/connector": minor
+---
+
+Bind every snapshot and accessibility ref to the exact browser target that
+produced it.
+
+- `@understudy/protocol`: `snapshot_result` and `screenshot_result` now require
+  the attached CDP `tabId` and main-frame `url`. Accessibility refs remain
+  opaque, but are now namespaced to the extension session, CDP attachment, and
+  snapshot generation so a ref cannot resolve against a replacement browser
+  connection or tab.
+- `@understudy/connector`: snapshot reads expose `target: { tabId, url }`.
+  Driver failures, including expected-tab mismatches and pages that change
+  during capture, return structured `{ ok: false, error }` output with the
+  original reason.
+
+This is a breaking wire change. Upgrade and deploy the protocol, service,
+extension, and connector together. Protocol 0.6 rejects snapshot events from
+older extensions because they lack the required target fields, and connector
+0.4 requires protocol 0.6.

README.md

@@ -28,7 +28,7 @@ drive it over HTTP (Topology 1).
   no LLM and embeds no agent framework — the brain and governance (breakwater/flowsafe) live in
   the consumers. See its README.
 
-M4 status: `@understudy/protocol@0.5.0` and `@understudy/connector@0.3.0` are
+M4 status: `@understudy/protocol@0.6.0` and `@understudy/connector@0.4.0` are
 published on npm. Metamind contains the cross-repository Mastra workflow,
 flowsafe approval gate, breakwater browser connectors, and attended runbook.
 The remaining M4 step is running that proof with a connected Chromium

apps/backend/README.md

@@ -35,7 +35,7 @@ route (internally the coordinator still rejects with the `src/coordinator.ts`
 prefix constants; `SessionAgent.dispatchFailure` maps them in-isolate). The
 route maps reasons to statuses:
 **503** `{error: "extension not connected"}` when the session has no live,
-onConnect-authorized extension socket — the gate consults that delivery
+authoritative onConnect-authorized extension socket — the gate consults that delivery
 predicate directly (not the persisted `status` scalar), fails fast instead of
 burning the 30s timeout, and answers non-2xx deliberately: a 200 `ok:false`
 Event would be cached by a consumer's idempotency store and replayed after
@@ -86,17 +86,22 @@ all.
 The in-DO gate remains as defense in depth for any path that reaches the DO
 without that router: `onConnect` runs the same async checks
 (`verifyExtensionToken` + `scopeSession`) before marking a connection
-`connection.setState({ authorized: true })`, closing with 1008 otherwise.
+`connection.setState({ authorized: true })`, closing with 1008 otherwise. Once
+authenticated, the newest socket becomes the session's sole authority through
+persisted `SessionState.activeConnectionId`; prior authorized sockets are
+demoted and closed with 4001 (`"replaced by newer extension connection"`).
 The Agents SDK accepts a socket — and admits it to the connection set
 `getConnections()` returns — before that async check resolves, so an
 unauthenticated or wrong-tenant socket could sit in the connection set
 during that window. Four things close this gap:
 
-- `sendToExtension` (the coordinator's outbound path) iterates
-  `getConnections()` but filters to `isAuthorizedConnection`, so a command is
-  never written to a socket still pending auth.
-- `onMessage` returns immediately for any connection that isn't yet authorized,
-  so an unauthenticated socket cannot inject events.
+- `sendToExtension` (the coordinator's outbound path) resolves exactly the
+  authorized socket named by `activeConnectionId`, so a command is never
+  broadcast, sent to a socket still pending auth, or duplicated during a
+  reconnect overlap.
+- `onMessage` returns immediately unless its sender is that same authoritative,
+  authorized socket, so neither an unauthenticated nor a replaced socket can
+  inject events or resolve another socket's command.
 - `shouldSendProtocolMessages` returns `false` unconditionally, suppressing the
   SDK's own connect-time protocol frames — the extension speaks only the
   `@understudy/protocol` wire shape and already discards anything else, so this
@@ -106,6 +111,13 @@ during that window. Four things close this gap:
   any accepted connection, including one still awaiting auth, and this DO's state
   is server-driven only.
 
+Persisted sessions created before `activeConnectionId` existed migrate lazily
+only when exactly one authorized socket is live. Multiple legacy candidates are
+ambiguous and fail closed until a newly authenticated socket claims authority;
+the service never falls back to the former broadcast behavior. Closing a
+replaced socket cannot detach its replacement because `onClose` clears status
+only when the closing connection owns the persisted authority.
+
 ## Design decisions
 
 - **Per-session DO, not per-user**: a user can have multiple concurrent
@@ -232,6 +244,10 @@ during that window. Four things close this gap:
 - One Durable Object per `sessionId`; a sessionId whose embedded tenant disagrees
   with the authenticated caller is refused with 404, never 403 — on the /v1
   API and on the agent WS/HTTP path alike.
+- Exactly one authenticated extension socket is authoritative per session.
+  Commands and Events use only its persisted `activeConnectionId`; a newer
+  authenticated socket atomically replaces and closes prior sockets, and a
+  late predecessor close cannot detach the replacement.
 - A mid-command DO hibernation cannot happen (see above); an interrupting
   shutdown/restart is bounded by the per-command timeout, and the persisted
   awaiting-marker reconciles any orphaned late result rather than mis-resolving it.

apps/backend/scripts/stub-consumer.mjs

@@ -17,12 +17,13 @@
  *   3. Load the M2 extension (apps/extension) in a real, logged-in
  *      Chromium and connect it to that printed WS URL.
  *   4. Press Enter in this terminal once the extension shows connected.
- *   5. Watch snapshot -> type -> click -> fill_secret drive the page; each
+ *   5. Confirm the snapshot's tabId and URL before the script uses its refs.
+ *   6. Watch snapshot -> type -> click -> fill_secret drive the page; each
  *      returned Event is printed as it comes back.
  *
- * Without the extension connected, every command below will time out (the
- * service's per-command timeout - 30s by default) because nothing ever
- * answers the forwarded command. The `POST /v1/sessions` step (through
+ * Without an authoritative extension connected, every command below fails
+ * fast with HTTP 503 (`extension not connected`) instead of waiting for the
+ * service's per-command timeout. The `POST /v1/sessions` step (through
  * printing the WS URL) is verifiable standalone, with no extension attached.
  *
  * `type` and `click` target the first ref found in the snapshot's a11y tree
@@ -102,6 +103,33 @@ function findFirstRef(nodes) {
   return undefined;
 }
 
+function snapshotTarget(event) {
+  if (event.type !== "snapshot_result") {
+    throw new Error(`snapshot failed: ${JSON.stringify(event)}`);
+  }
+  if (!Number.isInteger(event.tabId) || event.tabId < 0) {
+    throw new Error(`snapshot returned an invalid tabId: ${JSON.stringify(event.tabId)}`);
+  }
+  if (typeof event.url !== "string" || event.url.length === 0) {
+    throw new Error(`snapshot returned an invalid URL: ${JSON.stringify(event.url)}`);
+  }
+  return { tabId: event.tabId, url: event.url };
+}
+
+async function confirmSnapshotTarget(target) {
+  const rl = createInterface({ input: process.stdin, output: process.stdout });
+  try {
+    const answer = await rl.question(
+      `Confirm snapshot target tabId=${target.tabId}, url=${JSON.stringify(target.url)} [y/N]: `,
+    );
+    if (!["y", "yes"].includes(answer.trim().toLowerCase())) {
+      throw new Error("snapshot target was not confirmed; refusing to use its refs");
+    }
+  } finally {
+    rl.close();
+  }
+}
+
 async function main() {
   console.log("understudy stub consumer - M3 runbook harness");
   console.log(`BASE_URL=${BASE_URL}`);
@@ -124,7 +152,8 @@ async function main() {
     commandId: "stub-1",
     mode: "a11y",
   });
-  const ref = snapshotEvent.type === "snapshot_result" ? findFirstRef(snapshotEvent.tree) : undefined;
+  await confirmSnapshotTarget(snapshotTarget(snapshotEvent));
+  const ref = findFirstRef(snapshotEvent.tree);
   if (!ref) {
     console.log("(no ref found in the snapshot - type/click below will return ok:false)");
   }

apps/backend/src/coordinator-cf.ts

@@ -21,15 +21,16 @@ const DEFAULT_TIMEOUT_MS = 30_000;
  * imports session.ts or `agents` directly.
  */
 export interface CoordinatorHost {
-  /** Writes a JSON frame to the live extension WebSocket. */
+  /** Writes a JSON frame to the session's one authoritative extension WebSocket. */
   sendToExtension(payload: string): void;
   /**
-   * The delivery predicate: does a live, onConnect-authorized extension
-   * socket exist right now? This is the exact precondition sendToExtension
-   * relies on - NOT the persisted SessionState.status scalar, which is a
-   * last-writer-wins echo maintained by lifecycle hooks (onClose guards the
-   * known stamping races, but the scalar is still eventually-consistent
-   * bookkeeping, not the delivery truth).
+   * The delivery predicate: does the persisted authority identify a live,
+   * onConnect-authorized extension socket right now? This is the exact
+   * precondition sendToExtension relies on - NOT the persisted
+   * SessionState.status scalar, which is a last-writer-wins echo maintained
+   * by lifecycle hooks (onClose guards the known stamping races, but the
+   * scalar is still eventually-consistent bookkeeping, not the delivery
+   * truth).
    */
   hasAuthorizedConnection(): boolean;
   /** Reads the persisted awaiting-commandId marker (SessionState.awaitingCommandIds). */
@@ -112,7 +113,18 @@ export class CfSessionCoordinator implements SessionCoordinator {
       // log, by construction.
       console.log("coordinator.send", { commandId: cmd.commandId, type: cmd.type });
 
-      this.host.sendToExtension(JSON.stringify(cmd));
+      try {
+        this.host.sendToExtension(JSON.stringify(cmd));
+      } catch {
+        clearTimeout(timer);
+        this.pending.delete(cmd.commandId);
+        this.dropAwaiting(cmd.commandId);
+        reject(
+          new Error(
+            `${SESSION_NOT_CONNECTED}: authoritative extension connection unavailable during send`,
+          ),
+        );
+      }
     });
   }
 

apps/backend/src/session.ts

@@ -33,6 +33,7 @@ export class SessionAgent extends Agent<Env, SessionState> {
     generation: 0,
     awaitingCommandIds: [],
     status: "pending",
+    activeConnectionId: null,
     completedWrites: [],
     dialogs: [],
   };
@@ -42,16 +43,12 @@ export class SessionAgent extends Agent<Env, SessionState> {
   constructor(ctx: AgentContext, env: Env) {
     super(ctx, env);
     this.coordinator = new CfSessionCoordinator({
-      // getConnections() is hibernation-safe like broadcast(), but filtered to
-      // connections onConnect has marked authorized: the SDK accepts a socket
-      // (and admits it to the connection set) before onConnect's async auth
-      // check resolves, so an unverified or wrong-tenant socket can sit here
-      // during that gap - sending to it directly would hand it a plaintext
-      // command.
       sendToExtension: (payload) => {
-        for (const connection of this.getConnections()) {
-          if (this.isAuthorizedConnection(connection)) connection.send(payload);
+        const connection = this.authoritativeConnection();
+        if (connection === undefined) {
+          throw new Error("authoritative extension connection disappeared before send");
         }
+        connection.send(payload);
       },
       hasAuthorizedConnection: () => this.hasAuthorizedConnection(),
       getAwaitingCommandIds: () => this.state.awaitingCommandIds,
@@ -72,8 +69,28 @@ export class SessionAgent extends Agent<Env, SessionState> {
       connection.close(1008, "tenant mismatch");
       return;
     }
+    this.makeConnectionAuthoritative(connection);
+  }
+
+  private makeConnectionAuthoritative(connection: Connection): void {
     connection.setState({ authorized: true });
-    this.setState({ ...this.state, status: "connected" });
+    this.setState({
+      ...this.state,
+      activeConnectionId: connection.id,
+      status: "connected",
+    });
+
+    for (const previous of this.getConnections()) {
+      if (previous.id === connection.id || !this.isAuthorizedConnection(previous)) continue;
+      previous.setState({ authorized: false });
+      try {
+        previous.close(4001, "replaced by newer extension connection");
+      } catch {
+        // Authority already moved and the predecessor is demoted. A socket
+        // that raced to CLOSED must not make the successful replacement's
+        // onConnect reject.
+      }
+    }
   }
 
   /**
@@ -105,7 +122,7 @@ export class SessionAgent extends Agent<Env, SessionState> {
   }
 
   async onMessage(connection: Connection, message: WSMessage): Promise<void> {
-    if (!this.isAuthorizedConnection(connection)) return;
+    if (!this.isAuthoritativeConnection(connection)) return;
     if (typeof message !== "string") return;
 
     let parsed: unknown;
@@ -147,19 +164,24 @@ export class SessionAgent extends Agent<Env, SessionState> {
   }
 
   async onClose(connection: Connection, code: number, reason: string, wasClean: boolean): Promise<void> {
-    // A socket that never passed onConnect's auth check never contributed
-    // to the session's status, so its close must not change it either - a
-    // rejected/never-authorized socket closing on a fresh session would
-    // otherwise stamp "pending" over with "detached".
     if (!this.isAuthorizedConnection(connection)) return;
-    // Only the LAST authorized socket's close detaches the session: a late
-    // close event from a replaced socket must not stamp "detached" over a
-    // healthy reconnect (the closing connection is excluded explicitly, in
-    // case the SDK has not yet reaped it from the connection set).
-    const stillLive = [...this.getConnections()].some(
-      (c) => c !== connection && this.isAuthorizedConnection(c),
-    );
-    if (!stillLive) this.setState({ ...this.state, status: "detached" });
+
+    const activeConnectionId = this.persistedActiveConnectionId();
+    if (activeConnectionId === undefined) {
+      const anotherAuthorizedConnection = [...this.getConnections()].some(
+        (candidate) =>
+          candidate.id !== connection.id && this.isAuthorizedConnection(candidate),
+      );
+      if (anotherAuthorizedConnection) return;
+    } else if (activeConnectionId !== connection.id) {
+      return;
+    }
+
+    this.setState({
+      ...this.state,
+      activeConnectionId: null,
+      status: "detached",
+    });
   }
 
   async dispatch(command: Command, dryRun?: boolean): Promise<DispatchOutcome> {
@@ -426,14 +448,56 @@ export class SessionAgent extends Agent<Env, SessionState> {
     return (connection.state as { authorized?: boolean } | null)?.authorized === true;
   }
 
-  // The delivery predicate the coordinator's fail-fast gate consults: the
-  // same precondition sendToExtension relies on, NOT the persisted status
-  // scalar (onClose guards the stamping races, but the scalar remains an
-  // eventually-consistent echo, not the delivery truth).
   private hasAuthorizedConnection(): boolean {
-    for (const connection of this.getConnections()) {
-      if (this.isAuthorizedConnection(connection)) return true;
+    return this.authoritativeConnection() !== undefined;
+  }
+
+  private isAuthoritativeConnection(connection: Connection): boolean {
+    if (!this.isAuthorizedConnection(connection)) return false;
+
+    const activeConnectionId = this.persistedActiveConnectionId();
+    if (activeConnectionId !== undefined) {
+      return activeConnectionId !== null && connection.id === activeConnectionId;
     }
-    return false;
+
+    return this.authoritativeConnection()?.id === connection.id;
+  }
+
+  /**
+   * Returns the one live socket Commands may be sent to. A state persisted
+   * before activeConnectionId existed is migrated only when its connection
+   * set has exactly one authorized socket; zero or multiple candidates fail
+   * closed rather than reviving the old broadcast behavior.
+   */
+  private authoritativeConnection(): Connection | undefined {
+    const activeConnectionId = this.persistedActiveConnectionId();
+    if (activeConnectionId === null) return undefined;
+
+    const authorized = [...this.getConnections()].filter((connection) =>
+      this.isAuthorizedConnection(connection),
+    );
+    if (activeConnectionId !== undefined) {
+      return authorized.find((connection) => connection.id === activeConnectionId);
+    }
+    if (authorized.length !== 1) return undefined;
+
+    const [connection] = authorized;
+    if (connection === undefined) return undefined;
+    this.setState({
+      ...this.state,
+      activeConnectionId: connection.id,
+      status: "connected",
+    });
+    return connection;
+  }
+
+  // Persisted before this field existed, a session's state can lack it;
+  // initialState only seeds brand-new DOs.
+  private persistedActiveConnectionId(): string | null | undefined {
+    return (
+      this.state as SessionState & {
+        activeConnectionId?: string | null;
+      }
+    ).activeConnectionId;
   }
 }

apps/backend/src/types.ts

@@ -90,6 +90,15 @@ export interface SessionState {
   generation: number;
   awaitingCommandIds: string[];
   status: SessionStatus;
+  /**
+   * The one authenticated extension connection allowed to receive Commands
+   * and submit Events. `null` means no authoritative connection. Sessions
+   * persisted before this field existed can lack it at runtime; session.ts
+   * migrates that legacy shape only when exactly one authorized connection
+   * exists, so an ambiguous multi-connection state never falls back to
+   * broadcasting.
+   */
+  activeConnectionId: string | null;
   /**
    * Completed WRITE commands' Events, oldest first, capped in session.ts -
    * the service half of the idempotent-retry contract. A consumer retrying