Preserve task continuity across transient reconnects

CHANGELOG.md

@@ -5,6 +5,7 @@
 - Added a DOM editable fallback to accessibility snapshots. When Chrome's debugger-backed accessibility tree omits an editable element, for example a framework's pre-hydration composer textarea, full-tree snapshots lead with up to 40 `dom_fallback` textbox nodes minted from DOM query results, scanning past accessibility-covered and unsupported matches until the limit of accepted editables or a bounded inspection budget is reached. Each carries a normal revisioned ref that the existing fill, type, and click paths already accept; password inputs are skipped, partial (`root_ref`) snapshots are unchanged, the `max_nodes` budget and `truncated` flag now cover the combined result, and a fallback capture failure never fails the snapshot itself.
 - Added unique CSS selector targets for Standard `click`, `type`, and `fill`, retaining sensitive-field and Commit checks and executing against the checked backend node. Normalize and truncate selector text in operation displays.
 - Replaced blocking human handoffs with task-owned attention notices. Agents can request, inspect, resolve, or dismiss a notice without popup acknowledgement; open notices do not block ordinary commands or task cleanup. Legacy handoffs migrate without restoring a lock, and stale notice IDs cannot clear newer requests. Explicit Pause, ownership, sensitive-field restrictions, and configured Commit review remain independent.
+- Preserve active task ownership when reconnecting alongside stale tabs from closed or missing tasks. Retry transient endpoint failures within the connection deadline without replaying dispatched handshakes or mutations; rejected resume capabilities still fail closed.
 - Enabled YOLO mode for new and legacy default state so recognizable consequential controls execute in the original `browser_act` call. Turning YOLO mode off restores staged Commit review. Task ownership, origin policy, revision checks, credential isolation, and all other Standard boundaries remain enforced.
 - Preserved existing permissive-mode settings and managed 1Password defaults during the notice cutover, including GUI-host executable discovery and legacy created-tab provenance sanitization.
 - Replaced the Chrome Bridge v1 runtime with the AgentTab 2.0 release candidate: a Rust production host over OS-native local IPC, nine task-scoped Standard methods, explicit resumable capabilities, a developer-only tenth method, TypeScript and Python SDKs, MCP and OMP adapters, a transactional installer, and a minimal extension. Consequential controls now use a two-party Commit flow: `browser_act` stages an exact effect, the popup approves the durable review record without executing it, and the requesting task must consume its private one-use token through `browser_commit`.

host-rs/crates/agenttab-host/src/journal.rs

@@ -446,7 +446,7 @@ impl Journal {
                 )
                 .optional()?;
             if active.is_none() {
-                return Err(JournalError::MissingTask);
+                continue;
             }
             let floor: Option<i64> = transaction
                 .query_row(
@@ -2018,4 +2018,61 @@ mod tests {
             .unwrap();
         assert_eq!(legacy_receipts, 0);
     }
+
+    #[test]
+    fn reconcile_inventory_skips_tabs_for_closed_or_missing_tasks_without_error() {
+        let temp = tempfile::tempdir().unwrap();
+        let journal = open_journal(&temp);
+        let active_task = journal.create_task(Some("conversation")).unwrap();
+        let missing_task_id = Uuid::now_v7();
+
+        let mut tab_active = owned_tab(active_task.task_id, 1);
+        tab_active.tab_id = 7;
+        let mut tab_missing = owned_tab(missing_task_id, 1);
+        tab_missing.tab_id = 8;
+
+        let inventory = vec![tab_active, tab_missing];
+
+        let result = journal.reconcile_inventory(&inventory).unwrap();
+        assert_eq!(result, InventoryReconciliation::Applied);
+
+        // Active task tab is retained and owned
+        assert_eq!(
+            journal
+                .verify_task_tab(active_task.task_id, 7, Some(1))
+                .unwrap(),
+            1
+        );
+        // Missing task tab was skipped, not adopted
+        assert!(matches!(
+            journal.verify_task_tab(active_task.task_id, 8, None),
+            Err(JournalError::TabNotOwned { tab_id: 8 })
+        ));
+    }
+
+    #[test]
+    fn closed_task_refuses_resume_and_retains_tab_denial() {
+        let temp = tempfile::tempdir().unwrap();
+        let journal = open_journal(&temp);
+        let task = journal.create_task(Some("conversation")).unwrap();
+        journal
+            .reconcile_inventory(&[owned_tab(task.task_id, 1)])
+            .unwrap();
+        assert_eq!(
+            journal.verify_task_tab(task.task_id, 7, Some(1)).unwrap(),
+            1
+        );
+
+        journal.close_task(task.task_id).unwrap();
+
+        // Resume is denied on closed task
+        let resume_result = journal.resume_task(&task.resume_capability).unwrap();
+        assert!(resume_result.is_none());
+
+        // Tab is denied
+        assert!(matches!(
+            journal.verify_task_tab(task.task_id, 7, None),
+            Err(JournalError::TabNotOwned { tab_id: 7 })
+        ));
+    }
 }

