From 4335298575a8e3f6a0681e73f31250b07190a34c Mon Sep 17 00:00:00 2001 From: Louis Amaudruz Date: Mon, 31 Aug 2026 14:27:00 +0000 Subject: [PATCH] fix(sessions): close the connection on teardown instead of waiting for the TTL (CHOO-2497) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Deleting a session tore the runtime down locally and told switch-core nothing. The server kept the session's connection, and with it the claim on its room, until the heartbeat sweep collected it. For those seconds the agent's all-scope watcher was dark on that room, so a message arriving in the gap was never delivered and no replacement session was spawned — the user saw a delay before a ping did anything. Add POST /agents/{id}/connection/close and call it from RoomConnection's teardown. Closing releases the room slot at once, handing coverage back to the watcher. The call is idempotent, detached and best-effort: it is logged rather than raised on failure, and the heartbeat sweep remains the backstop. Co-Authored-By: Claude Opus 5 (1M context) --- .../core/switch-rooms/room-connection.test.ts | 54 +++++++++ .../main/core/switch-rooms/room-connection.ts | 47 ++++++++ .../switch_core/bridges/agent/api/handlers.py | 27 +++++ core/switch_core/bridges/agent/api/schemas.py | 7 ++ .../agent/api/test_connection_close.py | 107 ++++++++++++++++++ 5 files changed, 242 insertions(+) create mode 100644 core/tests/switch_core/bridges/agent/api/test_connection_close.py diff --git a/console/apps/switch-console-desktop/src/main/core/switch-rooms/room-connection.test.ts b/console/apps/switch-console-desktop/src/main/core/switch-rooms/room-connection.test.ts index 3e5e4652a..0e7ea8a04 100644 --- a/console/apps/switch-console-desktop/src/main/core/switch-rooms/room-connection.test.ts +++ b/console/apps/switch-console-desktop/src/main/core/switch-rooms/room-connection.test.ts @@ -682,6 +682,60 @@ describe('RoomConnection', () => { expect(fetchMock.mock.calls.length).toBe(before); }); + /** + * Closing the connection on teardown (CHOO-2497). + * + * Deleting a session used to tell switch-core nothing, so the server kept the + * session's claim on the room until the heartbeat sweep expired it — and for + * those seconds the agent's watcher stayed dark on that room, so pinging it + * spawned no replacement session. + */ + describe('stop()', () => { + function closeCalls(fetchMock: ReturnType) { + return fetchMock.mock.calls + .filter((c) => String(c[0]).includes('/connection/close')) + .map((c) => JSON.parse((c[1] as RequestInit).body as string)); + } + + it('tells the server to drop the connection so the room is freed at once', async () => { + const { conn, fetchMock } = connect({ acquire: () => null }, []); + await flush(); + + conn.stop(); + await flush(); + + expect(closeCalls(fetchMock)).toEqual([ + { connection_id: 'conn-1', reason: 'session stopped' }, + ]); + }); + + it('closes once even if stop is called twice', async () => { + const { conn, fetchMock } = connect({ acquire: () => null }, []); + await flush(); + + conn.stop(); + conn.stop(); + await flush(); + + expect(closeCalls(fetchMock)).toHaveLength(1); + }); + + it('warns and carries on when the close cannot be delivered', async () => { + const { conn, fetchMock } = connect({ acquire: () => null }, []); + await flush(); + fetchMock.mockImplementation(async () => { + throw new Error('network down'); + }); + + expect(() => conn.stop()).not.toThrow(); + await flush(); + + expect( + silentLog.warn.mock.calls.some((c) => c[1]?.event === 'room_connection_close_failed') + ).toBe(true); + }); + }); + it('reports control_capabilities in the runtime-state report', async () => { const target: InjectionTarget = { write: vi.fn() }; const { conn, fetchMock } = connect({ acquire: () => target }, []); diff --git a/console/apps/switch-console-desktop/src/main/core/switch-rooms/room-connection.ts b/console/apps/switch-console-desktop/src/main/core/switch-rooms/room-connection.ts index e5a8b8810..ffee5e7d8 100644 --- a/console/apps/switch-console-desktop/src/main/core/switch-rooms/room-connection.ts +++ b/console/apps/switch-console-desktop/src/main/core/switch-rooms/room-connection.ts @@ -478,6 +478,12 @@ export class RoomConnection { this.currentAnchorId = null; void this.postRuntimeState('idle', null, { detached: true }).catch(() => {}); } + // Tell the server the connection is gone instead of letting the heartbeat + // lapse. Until it is closed the room slot stays claimed, the agent's + // all-scope watcher stays dark on that room, and a message arriving in the + // meantime spawns nothing (CHOO-2497). Detached and best-effort for the + // same reason as the idle report above; the sweep remains the backstop. + void this.closeConnection(); this.abort.abort(); if (this.busyFallback) clearTimeout(this.busyFallback); if (this.humanGateTimer) clearTimeout(this.humanGateTimer); @@ -534,6 +540,47 @@ export class RoomConnection { return fetch(url, { ...init, signal: AbortSignal.any(signals) }); } + /** + * Ask the server to drop this session's connection, releasing its room slot + * at once. Failure is logged rather than thrown: teardown must not depend on + * the network, and the heartbeat sweep still collects the connection a few + * seconds later. + */ + private async closeConnection(): Promise { + try { + const resp = await this.fetchWithTimeout( + `${this.creds.apiEndpoint}/agents/${this.creds.agentId}/connection/close`, + { + method: 'POST', + headers: { + Authorization: `Bearer ${this.creds.token}`, + 'Content-Type': 'application/json', + }, + body: JSON.stringify({ + connection_id: this.connectionId, + reason: 'session stopped', + }), + }, + RUNTIME_STATE_REQUEST_TIMEOUT_MS, + { detached: true } + ); + if (!resp.ok) { + throw new Error(`HTTP ${resp.status}: ${await resp.text()}`); + } + } catch (error) { + this.log.warn( + 'RoomConnection: could not close the connection; falling back to the server sweep', + { + event: 'room_connection_close_failed', + sessionId: this.sessionId, + roomId: this.roomId, + connectionId: this.connectionId, + error: error instanceof Error ? error.message : String(error), + } + ); + } + } + private async postRuntimeState( state: RuntimeState, threadId: string | null, diff --git a/core/switch_core/bridges/agent/api/handlers.py b/core/switch_core/bridges/agent/api/handlers.py index e70627ea8..6239cdf96 100644 --- a/core/switch_core/bridges/agent/api/handlers.py +++ b/core/switch_core/bridges/agent/api/handlers.py @@ -25,6 +25,7 @@ BulkRegisterResult, CancelTaskRequest, ConnectionBeatRequest, + ConnectionCloseRequest, ConnectionRenewRequest, ConnectionSubscribeRequest, CreateModerationRoomRequest, @@ -857,6 +858,32 @@ async def connection_beat( return {"ok": True, "rooms": sorted(conn.rooms), "cursor": conn.cursor} +@router.post("/{agent_id}/connection/close") +async def connection_close( + agent_id: str, + req: ConnectionCloseRequest, + agent: Annotated[Agent, Depends(get_agent_from_scope)], + protocol: Annotated[ProtocolService, Depends(get_protocol)], +) -> dict[str, Any]: + """Close a connection now, releasing its room slots immediately. + + Without this a client that goes away — a session the user deleted, above + all — keeps its room claim until the heartbeat sweep notices, and for those + few seconds the agent's all-scope watcher stays dark on that room and no + replacement session is spawned. + + Idempotent: closing a connection that is already gone is the outcome the + caller wanted, so it reports `closed: false` rather than failing. + """ + try: + conn = protocol.connections.require(agent.id, req.connection_id) + except UnknownConnectionError: + return {"ok": True, "closed": False} + + protocol.connections.close(conn.id, req.reason) + return {"ok": True, "closed": True} + + @router.post("/{agent_id}/connection/subscribe") async def connection_subscribe( agent_id: str, diff --git a/core/switch_core/bridges/agent/api/schemas.py b/core/switch_core/bridges/agent/api/schemas.py index 5faa1e8dc..2c2847984 100644 --- a/core/switch_core/bridges/agent/api/schemas.py +++ b/core/switch_core/bridges/agent/api/schemas.py @@ -119,6 +119,13 @@ class ConnectionSubscribeRequest(BaseModel): takeover: bool = False +class ConnectionCloseRequest(BaseModel): + """Tear a connection down now rather than letting its heartbeat lapse.""" + + connection_id: str + reason: str = "client closed" + + class ConnectionBeatRequest(BaseModel): """The single client tick that keeps a connection alive (CHOO-1857). diff --git a/core/tests/switch_core/bridges/agent/api/test_connection_close.py b/core/tests/switch_core/bridges/agent/api/test_connection_close.py new file mode 100644 index 000000000..511f30d05 --- /dev/null +++ b/core/tests/switch_core/bridges/agent/api/test_connection_close.py @@ -0,0 +1,107 @@ +"""Closing a connection releases its room slot at once (CHOO-2497). + +Deleting a session used to tell switch-core nothing. The session's connection +kept its room claim until the heartbeat sweep collected it, and for those +seconds the agent's all-scope watcher stayed dark on the room — so a message +arriving in the gap spawned no replacement session and the user saw a delay. +""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest + +from switch_core.bridges.agent.api.handlers import connection_close +from switch_core.bridges.agent.api.schemas import ConnectionCloseRequest +from switch_core.bridges.agent.protocol.connections import ( + PROTOCOL_VERSION, + ClientDeclaration, + ConnectionRegistry, +) + +AGENT_ID = "agent-1" +OTHER_AGENT_ID = "agent-2" +SESSION_CONN = "session-conn" +WATCHER_CONN = "watcher-conn" +ROOM_ID = "room-1" + + +class _Protocol: + def __init__(self) -> None: + self.connections = ConnectionRegistry() + + +def _open( + protocol: _Protocol, connection_id: str, scope: str, agent_id: str = AGENT_ID +) -> Any: + conn = protocol.connections.open( + agent_id=agent_id, + connection_id=connection_id, + scope=scope, # type: ignore[arg-type] + delivery_filter="all", + spawn_capable=scope == "all", + cursor=0, + declaration=ClientDeclaration(speaks=PROTOCOL_VERSION), + ) + conn.stream_attached = True + return conn + + +async def _close( + protocol: _Protocol, connection_id: str, agent_id: str = AGENT_ID +) -> Any: + return await connection_close( + agent_id, + ConnectionCloseRequest(connection_id=connection_id, reason="session stopped"), + SimpleNamespace(id=agent_id), # type: ignore[arg-type] + protocol, # type: ignore[arg-type] + ) + + +@pytest.mark.asyncio +async def test_closing_a_session_hands_the_room_back_to_the_watcher() -> None: + protocol = _Protocol() + watcher = _open(protocol, WATCHER_CONN, "all") + session = _open(protocol, SESSION_CONN, "single") + protocol.connections.claim_room(session, ROOM_ID) + assert not protocol.connections.covers(watcher, ROOM_ID) + + result = await _close(protocol, SESSION_CONN) + + assert result == {"ok": True, "closed": True} + # No sweep, no waiting on the heartbeat TTL: the watcher covers the room + # again immediately, so the next message spawns a session. + assert protocol.connections.covers(watcher, ROOM_ID) + assert protocol.connections.holder_of(AGENT_ID, ROOM_ID) is watcher + assert not protocol.connections.has_session_in(AGENT_ID, ROOM_ID) + + +@pytest.mark.asyncio +async def test_closing_an_unknown_connection_is_the_outcome_the_caller_wanted() -> None: + protocol = _Protocol() + + result = await _close(protocol, SESSION_CONN) + + assert result == {"ok": True, "closed": False} + + +@pytest.mark.asyncio +async def test_closing_twice_is_harmless() -> None: + protocol = _Protocol() + _open(protocol, SESSION_CONN, "single") + + assert await _close(protocol, SESSION_CONN) == {"ok": True, "closed": True} + assert await _close(protocol, SESSION_CONN) == {"ok": True, "closed": False} + + +@pytest.mark.asyncio +async def test_one_agent_cannot_close_another_agents_connection() -> None: + protocol = _Protocol() + _open(protocol, SESSION_CONN, "single") + + result = await _close(protocol, SESSION_CONN, agent_id=OTHER_AGENT_ID) + + assert result == {"ok": True, "closed": False} + assert protocol.connections.require(AGENT_ID, SESSION_CONN) is not None