fix(sessions): close the connection on teardown instead of waiting for the TTL (CHOO-2497)

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<typeof makeFetch>) {
+      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 }, []);

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<void> {
+    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,

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,

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).
 

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