packages/omp/test/extension.test.ts

@@ -843,6 +843,65 @@ test("OMP shares one in-flight Core connection across concurrent first calls", a
   expect(connections).toBe(1);
 });
 
+test("OMP reconnects and resumes task after transport failure on subsequent tool call", async () => {
+  const tools: Array<Record<string, unknown>> = [];
+  let client1Closed = false;
+  const client1 = {
+    connection: { task_id: "task-resumed-42" },
+    get closed() {
+      return client1Closed;
+    },
+    request: async () => ({
+      ok: true,
+      outcome: "completed",
+      request_id: "req-1",
+      result: { tabs: [{ tab_id: 1, url: "https://example.com" }] },
+    }),
+  } as unknown as AgentTabClient;
+
+  const client2 = {
+    connection: { task_id: "task-resumed-42" },
+    get closed() {
+      return false;
+    },
+    request: async () => ({
+      ok: true,
+      outcome: "completed",
+      request_id: "req-2",
+      result: { mode: "text", content: "hello resumed", page_revision: 1 },
+    }),
+  } as unknown as AgentTabClient;
+
+  let connectCount = 0;
+  makeExtension(async () => {
+    connectCount += 1;
+    return connectCount === 1 ? client1 : client2;
+  })({
+    zod,
+    registerTool: (tool: Record<string, unknown>) => tools.push(tool),
+  } as unknown as AgentApi);
+
+  const tabsTool = tools.find((t) => t.name === "browser_tabs");
+  const snapshotTool = tools.find((t) => t.name === "browser_snapshot");
+
+  const res1 = await executeTool(tabsTool, {}, "call-1");
+  expect(res1.details).toMatchObject({
+    _agenttab: { outcome: "completed", task_id: "task-resumed-42" },
+  });
+  expect(connectCount).toBe(1);
+
+  // Simulate transport failure on client1
+  client1Closed = true;
+
+  // Next tool call detects closed client, triggers reconnect, resumes task
+  const res2 = await executeTool(snapshotTool, { tab_id: 1, mode: "text" }, "call-2");
+  expect(res2.details).toMatchObject({
+    mode: "text",
+    _agenttab: { outcome: "completed", task_id: "task-resumed-42" },
+  });
+  expect(connectCount).toBe(2);
+});
+
 test("OMP tools expose compact and expanded custom renderers", () => {
   for (const tool of register(false).tools) {
     expect(typeof tool.renderCall).toBe("function");

packages/sdk-python/agenttab/client.py

@@ -680,13 +680,31 @@ def _negotiate_connection(
         connect_timeout: float,
         request_timeout: float,
     ) -> tuple[BinaryIO | socket.socket, JsonObject]:
-        if os.name == "nt":
-            stream: BinaryIO | socket.socket = _open_windows_named_pipe(address)
-        else:
-            unix_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
-            unix_socket.settimeout(connect_timeout)
-            unix_socket.connect(address)
-            stream = unix_socket
+        connect_deadline = time.monotonic() + connect_timeout
+        while True:
+            remaining_connect = connect_deadline - time.monotonic()
+            if remaining_connect <= 0:
+                raise TimeoutError(f"Timed out connecting to AgentTab at {address}")
+            unix_socket: socket.socket | None = None
+            try:
+                if os.name == "nt":
+                    stream: BinaryIO | socket.socket = _open_windows_named_pipe(address)
+                else:
+                    unix_socket = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+                    unix_socket.settimeout(remaining_connect)
+                    unix_socket.connect(address)
+                    stream = unix_socket
+                break
+            except (FileNotFoundError, ConnectionRefusedError):
+                if unix_socket is not None:
+                    unix_socket.close()
+                if time.monotonic() >= connect_deadline:
+                    raise
+                delay = min(0.05, max(0.0, connect_deadline - time.monotonic()))
+                if delay > 0:
+                    time.sleep(delay)
+                if time.monotonic() >= connect_deadline:
+                    raise
         request: JsonObject = {
             "protocol": RPC_PROTOCOL,
             "version": RPC_VERSION,

packages/sdk-python/tests/test_client.py

@@ -1071,5 +1071,89 @@ def serve() -> None:
             worker.join(timeout=2)
             server.close()
 
+    def test_connect_retries_transient_connection_refusal_within_deadline(self) -> None:
+        with tempfile.TemporaryDirectory() as root:
+            endpoint = str(Path(root) / "agenttab.sock")
+
+            def delayed_serve() -> None:
+                time.sleep(0.08)
+                server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+                server.settimeout(1)
+                server.bind(endpoint)
+                server.listen(1)
+                try:
+                    connection, _ = server.accept()
+                    connection.settimeout(1)
+                    try:
+                        _ = read_frame(connection)
+                        connection.sendall(encode_frame({
+                            "protocol": "agenttab.rpc",
+                            "version": 1,
+                            "kind": "connected",
+                            "connection_id": "018f22b2-4126-7c1a-8c31-3f45a783da42",
+                            "resumed": False,
+                            "state": "ready",
+                        }, 1024 * 1024))
+                    finally:
+                        connection.close()
+                finally:
+                    server.close()
+
+            worker = threading.Thread(target=delayed_serve)
+            worker.start()
+            client = None
+            try:
+                client = AgentTabClient.connect(endpoint=endpoint, connect_timeout=1.5)
+                self.assertEqual(client.connection.get("kind"), "connected")
+                self.assertEqual(client.connection.get("state"), "ready")
+                self.assertFalse(client.connection.get("resumed"))
+            finally:
+                if client is not None:
+                    client.close()
+                worker.join(timeout=2)
+                self.assertFalse(worker.is_alive(), "connect retry worker did not stop")
+
+    def test_does_not_replay_connect_after_reset_or_close_once_handshake_dispatched(
+        self,
+    ) -> None:
+        with tempfile.TemporaryDirectory() as root:
+            endpoint = str(Path(root) / "agenttab.sock")
+            server = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
+            server.settimeout(0.05)
+            server.bind(endpoint)
+            server.listen(5)
+            connect_attempts = 0
+            stop = threading.Event()
+
+            def serve() -> None:
+                nonlocal connect_attempts
+                try:
+                    while not stop.is_set():
+                        try:
+                            connection, _ = server.accept()
+                        except socket.timeout:
+                            continue
+                        with connection:
+                            connection.settimeout(0.5)
+                            _ = read_frame(connection)
+                            connect_attempts += 1
+                            # Close without ACK; a dispatched handshake must not replay.
+                finally:
+                    server.close()
+
+            worker = threading.Thread(target=serve)
+            worker.start()
+            try:
+                with self.assertRaises(Exception):
+                    AgentTabClient.connect(
+                        endpoint=endpoint,
+                        resume_capability="a" * 32,
+                        connect_timeout=0.5,
+                    )
+            finally:
+                stop.set()
+                worker.join(timeout=2)
+                self.assertFalse(worker.is_alive(), "no-replay worker did not stop")
+            self.assertEqual(connect_attempts, 1)
 if __name__ == "__main__":
     unittest.main()