diff --git a/console/apps/switch-console-desktop/src/main/core/agent-runtime/impl/remote-sidecar-launcher.ts b/console/apps/switch-console-desktop/src/main/core/agent-runtime/impl/remote-sidecar-launcher.ts index 07ce46f22..c2bbd912b 100644 --- a/console/apps/switch-console-desktop/src/main/core/agent-runtime/impl/remote-sidecar-launcher.ts +++ b/console/apps/switch-console-desktop/src/main/core/agent-runtime/impl/remote-sidecar-launcher.ts @@ -523,15 +523,53 @@ export class RemoteSidecarLauncher { * and relaunched — otherwise a bundle upgrade never takes effect while the old * process keeps running. */ - /** The running sidecar's ready line, or null when none is running. */ + /** The running sidecar's ready line, or null when none is running. + * + * Checks both tmux (the process's supervisor) and the ready file's PID + * directly (CHOO-2653). The PID check catches sidecars running outside + * tmux, under a renamed tmux session, or via systemd — any case where the + * process is alive but tmux `has-session` misses it. + */ private async readRunning(): Promise { + let tmuxAlive = false; try { await this.host.exec('tmux', ['has-session', '-t', exactTmuxTarget(this.sidecarTmuxName)]); + tmuxAlive = true; } catch { - return null; // not running + // tmux session not found — fall through to PID check } + const raw = await this.readReadyFile(); - return raw ? parseReady(raw) : null; + const ready = raw ? parseReady(raw) : null; + if (!ready) return null; + + // tmux found it — the common case + if (tmuxAlive) return ready; + + // tmux didn't find it, but the ready file reports a PID. Check if that + // process is still alive on the host. + if (ready.pid == null) return null; + try { + await this.host.exec('kill', ['-0', String(ready.pid)]); + } catch { + return null; // PID is gone + } + // A live PID can be a recycled one. Confirm the process is actually a + // sidecar (a node process) before trusting the ready line — a wrong yes + // here leaves the launcher believing a sidecar serves this agent while + // nothing listens on the reported port. A wrong no is safe: relaunching + // hits the sidecar's own single-instance guard and exits cleanly. + try { + const { stdout } = await this.host.exec('ps', ['-p', String(ready.pid), '-o', 'args=']); + if (!/node|sidecar/i.test(stdout)) return null; + } catch { + return null; // ps unavailable or PID vanished — treat as not running + } + this.log.debug('RemoteSidecarLauncher: sidecar PID alive outside its tmux session', { + sidecarTmuxName: this.sidecarTmuxName, + pid: ready.pid, + }); + return ready; } /** diff --git a/console/apps/switch-console-desktop/src/main/core/agent-runtime/impl/ssh-agent-runtime.ts b/console/apps/switch-console-desktop/src/main/core/agent-runtime/impl/ssh-agent-runtime.ts index 9daf9112b..6e300a297 100644 --- a/console/apps/switch-console-desktop/src/main/core/agent-runtime/impl/ssh-agent-runtime.ts +++ b/console/apps/switch-console-desktop/src/main/core/agent-runtime/impl/ssh-agent-runtime.ts @@ -415,6 +415,9 @@ export class SshAgentRuntime implements AgentRuntimeProvider, AttachableRuntime const host = this.createSidecarHost(); const credsSlug = agentCredsSlug(session); const specialization = await agentLaunchSpecialization(session.agentId); + // Drop any sidecar left in this directory by an earlier generation of the + // agent's name BEFORE deploying, so a leftover does not collide (CHOO-2653). + if (agent) await reapStaleSidecarsForAgent(agent, host, this.sessionPath); // Every session in this dir would otherwise re-run the same deploy+launch on // startup; coalesce so one host sees one ensure, not one per session. const endpoint = await dedupeInFlight( @@ -434,9 +437,6 @@ export class SshAgentRuntime implements AgentRuntimeProvider, AttachableRuntime host, }) ); - // Drop any sidecar left in this directory by an earlier generation of the - // agent's name — it is still polling Switch and no other path can see it. - if (agent) await reapStaleSidecarsForAgent(agent, host, this.sessionPath); this.sidecarEndpoint = endpoint; this.joinRelay(endpoint, session, credsSlug); return endpoint; diff --git a/console/apps/switch-console-desktop/src/main/core/agents/remote-watcher.ts b/console/apps/switch-console-desktop/src/main/core/agents/remote-watcher.ts index 39bc45339..129d4c1a3 100644 --- a/console/apps/switch-console-desktop/src/main/core/agents/remote-watcher.ts +++ b/console/apps/switch-console-desktop/src/main/core/agents/remote-watcher.ts @@ -111,6 +111,9 @@ export async function ensureRemoteWatcher(agentId: string): Promise { } const { ctx, connectionId, remoteRepoDir, host } = await connectRemoteAgent(agent); + // Reap stale sidecars BEFORE deploying, so a leftover generation from a + // renamed agent does not collide with the new one (CHOO-2653). + await reapStaleSidecarsForAgent(agent, host, remoteRepoDir); await ensureAgentSidecar({ providerId: agent.providerId, repoDir: remoteRepoDir, @@ -124,7 +127,6 @@ export async function ensureRemoteWatcher(agentId: string): Promise { host, }); await writeWatchEnabled(host, agent.name ?? agent.id, true); - await reapStaleSidecarsForAgent(agent, host, remoteRepoDir); log.info('ensureRemoteWatcher: sidecar deployed + watching', { agentId, switchAgentId: agent.switchAgentId, diff --git a/console/apps/switch-console-desktop/src/main/core/sidecar/controller.ts b/console/apps/switch-console-desktop/src/main/core/sidecar/controller.ts index e2cf209c1..0fd73f376 100644 --- a/console/apps/switch-console-desktop/src/main/core/sidecar/controller.ts +++ b/console/apps/switch-console-desktop/src/main/core/sidecar/controller.ts @@ -101,8 +101,10 @@ export const sidecarController = createRPCController({ upgrade: async (agentId: string): Promise => { const { agent } = await requireRemoteAgent(agentId); const params = await paramsForAgent(agent); - await ensureAgentSidecar(params); + // Reap stale sidecars BEFORE deploying, so a leftover generation from a + // renamed agent does not collide with the new one (CHOO-2653). await reapStaleSidecarsForAgent(agent, params.host, params.repoDir); + await ensureAgentSidecar(params); return readAndBroadcast(agentId); }, @@ -114,8 +116,8 @@ export const sidecarController = createRPCController({ restart: async (agentId: string): Promise => { const { agent } = await requireRemoteAgent(agentId); const params = await paramsForAgent(agent); - await restartAgentSidecar(params); await reapStaleSidecarsForAgent(agent, params.host, params.repoDir); + await restartAgentSidecar(params); return readAndBroadcast(agentId); }, diff --git a/console/apps/switch-console-desktop/src/sidecar/index.ts b/console/apps/switch-console-desktop/src/sidecar/index.ts index 3d372e2a4..2053eaedc 100644 --- a/console/apps/switch-console-desktop/src/sidecar/index.ts +++ b/console/apps/switch-console-desktop/src/sidecar/index.ts @@ -36,6 +36,7 @@ import { import { defaultRoomConnectionFactory, SidecarRuntime } from './sidecar-runtime'; import { SidecarStateStore } from './sidecar-state'; import { SIDECAR_CONTROL, SIDECAR_VERSION } from './sidecar-version'; +import { existingSidecarIsHealthy } from './single-instance-guard'; import { exactTmuxTarget, parseAgentTmuxSessionName } from './vm-tmux'; /** @@ -139,6 +140,15 @@ async function main(): Promise { // Per-agent state paths, so multiple agents in one repo dir each drive their // own sidecar without clobbering each other's spec/watch flag (CHOO-1440). // Fall back to the legacy shared paths when launched without a slug. + const stateSlug = credsSlug ?? 'default'; + + // Single-instance guard (CHOO-2653): refuse to start if a healthy sidecar for + // this agent is already running. Checked before any port binding or state + // loading, so the duplicate never touches shared files. + if (await existingSidecarIsHealthy(repoDir, stateSlug, log)) { + process.exit(0); + } + const launchSpecRel = credsSlug ? sidecarLaunchSpecRelPath(credsSlug) : LEGACY_LAUNCH_SPEC_REL_PATH; @@ -178,7 +188,6 @@ async function main(): Promise { // Durable session registry. Restored entries whose pane is gone are dropped // here, so what survives is what is actually still running on the host. - const stateSlug = credsSlug ?? 'default'; const store = await SidecarStateStore.open({ repoDir, slug: stateSlug, diff --git a/console/apps/switch-console-desktop/src/sidecar/single-instance-guard.test.ts b/console/apps/switch-console-desktop/src/sidecar/single-instance-guard.test.ts new file mode 100644 index 000000000..8242916ea --- /dev/null +++ b/console/apps/switch-console-desktop/src/sidecar/single-instance-guard.test.ts @@ -0,0 +1,75 @@ +import { mkdtemp, mkdir, writeFile, rm } from 'node:fs/promises'; +import http from 'node:http'; +import os from 'node:os'; +import path from 'node:path'; +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest'; +import { existingSidecarIsHealthy } from './single-instance-guard'; + +const SLUG = 'test-agent'; +const READY_REL = `.switchdash/agents/${SLUG}/sidecar.ready`; + +const noop = { info: vi.fn(), warn: vi.fn() }; + +let tmpDir: string; + +beforeEach(async () => { + tmpDir = await mkdtemp(path.join(os.tmpdir(), 'guard-test-')); + await mkdir(path.join(tmpDir, `.switchdash/agents/${SLUG}`), { recursive: true }); +}); + +afterEach(async () => { + await rm(tmpDir, { recursive: true, force: true }); +}); + +function writeReady(overrides: Record = {}): Promise { + const line = JSON.stringify({ + event: 'ready', + port: 99999, + token: 'test-token', + pid: 999999999, + ...overrides, + }); + return writeFile(path.join(tmpDir, READY_REL), line + '\n'); +} + +describe('existingSidecarIsHealthy', () => { + it('returns false when the ready file is missing', async () => { + expect(await existingSidecarIsHealthy(tmpDir, SLUG, noop)).toBe(false); + }); + + it('returns false when the ready file has a dead PID', async () => { + // PID 999999999 should not exist on any system. + await writeReady({ pid: 999999999 }); + expect(await existingSidecarIsHealthy(tmpDir, SLUG, noop)).toBe(false); + }); + + it('returns false when the PID is alive but the port does not respond', async () => { + // Use our own PID (known alive) with a port nothing is listening on. + await writeReady({ pid: process.pid, port: 1 }); + // process.pid === our PID is excluded by the self-check, so use ppid. + await writeReady({ pid: process.ppid, port: 1 }); + expect(await existingSidecarIsHealthy(tmpDir, SLUG, noop)).toBe(false); + }); + + it('returns true when the PID is alive and the port responds OK', async () => { + // Start a tiny HTTP server that returns 200 on any request. + const server = http.createServer((_req, res) => { + res.writeHead(200); + res.end('[]'); + }); + await new Promise((resolve) => server.listen(0, '127.0.0.1', resolve)); + const port = (server.address() as { port: number }).port; + + try { + await writeReady({ pid: process.ppid, port, token: 'anything' }); + expect(await existingSidecarIsHealthy(tmpDir, SLUG, noop)).toBe(true); + } finally { + server.close(); + } + }); + + it('returns false when the PID matches the current process (self-detection)', async () => { + await writeReady({ pid: process.pid }); + expect(await existingSidecarIsHealthy(tmpDir, SLUG, noop)).toBe(false); + }); +}); diff --git a/console/apps/switch-console-desktop/src/sidecar/single-instance-guard.ts b/console/apps/switch-console-desktop/src/sidecar/single-instance-guard.ts new file mode 100644 index 000000000..76e7e9859 --- /dev/null +++ b/console/apps/switch-console-desktop/src/sidecar/single-instance-guard.ts @@ -0,0 +1,68 @@ +import { readFile } from 'node:fs/promises'; +import path from 'node:path'; +import { sidecarReadyRelPath } from './sidecar-paths'; + +/** + * Single-instance guard (CHOO-2653). + * + * Reads the existing ready file for this agent, checks whether its PID is still + * alive, and probes its HTTP port. If a healthy sidecar is already running, + * returns true — so the caller can exit cleanly instead of starting a duplicate + * that would livelock the first. + */ +export async function existingSidecarIsHealthy( + repoDir: string, + stateSlug: string, + log: { info(...input: unknown[]): void; warn(...input: unknown[]): void } +): Promise { + const readyPath = path.join(repoDir, sidecarReadyRelPath(stateSlug)); + let raw: string; + try { + raw = await readFile(readyPath, 'utf8'); + } catch { + return false; + } + + const line = raw + .split('\n') + .map((l) => l.trim()) + .find(Boolean); + if (!line) return false; + + let parsed: { event?: string; pid?: number; port?: number; token?: string }; + try { + parsed = JSON.parse(line); + } catch { + return false; + } + + if (parsed.event !== 'ready') return false; + if (typeof parsed.pid !== 'number' || parsed.pid === process.pid) return false; + + try { + process.kill(parsed.pid, 0); + } catch { + return false; // process is gone + } + + // PID is alive — probe its HTTP endpoint to confirm it is a sidecar and not + // a recycled PID. The /sessions endpoint is token-gated and returns JSON. + if (typeof parsed.port !== 'number' || typeof parsed.token !== 'string') return false; + try { + const resp = await fetch(`http://127.0.0.1:${parsed.port}/sessions`, { + // The hook server gates on this header, not Authorization (hook-server.ts). + headers: { 'x-switchdash-token': parsed.token }, + signal: AbortSignal.timeout(2000), + }); + if (resp.ok) { + log.info('sidecar: another instance is already running for this agent', { + existingPid: parsed.pid, + existingPort: parsed.port, + }); + return true; + } + } catch { + // Port not responding — stale ready file with a recycled PID. + } + return false; +} diff --git a/console/packages/switch-agent-runtime/src/event-stream.ts b/console/packages/switch-agent-runtime/src/event-stream.ts index eb2eba980..2390d9d0e 100644 --- a/console/packages/switch-agent-runtime/src/event-stream.ts +++ b/console/packages/switch-agent-runtime/src/event-stream.ts @@ -113,6 +113,9 @@ export class SwitchEventStream { * tearing down the connection. */ private socketAbort: AbortController | null = null; private rooms: string[]; + /** Set when the current stream was ended by an eviction frame, so the + * reconnect loop backs off instead of hammering immediately (CHOO-2653). */ + private wasEvicted = false; constructor(deps: SwitchEventStreamDeps) { this.deps = deps; @@ -292,6 +295,24 @@ export class SwitchEventStream { if (frame.id) this.cursor = Math.max(this.cursor, Number(frame.id) || 0); await this.handleFrame(frame); } + + // An eviction ends the stream cleanly (no error), so without an + // explicit check the loop restarts immediately — and two supervisors + // on one connection id evict each other at ~9/s (CHOO-2653). Back off + // on eviction the same way the error path does on failures. + if (this.wasEvicted) { + this.wasEvicted = false; + failures += 1; + if ((failures & (failures - 1)) === 0) { + log.warn('SwitchEventStream: backing off after eviction', { + event: 'switch_stream_eviction_backoff', + failures, + backoffMs: backoff, + }); + } + await new Promise((r) => setTimeout(r, backoff)); + backoff = Math.min(backoff * 2, MAX_BACKOFF_MS); + } } catch (error) { if (signal.aborted) return; // A deliberate reopen (repoint) aborts the socket; that is not an error. @@ -346,6 +367,7 @@ export class SwitchEventStream { }); return; case 'evicted': + this.wasEvicted = true; log.warn('SwitchEventStream: evicted', { event: 'switch_stream_evicted', reason: frame.data.reason, diff --git a/core/switch_core/bridges/agent/protocol/connections.py b/core/switch_core/bridges/agent/protocol/connections.py index a37fa6206..508121d0e 100644 --- a/core/switch_core/bridges/agent/protocol/connections.py +++ b/core/switch_core/bridges/agent/protocol/connections.py @@ -58,6 +58,13 @@ # visible error instead of quiet resource creep. MAX_CONNECTIONS_PER_AGENT = 32 +# Minimum interval between reattaches on the same connection (CHOO-2653). +# Two supervisors on one connection id reattach each other at full rate +# (~9/s in the field), which dominates the log and makes neither one usable. +# The server refuses a reattach that arrives too soon, forcing the client to +# back off. +MIN_REATTACH_INTERVAL_SECONDS = 2.0 + class ConnectionError_(Exception): """Base for connection faults that a client must be told about.""" @@ -180,6 +187,20 @@ def __init__(self, *, client_speaks: int, client_accepts: int) -> None: self.remedy = remedy +class ReattachTooSoonError(ConnectionError_): + """A reattach arrived before the minimum interval elapsed (CHOO-2653).""" + + def __init__(self, connection_id: str, elapsed_seconds: float) -> None: + retry_after = max(0.0, MIN_REATTACH_INTERVAL_SECONDS - elapsed_seconds) + super().__init__( + f"connection {connection_id} was reattached {elapsed_seconds:.1f}s " + f"ago; retry in {retry_after:.1f}s" + ) + self.connection_id = connection_id + self.elapsed_seconds = elapsed_seconds + self.retry_after_seconds = retry_after + + class TooManyConnectionsError(ConnectionError_): def __init__(self, agent_id: str, limit: int) -> None: super().__init__( @@ -212,6 +233,9 @@ class Connection: # has been replaced and stop writing. stream_generation: int = 0 closed_reason: str | None = None + # Monotonic time of the last stream reattach, so the server can refuse a + # reattach that arrives too soon (CHOO-2653). + last_reattach: float = 0.0 # What the client said about itself on connect (CHOO-1865). Defaults to an # empty declaration, which means unknown — never "current". declaration: ClientDeclaration = field(default_factory=lambda: ClientDeclaration()) @@ -274,19 +298,30 @@ def open( if existing.agent_id != agent_id: # Never let one agent attach to another's connection. raise UnknownConnectionError(connection_id) + + # Refuse a reattach that arrives too soon after the previous one + # (CHOO-2653). Two supervisors on one connection id would otherwise + # evict each other at ~9 reattaches/second. + now = time.monotonic() + if existing.last_reattach > 0: + elapsed = now - existing.last_reattach + if elapsed < MIN_REATTACH_INTERVAL_SECONDS: + raise ReattachTooSoonError(connection_id, elapsed) + existing.scope = scope existing.delivery_filter = delivery_filter existing.spawn_capable = spawn_capable existing.cursor = cursor - existing.last_beat = time.monotonic() + existing.last_beat = now existing.closed_reason = None existing.stream_attached = True existing.stream_generation += 1 + existing.last_reattach = now # A reattach can come from an upgraded client, so the declaration # is replaced rather than kept. The connection outlives the socket; # what is on the other end of it need not. existing.declaration = declaration - logger.info( + logger.debug( "[CONN] reattached agent=%s connection=%s scope=%s generation=%s", agent_id, connection_id, diff --git a/core/tests/switch_core/bridges/agent/protocol/test_connections.py b/core/tests/switch_core/bridges/agent/protocol/test_connections.py index 0bf89eb47..ad3102e07 100644 --- a/core/tests/switch_core/bridges/agent/protocol/test_connections.py +++ b/core/tests/switch_core/bridges/agent/protocol/test_connections.py @@ -9,12 +9,14 @@ from switch_core.bridges.agent.protocol.connections import ( HEARTBEAT_TTL_SECONDS, MAX_CONNECTIONS_PER_AGENT, + MIN_REATTACH_INTERVAL_SECONDS, PROTOCOL_ACCEPTS, PROTOCOL_VERSION, ClientDeclaration, ConnectionRegistry, NoStreamAttachedError, ProtocolVersionError, + ReattachTooSoonError, RoomOccupiedError, TooManyConnectionsError, UnknownConnectionError, @@ -176,6 +178,45 @@ def test_a_reattach_replaces_the_declaration() -> None: assert conn.declaration.version == "1.1.0" +# ── Reattach throttle (CHOO-2653) ───────────────────────────────────────── + + +def test_first_reattach_always_succeeds() -> None: + """The interval check is skipped on the very first reattach (last_reattach == 0).""" + registry = ConnectionRegistry() + conn = _open(registry, "c1") + assert conn.stream_generation == 0 + + # Immediate reattach — no prior reattach timestamp to throttle against. + _open(registry, "c1") + assert conn.stream_generation == 1 + + +def test_reattach_too_soon_is_refused() -> None: + """Two reattaches within MIN_REATTACH_INTERVAL_SECONDS raises.""" + registry = ConnectionRegistry() + _open(registry, "c1") + _open(registry, "c1") # first reattach (sets last_reattach) + + with pytest.raises(ReattachTooSoonError) as exc_info: + _open(registry, "c1") # immediate second — refused + # The error exposes how long a client should wait before retrying. + assert 0 < exc_info.value.retry_after_seconds <= 2.0 + + +def test_reattach_after_interval_succeeds() -> None: + """A reattach that waits long enough is allowed.""" + registry = ConnectionRegistry() + _open(registry, "c1") + conn = _open(registry, "c1") # first reattach (generation 0 → 1) + + # Simulate time passing beyond the minimum interval. + conn.last_reattach = time.monotonic() - MIN_REATTACH_INTERVAL_SECONDS - 0.1 + + _open(registry, "c1") + assert conn.stream_generation == 2 + + def test_connection_cap_is_enforced_loudly() -> None: registry = ConnectionRegistry() for i in range(MAX_CONNECTIONS_PER_AGENT):