From 0198aece3b2cce0dab267752321a9a4939e3b95f Mon Sep 17 00:00:00 2001 From: Wolfgang Schoenberger <221313372+wolfiesch@users.noreply.github.com> Date: Sat, 19 Sep 2026 00:32:53 -0700 Subject: [PATCH 1/2] Preserve task continuity across transient reconnects --- CHANGELOG.md | 1 + host-rs/crates/agenttab-host/src/journal.rs | 59 ++++++- packages/omp/test/extension.test.ts | 59 +++++++ packages/sdk-python/agenttab/client.py | 32 +++- packages/sdk-python/tests/test_client.py | 72 ++++++++ packages/sdk-typescript/src/index.ts | 123 +++++++++----- packages/sdk-typescript/test/client.test.ts | 176 +++++++++++++++++++- 7 files changed, 469 insertions(+), 53 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index d5f08a2..aea6ad3 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,7 @@ ## Unreleased - 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. - 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`. diff --git a/host-rs/crates/agenttab-host/src/journal.rs b/host-rs/crates/agenttab-host/src/journal.rs index 124376c..3a9d6d2 100644 --- a/host-rs/crates/agenttab-host/src/journal.rs +++ b/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 = 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 }) + )); + } } diff --git a/packages/omp/test/extension.test.ts b/packages/omp/test/extension.test.ts index b1ee1bb..8eb0a82 100644 --- a/packages/omp/test/extension.test.ts +++ b/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> = []; + 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) => 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"); diff --git a/packages/sdk-python/agenttab/client.py b/packages/sdk-python/agenttab/client.py index cf05311..49bce03 100644 --- a/packages/sdk-python/agenttab/client.py +++ b/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, diff --git a/packages/sdk-python/tests/test_client.py b/packages/sdk-python/tests/test_client.py index 2c47e51..89f6b27 100644 --- a/packages/sdk-python/tests/test_client.py +++ b/packages/sdk-python/tests/test_client.py @@ -1071,5 +1071,77 @@ 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.bind(endpoint) + server.listen(1) + try: + connection, _ = server.accept() + 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 = 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")) + client.close() + worker.join(timeout=2) + + 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.bind(endpoint) + server.listen(5) + connect_attempts = 0 + + def serve() -> None: + nonlocal connect_attempts + try: + while True: + connection, _ = server.accept() + try: + _ = read_frame(connection) + connect_attempts += 1 + # Reset/close connection immediately without sending ACK + connection.close() + except Exception: + connection.close() + except Exception: + pass + + worker = threading.Thread(target=serve) + worker.daemon = True + worker.start() + with self.assertRaises(Exception): + AgentTabClient.connect( + endpoint=endpoint, + resume_capability="a" * 32, + connect_timeout=0.5, + ) + server.close() + worker.join(timeout=2) + self.assertEqual(connect_attempts, 1) if __name__ == "__main__": unittest.main() diff --git a/packages/sdk-typescript/src/index.ts b/packages/sdk-typescript/src/index.ts index afc8ca3..a11f8fd 100644 --- a/packages/sdk-typescript/src/index.ts +++ b/packages/sdk-typescript/src/index.ts @@ -581,52 +581,87 @@ async function negotiateConnection( conversationId: string | undefined, resumeCapability: string | undefined, ): Promise { - const socket = createConnection(endpoint); - const decoder = new FrameDecoder(); - const connected = await new Promise((resolve, reject) => { - const timer = setTimeout(() => { - socket.destroy(); - reject(new Error(`Timed out after ${timeoutMs} ms connecting to AgentTab at ${endpoint}`)); - }, timeoutMs); - const closed = () => fail(new Error("AgentTab closed during connection negotiation")); - const fail = (error: Error) => { - clearTimeout(timer); - socket.off("error", fail); - socket.off("close", closed); - reject(error); - }; - const accept = (value: ConnectionAck) => { - clearTimeout(timer); - socket.off("error", fail); - socket.off("close", closed); - socket.removeAllListeners("data"); - resolve(value); - }; - socket.once("error", fail); - socket.once("close", closed); - socket.on("data", (chunk) => { - try { - for (const value of decoder.push(chunk)) { - if (!isConnectionAck(value)) throw new Error("AgentTab sent a response before connection negotiation completed"); - accept(value); - return; + const deadline = Date.now() + timeoutMs; + while (true) { + const remainingMs = deadline - Date.now(); + if (remainingMs <= 0) { + throw new Error(`Timed out after ${timeoutMs} ms connecting to AgentTab at ${endpoint}`); + } + let handshakeWritten = false; + let activeSocket: Socket | undefined; + try { + return await new Promise((resolve, reject) => { + const socket = createConnection(endpoint); + activeSocket = socket; + const decoder = new FrameDecoder(); + const timer = setTimeout(() => { + cleanup(); + socket.destroy(); + reject(new Error(`Timed out after ${timeoutMs} ms connecting to AgentTab at ${endpoint}`)); + }, remainingMs); + const cleanup = () => { + clearTimeout(timer); + socket.off("error", onError); + socket.off("close", onClose); + socket.off("data", onData); + socket.off("connect", onConnect); + }; + const fail = (error: Error) => { + cleanup(); + socket.destroy(); + reject(error); + }; + const onError = (error: Error) => fail(error); + const onClose = () => fail(new Error("AgentTab closed during connection negotiation")); + const onData = (chunk: Buffer) => { + try { + for (const value of decoder.push(chunk)) { + if (!isConnectionAck(value)) throw new Error("AgentTab sent a response before connection negotiation completed"); + cleanup(); + resolve({ socket, decoder, connected: value }); + return; + } + } catch (error) { + fail(error instanceof Error ? error : new Error(String(error))); + } + }; + const onConnect = () => { + handshakeWritten = true; + socket.write(encodeFrame({ + protocol: RPC_PROTOCOL, + version: RPC_VERSION, + kind: "connect", + ...(conversationId ? { conversation_id: conversationId } : {}), + ...(resumeCapability ? { resume_capability: resumeCapability } : {}), + })); + }; + socket.once("error", onError); + socket.once("close", onClose); + socket.on("data", onData); + socket.once("connect", onConnect); + }); + } catch (error) { + if (activeSocket) { + activeSocket.destroy(); + } + // Retry ONLY pre-connect transport failure before the handshake payload is written. + // Once handshake write is dispatched, an ambiguous server ACK must never replay. + const isTransient = !handshakeWritten && isRecord(error) && ( + error.code === "ECONNREFUSED" || + error.code === "ENOENT" + ); + if (isTransient && deadline - Date.now() > 0) { + const delay = Math.min(50, Math.max(0, deadline - Date.now())); + if (delay > 0) { + await new Promise((resolve) => setTimeout(resolve, delay)); + } + if (deadline - Date.now() > 0) { + continue; } - } catch (error) { - socket.destroy(); - fail(error instanceof Error ? error : new Error(String(error))); } - }); - socket.once("connect", () => { - socket.write(encodeFrame({ - protocol: RPC_PROTOCOL, - version: RPC_VERSION, - kind: "connect", - ...(conversationId ? { conversation_id: conversationId } : {}), - ...(resumeCapability ? { resume_capability: resumeCapability } : {}), - })); - }); - }); - return { socket, decoder, connected }; + throw error; + } + } } async function confirmResumeCapability( diff --git a/packages/sdk-typescript/test/client.test.ts b/packages/sdk-typescript/test/client.test.ts index e5136ca..6d314dd 100644 --- a/packages/sdk-typescript/test/client.test.ts +++ b/packages/sdk-typescript/test/client.test.ts @@ -2,7 +2,7 @@ import { afterEach, describe, expect, test } from "bun:test"; import { mkdtempSync, rmSync, statSync } from "node:fs"; import { tmpdir } from "node:os"; import { join } from "node:path"; -import { createServer, type Server, type Socket } from "node:net"; +import { createServer, Socket, type Server } from "node:net"; import { AgentTabClient, AgentTabError, @@ -885,3 +885,177 @@ test("does not confirm a replacement when durable staging fails", async () => { await expect(AgentTabClient.connect({ endpoint, capabilityStore: store })).rejects.toThrow("fsync failed"); expect(confirmations).toBe(0); }); + +test("retries transient connection errors during host reload and resumes task", async () => { + const root = mkdtempSync(join(tmpdir(), "agenttab-sdk-")); + roots.push(root); + const endpoint = join(root, "agenttab.sock"); + const store = { + path: "memory", + load: async () => "a".repeat(32), + loadPending: async () => undefined, + save: async () => undefined, + prepareReplacement: async () => undefined, + activateReplacement: async () => undefined, + clear: async () => undefined, + }; + + const events: string[] = []; + const server = createServer((socket) => { + socket.on("error", () => {}); + const decoder = new FrameDecoder(); + socket.on("data", (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + if (value.kind === "connect") { + events.push("handshake"); + socket.write(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "connected", + connection_id: "018f22b2-4126-7c1a-8c31-3f45a783da99", + resumed: true, + task_id: "018f22b2-4126-7c1a-8c31-3f45a783da44", + resume_capability: "b".repeat(32), + state: "ready", + }, 1024 * 1024)); + } else if (value.kind === "resume_confirm") { + socket.write(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "resume_confirmed", + connection_id: "018f22b2-4126-7c1a-8c31-3f45a783da99", + }, 1024 * 1024)); + } + } + }); + }); + servers.push(server); + + let firstFailure: (Error & { code?: string }) | undefined; + let connectionAttempts = 0; + const originalConnect = Socket.prototype.connect; + let client: AgentTabClient | undefined; + + try { + Socket.prototype.connect = function (this: Socket, ...args: unknown[]) { + let first: unknown = args[0]; + if (Array.isArray(first)) { + first = first[0]; + } + let targetPath: string | undefined; + if (typeof first === "string") { + targetPath = first; + } else if (first && typeof first === "object" && "path" in first) { + const candidate = first.path; + if (typeof candidate === "string") { + targetPath = candidate; + } + } + + if (targetPath === endpoint) { + connectionAttempts += 1; + if (connectionAttempts === 1) { + this.once("error", (err: Error & { code?: string }) => { + firstFailure = err; + const code = "code" in err && typeof err.code === "string" ? err.code : undefined; + events.push(`failed:${code ?? err.message}`); + server.listen(endpoint); + }); + } + } + // Re-invoke original connect with preserved arguments + const connectMethod = originalConnect as (this: Socket, ...connectArgs: unknown[]) => Socket; + return connectMethod.apply(this, args); + }; + + client = await AgentTabClient.connect({ + endpoint, + connectTimeoutMs: 1500, + capabilityStore: store, + }); + + let firstFailureCode: string | undefined; + if (firstFailure && typeof firstFailure === "object" && "code" in firstFailure) { + const candidate = firstFailure.code; + if (typeof candidate === "string") { + firstFailureCode = candidate; + } + } + + expect(firstFailure).toBeDefined(); + expect(["ENOENT", "ECONNREFUSED"]).toContain(firstFailureCode ?? ""); + expect(connectionAttempts).toBeGreaterThanOrEqual(2); + expect(events[0]).toMatch(/^failed:(ENOENT|ECONNREFUSED)$/); + expect(events).toEqual([`failed:${firstFailureCode}`, "handshake"]); + expect(client.connection.resumed).toBe(true); + expect(client.connection.task_id).toBe("018f22b2-4126-7c1a-8c31-3f45a783da44"); + expect(client.closed).toBe(false); + } finally { + Socket.prototype.connect = originalConnect; + client?.close(); + } +}); + +test("refuses unauthenticated fallback when stored resume capability is rejected by host", async () => { + const endpoint = await listen((socket) => { + socket.on("error", () => {}); + const decoder = new FrameDecoder(); + socket.on("data", (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + if (value.kind === "connect") { + socket.end(encodeFrame({ + protocol: "agenttab.rpc", + version: 1, + kind: "connected", + connection_id: "018f22b2-4126-7c1a-8c31-3f45a783da99", + resumed: false, + state: "ready", + }, 1024 * 1024)); + } + } + }); + }); + const store = { + path: "memory", + load: async () => "a".repeat(32), + loadPending: async () => undefined, + save: async () => undefined, + prepareReplacement: async () => undefined, + activateReplacement: async () => undefined, + clear: async () => undefined, + }; + await expect(AgentTabClient.connect({ endpoint, capabilityStore: store })).rejects.toThrow( + "AgentTab rejected the stored resume capability", + ); +}); + +test("does not replay connect after reset or close once handshake write is dispatched", async () => { + let connectAttempts = 0; + const endpoint = await listen((socket) => { + socket.on("error", () => {}); + const decoder = new FrameDecoder(); + socket.on("data", (chunk) => { + for (const value of decoder.push(chunk) as Array>) { + if (value.kind === "connect") { + connectAttempts += 1; + socket.destroy(); + } + } + }); + }); + const store = { + path: "memory", + load: async () => "a".repeat(32), + loadPending: async () => undefined, + save: async () => undefined, + prepareReplacement: async () => undefined, + activateReplacement: async () => undefined, + clear: async () => undefined, + }; + await expect(AgentTabClient.connect({ + endpoint, + connectTimeoutMs: 500, + capabilityStore: store, + })).rejects.toThrow(); + expect(connectAttempts).toBe(1); +}); From d6e1417f5c3b53315228d589e1ca4edb053f9530 Mon Sep 17 00:00:00 2001 From: Wolfgang Schoenberger <221313372+wolfiesch@users.noreply.github.com> Date: Sat, 19 Sep 2026 00:43:37 -0700 Subject: [PATCH 2/2] Bound regression server shutdown and assert worker exit --- packages/sdk-python/tests/test_client.py | 58 ++++++++++++++---------- 1 file changed, 35 insertions(+), 23 deletions(-) diff --git a/packages/sdk-python/tests/test_client.py b/packages/sdk-python/tests/test_client.py index 89f6b27..0dcf5a9 100644 --- a/packages/sdk-python/tests/test_client.py +++ b/packages/sdk-python/tests/test_client.py @@ -1078,10 +1078,12 @@ def test_connect_retries_transient_connection_refusal_within_deadline(self) -> N 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({ @@ -1099,12 +1101,17 @@ def delayed_serve() -> None: worker = threading.Thread(target=delayed_serve) worker.start() - 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")) - client.close() - worker.join(timeout=2) + 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, @@ -1112,36 +1119,41 @@ def test_does_not_replay_connect_after_reset_or_close_once_handshake_dispatched( 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 True: - connection, _ = server.accept() + 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 - # Reset/close connection immediately without sending ACK - connection.close() - except Exception: - connection.close() - except Exception: - pass + # Close without ACK; a dispatched handshake must not replay. + finally: + server.close() worker = threading.Thread(target=serve) - worker.daemon = True worker.start() - with self.assertRaises(Exception): - AgentTabClient.connect( - endpoint=endpoint, - resume_capability="a" * 32, - connect_timeout=0.5, - ) - server.close() - worker.join(timeout=2) + 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()