diff --git a/crates/cli/tests/agent_target.rs b/crates/cli/tests/agent_target.rs index 0248a2ef9..647d8645a 100644 --- a/crates/cli/tests/agent_target.rs +++ b/crates/cli/tests/agent_target.rs @@ -331,3 +331,48 @@ fn agent_target_replays_pending_targets_to_a_late_overlay() { let (_, verdict) = held.join().unwrap(); assert_eq!(verdict["sessionId"], serde_json::json!("aabbccdd")); } + +#[test] +fn agent_target_reconnect_keeps_the_overlays_lease_and_word() { + let s = Server::start("reconnect"); + let mut a = Overlay::connect(s.port, &s.token, "tab-a"); + let mut b = Overlay::connect(s.port, &s.token, "tab-b"); + a.next(|m| m["type"] == "connected"); + b.next(|m| m["type"] == "connected"); + let held = s.hold(serde_json::json!({})); + let target_id = a.next(|m| m["type"] == "agent_target")["targetId"].as_str().unwrap().to_string(); + assert_eq!(s.claim(&target_id, "tab-a", true)["granted"], serde_json::json!(true)); + // An EventSource reconnect: the same page opens a replacement connection + // under its clientId before the old one is seen to close. + let mut a2 = Overlay::connect(s.port, &s.token, "tab-a"); + a2.next(|m| m["type"] == "agent_target"); + drop(a); + std::thread::sleep(Duration::from_millis(150)); + // The old connection's close must not hand tab-a's lease to anyone: + // tab-b stays denied, tab-a renews as the holder. + assert_eq!(s.claim(&target_id, "tab-b", true)["granted"], serde_json::json!(false), "the lease survived the reconnect"); + assert_eq!(s.claim(&target_id, "tab-a", true)["granted"], serde_json::json!(true)); + post_json(s.port, "/agent-target-result", serde_json::json!({ "token": s.token, "targetId": target_id, "ok": true, "sessionId": "aabbccdd" })); + let (_, verdict) = held.join().unwrap(); + assert_eq!(verdict["sessionId"], serde_json::json!("aabbccdd")); + let _ = (&mut a2, &mut b); +} + +#[test] +fn agent_target_roll_call_counts_overlays_not_connections() { + let s = Server::start("distinct"); + let mut a = Overlay::connect(s.port, &s.token, "tab-a"); + let mut a2 = Overlay::connect(s.port, &s.token, "tab-a"); + a.next(|m| m["type"] == "connected"); + a2.next(|m| m["type"] == "connected"); + let started = Instant::now(); + let held = s.hold(serde_json::json!({})); + let target_id = a.next(|m| m["type"] == "agent_target")["targetId"].as_str().unwrap().to_string(); + // One overlay behind two connections reports busy once: that completes + // the roll call instead of waiting on a "second" report until timeout. + assert_eq!(s.claim(&target_id, "tab-a", false), serde_json::json!({ "ok": true, "granted": false })); + let (_, verdict) = held.join().unwrap(); + assert_eq!(verdict["error"], serde_json::json!("busy")); + assert!(started.elapsed() < Duration::from_millis(350), "the busy verdict did not wait for the timeout"); + let _ = &mut a2; +} diff --git a/crates/live/src/server_state.rs b/crates/live/src/server_state.rs index c742ce5ef..f2e48d6c9 100644 --- a/crates/live/src/server_state.rs +++ b/crates/live/src/server_state.rs @@ -657,7 +657,12 @@ impl ServerState { /// Remove an SSE client; when none remain arm the exit timer (JS /// `req.on('close')`). A departed overlay's word no longer counts in any - /// agent-target roll call. + /// agent-target roll call. The overlay, not the connection, is the + /// participant: an EventSource reconnect opens a replacement connection + /// under the same page-level clientId before the old one is seen to + /// close, so its word is retired only once no connection carries that + /// id, while every roll call is still re-judged against the connections + /// that remain. pub fn remove_sse_client(&mut self, id: u64) { let before = self.sse_clients.len(); let agent_client_id = self @@ -667,7 +672,11 @@ impl ServerState { .and_then(|c| c.agent_client_id.clone()); self.sse_clients.retain(|c| c.id != id); if before != self.sse_clients.len() { - self.drop_agent_target_client(agent_client_id.as_deref()); + let still_connected = agent_client_id + .as_deref() + .map(|cid| self.sse_clients.iter().any(|c| c.agent_client_id.as_deref() == Some(cid))) + .unwrap_or(false); + self.drop_agent_target_client(if still_connected { None } else { agent_client_id.as_deref() }); if self.sse_clients.is_empty() { self.clear_exit_timer(); self.arm_exit_timer(); @@ -675,6 +684,26 @@ impl ServerState { } } + /// Connected overlays for a roll call: one per distinct clientId, plus + /// every connection that sent none (an older overlay build), so a + /// reconnect's momentary duplicate connection never waits on a second + /// report from the same tab. + pub fn connected_overlay_count(&self) -> usize { + let mut ids: Vec<&str> = Vec::new(); + let mut anonymous = 0; + for c in &self.sse_clients { + match c.agent_client_id.as_deref() { + Some(cid) => { + if !ids.contains(&cid) { + ids.push(cid); + } + } + None => anonymous += 1, + } + } + ids.len() + anonymous + } + // --------------------------------------------------------------------- // Agent-initiated element targeting (the `generate` command) // --------------------------------------------------------------------- @@ -777,7 +806,7 @@ impl ServerState { /// timeout. Judged against the connections of this moment, so it runs /// whenever a report lands and whenever an overlay leaves. pub fn maybe_complete_agent_target_roll_call(&mut self, target_id: &str) { - let connected = self.sse_clients.len(); + let connected = self.connected_overlay_count(); let verdict = self .pending_agent_targets .iter() diff --git a/docs/CLI-CONTRACT.md b/docs/CLI-CONTRACT.md index e90927079..0889331c9 100644 --- a/docs/CLI-CONTRACT.md +++ b/docs/CLI-CONTRACT.md @@ -1468,7 +1468,7 @@ Binds `127.0.0.1:PORT`. CORS: if request has `Origin` and (origin is loopback ht | `GET /design-system.json?token=` | 401 `Unauthorized` | 404 `{present:false}` if neither DESIGN.md nor `.impeccable/design.json`; else `{present:true, hasMd, hasSidecar, mdNewerThanJson, parsed?, parseError?, sidecar?, sidecarError?}` (`parsed` = parseDesignMd output; `sidecarError` = `'Failed to parse .impeccable/design.json: '+msg`) | | `GET /design-system/raw?token=` | 401 | 200 `text/markdown; charset=utf-8` DESIGN.md verbatim; 404 `Not found` | | `GET /source?token=&path=` | 401 | path required and no `..` else 400 `Bad path`; resolved must be inside cwd (relative check, not root itself) else 403 `Forbidden`; 404 `File not found`; 200 `text/html; charset=utf-8` raw file. Used by browser to read source, svelte manifest and `params.json`. | -| `GET /events?token=&clientId=` (SSE) | 401 | `clientId` (optional) is the overlay's per-page-load id; on close the server retires that client's agent-target roll-call report and releases a lease it held, then re-judges each pending roll call against the remaining clients. Headers `text/event-stream`, `Cache-Control: no-cache`, `Connection: keep-alive`; first frame `data: {"type":"connected","hasProjectContext":b,"agentPolling":b,"activeSessions":[…]}\n\n`; `: keepalive\n\n` every 30s; on connect: cancels exit timer and removes queued anonymous `exit` events. On close: if 0 clients, after 8000 ms (still 0) enqueue `{type:'exit'}`. | +| `GET /events?token=&clientId=` (SSE) | 401 | `clientId` (optional) is the overlay's per-page-load id; on close the server retires that client's agent-target roll-call report and releases a lease it held **only when no other connection still carries that id** (an EventSource reconnect opens the replacement before the old connection is seen to close), then re-judges each pending roll call against the remaining overlays (distinct ids, plus connections that sent none). The pending targets are replayed to every connection that opens; the overlay ignores a replay of a target it already handled. Headers `text/event-stream`, `Cache-Control: no-cache`, `Connection: keep-alive`; first frame `data: {"type":"connected","hasProjectContext":b,"agentPolling":b,"activeSessions":[…]}\n\n`; `: keepalive\n\n` every 30s; on connect: cancels exit timer and removes queued anonymous `exit` events. On close: if 0 clients, after 8000 ms (still 0) enqueue `{type:'exit'}`. | | `POST /events` | body JSON `token` mismatch → 401 `{"error":"Unauthorized"}`; invalid JSON → 400 `{"error":"Invalid JSON"}` | see 6.1 | | `GET /stop?token=` | 401 | 200 text `stopping`, then shutdown | | `GET /poll?token=&timeout=&leaseMs=&types=` | 401 `{"error":"Unauthorized"}` | see 6.3 | diff --git a/skill/scripts/live-browser.js b/skill/scripts/live-browser.js index e93a8bba9..9d6a3d1ba 100644 --- a/skill/scripts/live-browser.js +++ b/skill/scripts/live-browser.js @@ -7278,8 +7278,17 @@ } } + // Targets this page already answered (claimed, declined, or acted on). + // The server replays pending targets to every connection that opens, and + // an EventSource reconnect opens one for a page that already heard the + // target, so a replay must not start a second claim or a second Go. + const agentTargetsSeen = []; + function handleAgentTarget(msg) { if (!msg || typeof msg.targetId !== 'string') return; + if (agentTargetsSeen.includes(msg.targetId)) return; + agentTargetsSeen.push(msg.targetId); + if (agentTargetsSeen.length > 100) agentTargetsSeen.shift(); const busy = agentTargetBusyReason(); if (busy) { // Roll call: a busy tab reports itself and never acts. The server diff --git a/tests/live-agent-target.test.mjs b/tests/live-agent-target.test.mjs index 6d0ffcaca..320cf1538 100644 --- a/tests/live-agent-target.test.mjs +++ b/tests/live-agent-target.test.mjs @@ -598,6 +598,49 @@ describe('POST /agent-target', { skip: ENGINE_BIN ? false : ENGINE_MISSING_MESSA } }); + it('keeps a reconnected overlay\'s lease and word when its old connection closes', async () => { + // An EventSource reconnect opens a replacement connection under the same + // page-level clientId before the old one is seen to close. The close + // must retire nothing while the overlay is still connected. + const sleep = (ms) => new Promise((r) => setTimeout(r, ms)); + const tabA = await openSseClient(server, { clientId: 'tab-a' }); + const tabB = await openSseClient(server, { clientId: 'tab-b' }); + let tabA2 = null; + try { + await tabA.next((m) => m.type === 'connected'); + await tabB.next((m) => m.type === 'connected'); + const held = postJson(server, '/agent-target', { + token: server.token, selector: 'h1', action: 'bolder', count: 3, + }); + const pushed = await tabA.next((m) => m.type === 'agent_target'); + const holder = await (await postJson(server, '/agent-target-claim', { + token: server.token, targetId: pushed.targetId, clientId: 'tab-a', eligible: true, + })).json(); + assert.equal(holder.granted, true); + tabA2 = await openSseClient(server, { clientId: 'tab-a' }); + await tabA2.next((m) => m.type === 'agent_target'); + tabA.close(); + await sleep(150); + const rival = await (await postJson(server, '/agent-target-claim', { + token: server.token, targetId: pushed.targetId, clientId: 'tab-b', eligible: true, + })).json(); + assert.equal(rival.granted, false, 'the lease survived the reconnect'); + const renew = await (await postJson(server, '/agent-target-claim', { + token: server.token, targetId: pushed.targetId, clientId: 'tab-a', eligible: true, + })).json(); + assert.equal(renew.granted, true, 'the reconnected overlay still holds it'); + await postJson(server, '/agent-target-result', { + token: server.token, targetId: pushed.targetId, ok: true, matchCount: 1, sessionId: 'aabbccdd', + }); + const verdict = await (await held).json(); + assert.equal(verdict.sessionId, 'aabbccdd'); + } finally { + tabA.close(); + tabB.close(); + if (tabA2) tabA2.close(); + } + }); + it('completes the roll call when the last silent overlay disconnects', async () => { // Tab A reported busy; tab B never answered and then left. Every overlay // still connected has declined, so the verdict is busy now, not at the diff --git a/tests/live-browser-source.test.mjs b/tests/live-browser-source.test.mjs index 66c82868c..b2958d27d 100644 --- a/tests/live-browser-source.test.mjs +++ b/tests/live-browser-source.test.mjs @@ -846,6 +846,11 @@ describe('live-browser source contracts', () => { /\/events\?token=' \+ TOKEN \+ '&clientId=' \+ AGENT_TARGET_CLIENT_ID/, 'the SSE connection must carry the overlay id, so a disconnect retires its roll-call word', ); + assert.match( + SOURCE, + /function handleAgentTarget\(msg\) \{[\s\S]{0,120}?if \(agentTargetsSeen\.includes\(msg\.targetId\)\) return;/, + 'a replayed target this page already handled must not start a second claim or Go', + ); assert.match(helper, /setTimeout\(\(\) => claimAndActOnAgentTarget\(msg\), AGENT_TARGET_RESCUE_RETRY_MS\);/, 'a denied claim on a live request retries until the lease lapses'); assert.match( SOURCE,