diff --git a/apps/desktop/electron/backend-command.test.ts b/apps/desktop/electron/backend-command.test.ts index 2e49160d26..2746ce78da 100644 --- a/apps/desktop/electron/backend-command.test.ts +++ b/apps/desktop/electron/backend-command.test.ts @@ -2,16 +2,9 @@ import assert from 'node:assert/strict' import { test } from 'vitest' -import { - dashboardFallbackArgs, - serveBackendArgs, - sourceDeclaresServe, - sourceDeclaresWsOnly -} from './backend-command' +import { dashboardFallbackArgs, serveBackendArgs, sourceDeclaresServe } from './backend-command' -test('serveBackendArgs builds a headless serve invocation WITHOUT --ws-only by default', () => { - // Opt-in until the desktop REST plane migrates to JSON-RPC (#94484 ph.3): - // the WS-only server has no HTTP routes and would break hermes:api. +test('serveBackendArgs builds a headless serve invocation', () => { assert.deepEqual(serveBackendArgs(), ['serve', '--host', '127.0.0.1', '--port', '0']) }) @@ -19,11 +12,6 @@ test('serveBackendArgs pins a profile when provided', () => { assert.deepEqual(serveBackendArgs('worker'), ['--profile', 'worker', 'serve', '--host', '127.0.0.1', '--port', '0']) }) -test('serveBackendArgs emits --ws-only only on explicit opt-in', () => { - assert.deepEqual(serveBackendArgs(undefined, { wsOnly: true }), ['serve', '--host', '127.0.0.1', '--port', '0', '--ws-only']) - assert.deepEqual(serveBackendArgs(undefined, { wsOnly: false }), ['serve', '--host', '127.0.0.1', '--port', '0']) -}) - test('dashboardFallbackArgs rewrites serve -> dashboard --no-open, keeping the -m prefix', () => { const serve = ['-m', 'hermes_cli.main', 'serve', '--host', '127.0.0.1', '--port', '0'] assert.deepEqual(dashboardFallbackArgs(serve), [ @@ -54,20 +42,6 @@ test('dashboardFallbackArgs preserves a --profile flag ahead of serve', () => { ]) }) -test('dashboardFallbackArgs strips --ws-only when falling back to dashboard', () => { - const serve = ['-m', 'hermes_cli.main', 'serve', '--host', '127.0.0.1', '--port', '0', '--ws-only'] - assert.deepEqual(dashboardFallbackArgs(serve), [ - '-m', - 'hermes_cli.main', - 'dashboard', - '--no-open', - '--host', - '127.0.0.1', - '--port', - '0' - ]) -}) - test('dashboardFallbackArgs is a no-op (copy) when there is no serve token', () => { const args = ['-m', 'hermes_cli.main', 'dashboard', '--no-open'] const out = dashboardFallbackArgs(args) @@ -89,13 +63,3 @@ test('sourceDeclaresServe does not false-positive on the substring "server"', () assert.equal(sourceDeclaresServe(oldSource), false) }) - -test('sourceDeclaresWsOnly detects the --ws-only flag registration', () => { - assert.equal(sourceDeclaresWsOnly('serve_parser.add_argument("--ws-only", dest="ws_only")'), true) - assert.equal(sourceDeclaresWsOnly("serve_parser.add_argument('--ws-only')"), true) -}) - -test('sourceDeclaresWsOnly returns false when the flag is absent', () => { - assert.equal(sourceDeclaresWsOnly('serve_parser.add_argument("--no-open")'), false) - assert.equal(sourceDeclaresWsOnly(''), false) -}) diff --git a/apps/desktop/electron/backend-command.ts b/apps/desktop/electron/backend-command.ts index 7de0ea369a..23cb87d4f7 100644 --- a/apps/desktop/electron/backend-command.ts +++ b/apps/desktop/electron/backend-command.ts @@ -9,31 +9,16 @@ // fall back to the legacy `dashboard --no-open` invocation. Both produce the // exact same headless gateway; `serve` is just the decoupled name. // -// `--ws-only` is an even newer flag: it bypasses the FastAPI/uvicorn dashboard -// entirely and runs a bare `websockets` server that calls `handle_ws` directly. -// When a runtime understands `serve` but NOT `--ws-only`, the flag is silently -// dropped by the runtime's argparse (it would fail "unrecognized arguments" on -// older versions). The `sourceDeclaresWsOnly` check gates this: only emit -// `--ws-only` when the runtime's source actually registers it. -// // These helpers are pure so they can be unit-tested without Electron. /** * Build the canonical headless backend argv (always `serve`). * @param {string} [profile] optional Hermes profile to pin via `--profile`. - * @param {object} [opts] runtime capability flags. - * @param {boolean} [opts.wsOnly=false] emit `--ws-only` (the slim WS server). - * OPT-IN: the WS-only server has no HTTP routes, and the desktop's - * `hermes:api` REST plane still speaks http — flipping this on by default - * would break every REST consumer (see PR #94245 review F1). Stays off - * until the REST consumers migrate to JSON-RPC (#94484 phase 3). */ -export function serveBackendArgs(profile?: string, opts?: { wsOnly?: boolean }) { +export function serveBackendArgs(profile?: string) { const head = profile ? ['--profile', profile] : [] - const wsOnly = opts?.wsOnly === true // default false — see docstring - const tail = wsOnly ? ['--ws-only'] : [] - return [...head, 'serve', '--host', '127.0.0.1', '--port', '0', ...tail] + return [...head, 'serve', '--host', '127.0.0.1', '--port', '0'] } /** @@ -49,9 +34,7 @@ export function dashboardFallbackArgs(args) { return args.slice() } - // `--ws-only` is serve-only; strip it when falling back to `dashboard`. - const rest = args.slice(i + 1).filter(a => a !== '--ws-only') - return [...args.slice(0, i), 'dashboard', '--no-open', ...rest] + return [...args.slice(0, i), 'dashboard', '--no-open', ...args.slice(i + 1)] } /** @@ -63,13 +46,3 @@ export function dashboardFallbackArgs(args) { export function sourceDeclaresServe(dashboardPySource) { return /add_parser\(\s*["']serve["']/.test(String(dashboardPySource || '')) } - -/** - * True when a runtime's `hermes_cli/subcommands/dashboard.py` source registers - * the `--ws-only` flag on the `serve` subcommand. Used to gate emitting the - * flag: an older runtime that knows `serve` but not `--ws-only` would reject - * it as an unrecognized argument. - */ -export function sourceDeclaresWsOnly(dashboardPySource) { - return /["']--ws-only["']/.test(String(dashboardPySource || '')) -} diff --git a/apps/desktop/electron/backend-ready.test.ts b/apps/desktop/electron/backend-ready.test.ts index 63d02bf16c..b51987a773 100644 --- a/apps/desktop/electron/backend-ready.test.ts +++ b/apps/desktop/electron/backend-ready.test.ts @@ -89,14 +89,14 @@ test('resolves with the announced port', async () => { const child = makeFakeChild() const p = waitForDashboardPort(child, 1000) child.stdout.emit('data', 'noise before\nHERMES_DASHBOARD_READY port=54321\n') - assert.deepEqual(await p, { port: 54321, token: undefined }) + assert.equal(await p, 54321) }) test('resolves with a HERMES_BACKEND_READY port (headless `serve`)', async () => { const child = makeFakeChild() const p = waitForDashboardPort(child, 1000) child.stdout.emit('data', 'HERMES_BACKEND_READY port=43210\n') - assert.deepEqual(await p, { port: 43210, token: undefined }) + assert.equal(await p, 43210) }) test('parses the port even when the line arrives split across chunks', async () => { @@ -104,7 +104,7 @@ test('parses the port even when the line arrives split across chunks', async () const p = waitForDashboardPort(child, 1000) child.stdout.emit('data', 'HERMES_DASHBOARD_READY po') child.stdout.emit('data', 'rt=8080\n') - assert.deepEqual(await p, { port: 8080, token: undefined }) + assert.equal(await p, 8080) }) test('rejects when the child exits before announcing', async () => { @@ -185,7 +185,7 @@ test('waitForDashboardReadyFile resolves when the ready file appears', async () try { const p = waitForDashboardReadyFile(tmp.file, child, 1000) setTimeout(() => fs.writeFileSync(tmp.file, JSON.stringify({ port: 8765 })), 20) - assert.deepEqual(await p, { port: 8765 }) + assert.equal(await p, 8765) } finally { tmp.cleanup() } @@ -198,7 +198,7 @@ test('waitForDashboardPortAnnouncement uses ready file when provided', async () try { const p = waitForDashboardPortAnnouncement(child, { readyFile: tmp.file, timeoutMs: 1000 }) setTimeout(() => fs.writeFileSync(tmp.file, JSON.stringify({ port: 9876 })), 20) - assert.deepEqual(await p, { port: 9876 }) + assert.equal(await p, 9876) } finally { tmp.cleanup() } @@ -260,17 +260,3 @@ test('exit-before-announcement error stays clean when no output was buffered', a return true }) }) - -test('resolves with the token when the ready line includes one (--ws-only)', async () => { - const child = makeFakeChild() - const p = waitForDashboardPort(child, 1000) - child.stdout.emit('data', 'HERMES_BACKEND_READY port=9999 token=abc123-def456\n') - assert.deepEqual(await p, { port: 9999, token: 'abc123-def456' }) -}) - -test('token is undefined when the ready line omits it (dashboard path)', async () => { - const child = makeFakeChild() - const p = waitForDashboardPort(child, 1000) - child.stdout.emit('data', 'HERMES_DASHBOARD_READY port=7777\n') - assert.deepEqual(await p, { port: 7777, token: undefined }) -}) diff --git a/apps/desktop/electron/backend-ready.ts b/apps/desktop/electron/backend-ready.ts index a1a07b7541..d7644fc06a 100644 --- a/apps/desktop/electron/backend-ready.ts +++ b/apps/desktop/electron/backend-ready.ts @@ -3,10 +3,7 @@ import fs from 'node:fs' // `hermes serve` announces HERMES_BACKEND_READY; the legacy `hermes dashboard` // backend announces HERMES_DASHBOARD_READY. Accept either so the desktop spawn // works against both the headless backend and old/dashboard runtimes. -// The `--ws-only` path also emits `token=` on the same line (the slim -// server auto-generates a session token when HERMES_DASHBOARD_SESSION_TOKEN -// isn't set); capture it so the desktop can pass it to the renderer. -const _READY_RE = /^HERMES_(?:BACKEND|DASHBOARD)_READY port=(\d+)(?:\s+token=(\S+))?/m +const _READY_RE = /^HERMES_(?:BACKEND|DASHBOARD)_READY port=(\d+)/m // The announcement clock starts the instant the backend process is spawned — // before uvicorn binds its socket. On a cold install the child must first @@ -54,7 +51,7 @@ function resolvePortAnnounceTimeoutMs(env = process.env) { * on every terminal path — resolve, reject, or timeout — so repeated * backend spawns don't leak listener slots on the child. */ -function waitForDashboardPort(child, timeoutMs = resolvePortAnnounceTimeoutMs(), describeOutputTail = () => ''): Promise<{ port: number; token?: string }> { +function waitForDashboardPort(child, timeoutMs = resolvePortAnnounceTimeoutMs(), describeOutputTail = () => '') { return new Promise((resolve, reject) => { let buf = '' let done = false @@ -82,7 +79,7 @@ function waitForDashboardPort(child, timeoutMs = resolvePortAnnounceTimeoutMs(), if (m) { cleanup() - resolve({ port: parseInt(m[1], 10), token: m[2] || undefined }) + resolve(parseInt(m[1], 10)) return } @@ -130,7 +127,7 @@ function waitForDashboardReadyFile( child, timeoutMs = resolvePortAnnounceTimeoutMs(), describeOutputTail = () => '' -): Promise<{ port: number; token?: string }> { +) { return new Promise((resolve, reject) => { let done = false let interval = null @@ -156,7 +153,7 @@ function waitForDashboardReadyFile( if (port) { cleanup() - resolve({ port }) + resolve(port) } } @@ -195,7 +192,7 @@ function waitForDashboardPortAnnouncement( readyFile?: fs.PathOrFileDescriptor | null timeoutMs?: number } = {} -): Promise<{ port: number; token?: string }> { +) { const timeoutMs = options.timeoutMs ?? resolvePortAnnounceTimeoutMs() const describeOutputTail = options.describeOutputTail ?? (() => '') diff --git a/apps/desktop/electron/main.ts b/apps/desktop/electron/main.ts index c5354dfef1..4b339db9b2 100644 --- a/apps/desktop/electron/main.ts +++ b/apps/desktop/electron/main.ts @@ -44,7 +44,7 @@ import { probeStartMarker, processStartMarker } from './backend-claim' -import { dashboardFallbackArgs, sourceDeclaresServe, sourceDeclaresWsOnly } from './backend-command' +import { dashboardFallbackArgs, sourceDeclaresServe } from './backend-command' import { createBackendConnectionState } from './backend-connection-state' import { BackendDialClaims } from './backend-dial-claim' import { buildDesktopBackendEnv, hermesManagedNodePathEntries, normalizeHermesHomeRoot } from './backend-env' @@ -2354,51 +2354,11 @@ function backendSupportsServe(backend) { return supported } -// Same pattern as backendSupportsServe, but for the `--ws-only` flag. -// A runtime that knows `serve` but not `--ws-only` would reject the flag -// as an unrecognized argument, so we gate it on source inspection. -const _wsOnlySupportCache = new Map() - -function backendSupportsWsOnly(backend) { - if (!backend || !backend.root) { - // No source to check — assume supported (the flag is silently ignored - // by argparse on older runtimes only when it's not registered; a runtime - // without a source root is likely a dev checkout that has it). - return true - } - - const key = `${backend.command}::${backend.root}` - - if (_wsOnlySupportCache.has(key)) { - return _wsOnlySupportCache.get(key) - } - - let supported = true - - try { - const src = fs.readFileSync(path.join(backend.root, 'hermes_cli', 'subcommands', 'dashboard.py'), 'utf8') - supported = sourceDeclaresWsOnly(src) - } catch { - supported = true // source unreadable — assume supported (the current runtime) - } - - _wsOnlySupportCache.set(key, supported) - return supported -} - // Given a resolved backend whose args target `serve`, return the args the // runtime actually understands: unchanged when `serve` is supported, or -// rewritten to `dashboard --no-open` for older runtimes. Also strips -// `--ws-only` when the runtime knows `serve` but not the flag. +// rewritten to `dashboard --no-open` for older runtimes. function getBackendArgsForRuntime(backend) { - if (!backendSupportsServe(backend)) { - return dashboardFallbackArgs(backend.args) - } - // `serve` is supported — check whether `--ws-only` is too. - if (backend.args.includes('--ws-only') && !backendSupportsWsOnly(backend)) { - return backend.args.filter(a => a !== '--ws-only') - } - return backend.args + return backendSupportsServe(backend) ? backend.args : dashboardFallbackArgs(backend.args) } function normalizeExecutablePathForCompare(commandPath) { @@ -12023,9 +11983,7 @@ async function spawnPoolBackend(profile, entry, opts: { forceLocal?: boolean; po } } - let token = crypto.randomBytes(32).toString('base64url') - // The slim --ws-only server may override this with its own announcement; - // `announcedToken` (from the ready sentinel) takes precedence when present. + const token = crypto.randomBytes(32).toString('base64url') // Same update mutual exclusion as the primary window's waitForLocalStart // (#73822): pool backends spawn from the same venv, so an ungated respawn @@ -12052,14 +12010,7 @@ async function spawnPoolBackend(profile, entry, opts: { forceLocal?: boolean; po // --profile wins over the inherited HERMES_HOME env (see _apply_profile_override // step 3 in hermes_cli/main.py), so the child re-homes to this profile. // --port 0: the OS assigns an ephemeral port; the child announces it on stdout. - // `--ws-only` is OPT-IN (HERMES_DESKTOP_WS_ONLY=1): the slim server has no - // HTTP routes, and the desktop's `hermes:api` REST plane still requires - // http — enabling it by default breaks every REST consumer (review F1 on - // PR #94245). Flips to default once the REST plane migrates (#94484 ph.3). const backendArgs = ['--profile', profile, 'serve', '--host', '127.0.0.1', '--port', '0'] - if (process.env.HERMES_DESKTOP_WS_ONLY === '1') { - backendArgs.push('--ws-only') - } const backend = await ensureRuntime(resolveHermesBackend(backendArgs)) // Route old runtimes (no `serve`) through the legacy `dashboard --no-open`. backend.args = getBackendArgsForRuntime(backend) @@ -12152,15 +12103,10 @@ async function spawnPoolBackend(profile, entry, opts: { forceLocal?: boolean; po }) // Discover the ephemeral port the child bound to - const announced = await Promise.race([ + const port = await Promise.race([ waitForDashboardPortAnnouncement(child, { describeOutputTail: () => outputTail.describe(), readyFile }), startFailed - ]) as { port: number; token?: string } - - const port = announced.port - // The slim --ws-only server emits its token on the ready line; the - // dashboard path resolves it separately via adoptServedDashboardToken. - const announcedToken = announced.token + ]) if (readyFile) { fs.unlink(readyFile, () => {}) @@ -12168,49 +12114,29 @@ async function spawnPoolBackend(profile, entry, opts: { forceLocal?: boolean; po entry.port = port - // The --ws-only server has no HTTP routes — skip waitForHermes when we - // have a direct token from the announcement. The dashboard path still - // needs HTTP to adopt the served token. - if (announcedToken) { - token = announcedToken - } else { - const baseUrl = `http://127.0.0.1:${port}` - await Promise.race([waitForHermes(baseUrl, token), startFailed]) - } + const baseUrl = `http://127.0.0.1:${port}` + await Promise.race([waitForHermes(baseUrl, token), startFailed]) ready = true - let authToken - if (announcedToken) { - authToken = announcedToken - } else { - authToken = await adoptServedDashboardToken(`http://127.0.0.1:${port}`, token, { - childAlive: () => child.exitCode === null && !child.killed, - label: `Hermes backend for profile "${profile}"`, - rememberLog - }) - } + const authToken = await adoptServedDashboardToken(baseUrl, token, { + childAlive: () => child.exitCode === null && !child.killed, + label: `Hermes backend for profile "${profile}"`, + rememberLog + }) entry.token = authToken // Verify the WebSocket session token before declaring backend ready. - // The slim server listens on `/` (bare WS); the dashboard listens on - // `/api/ws`. Route accordingly. - const wsPath = announcedToken ? '/' : '/api/ws' - const wsUrl = `ws://127.0.0.1:${port}${wsPath}?token=${encodeURIComponent(authToken)}` + // HTTP /api/status can pass while WS auth fails (separate transport, separate guards). + const wsUrl = `ws://127.0.0.1:${port}/api/ws?token=${encodeURIComponent(authToken)}` const wsProbe = await probeGatewayWebSocket(wsUrl, { WebSocketImpl: globalThis.WebSocket }) if (!wsProbe.ok) { throw new Error( - `Hermes backend for profile "${profile}" WebSocket rejected the session token: ${wsProbe.reason}` + `Hermes backend for profile "${profile}" is HTTP-reachable but the WebSocket (/api/ws) rejected the session token: ${wsProbe.reason}` ) } - // baseUrl feeds the `hermes:api` REST plane (fetchJson rejects non-http). - // A WS-only backend has no HTTP routes; expose the WS transport via wsUrl - // and mark it on the descriptor instead of overloading baseUrl (F1). - const baseUrl = `http://127.0.0.1:${port}` - const wsOnlyTransport = Boolean(announcedToken) - return { baseUrl, mode: 'local', @@ -12219,7 +12145,6 @@ async function spawnPoolBackend(profile, entry, opts: { forceLocal?: boolean; po token: authToken, profile, wsUrl, - wsOnlyTransport, logs: hermesLog.slice(-80), ...getWindowState() } @@ -12431,11 +12356,7 @@ async function startHermes() { const token = crypto.randomBytes(32).toString('base64url') // --port 0: the OS assigns an ephemeral port; the child announces it on stdout. - // `--ws-only` opt-in — same gate as the pool spawn site above (F1). const backendArgs = ['serve', '--host', '127.0.0.1', '--port', '0'] - if (process.env.HERMES_DESKTOP_WS_ONLY === '1') { - backendArgs.push('--ws-only') - } // Pin the desktop's chosen profile via the global --profile flag. This is // deterministic (it wins over the sticky ~/.hermes/active_profile file) and // resolves HERMES_HOME the same way `hermes -p ` does on the CLI. An @@ -12622,49 +12543,36 @@ async function startHermes() { await advanceBootProgress('backend.port', 'Waiting for Hermes backend to launch', 86) // Discover the ephemeral port the child bound to - const announced = await Promise.race([ + const port = await Promise.race([ waitForDashboardPortAnnouncement(hermesProcess, { describeOutputTail: () => primaryOutputTail.describe(), readyFile }), backendStartFailed - ]) as { port: number; token?: string } - - const port = announced.port - const announcedToken = announced.token + ]) if (readyFile) { fs.unlink(readyFile, () => {}) } - // The --ws-only server has no HTTP routes — skip waitForHermes when we - // have a direct token from the announcement. The dashboard path still - // needs HTTP to adopt the served token. - let authToken - if (announcedToken) { - authToken = announcedToken - } else { - const baseUrl = `http://127.0.0.1:${port}` - await advanceBootProgress('backend.wait', 'Waiting for Hermes backend to become ready', 90) - await Promise.race([waitForHermes(baseUrl, token), backendStartFailed]) - authToken = await adoptServedDashboardToken(baseUrl, token, { - childAlive: () => hermesProcess.exitCode === null && !hermesProcess.killed, - rememberLog - }) - } + const baseUrl = `http://127.0.0.1:${port}` + await advanceBootProgress('backend.wait', 'Waiting for Hermes backend to become ready', 90) + await Promise.race([waitForHermes(baseUrl, token), backendStartFailed]) backendReady = true backendStartFailure = null + const authToken = await adoptServedDashboardToken(baseUrl, token, { + childAlive: () => hermesProcess.exitCode === null && !hermesProcess.killed, + rememberLog + }) + // Verify the WebSocket session token before declaring backend ready. - // The slim server listens on `/` (bare WS); the dashboard listens on - // `/api/ws`. Route accordingly. - const wsPath = announcedToken ? '/' : '/api/ws' - const wsUrl = `ws://127.0.0.1:${port}${wsPath}?token=${encodeURIComponent(authToken)}` + const wsUrl = `ws://127.0.0.1:${port}/api/ws?token=${encodeURIComponent(authToken)}` const wsProbe = await probeGatewayWebSocket(wsUrl, { WebSocketImpl: globalThis.WebSocket }) if (!wsProbe.ok) { throw new Error( - `Local Hermes backend WebSocket rejected the session token: ${wsProbe.reason}` + `Local Hermes backend is HTTP-reachable but the WebSocket (/api/ws) rejected the session token: ${wsProbe.reason}` ) } @@ -12683,11 +12591,6 @@ async function startHermes() { // accumulated count of the resolved episode. bootstrapRepairAttempt = 0 - // Same descriptor rule as spawnPoolBackend: baseUrl stays http for the - // REST plane; WS-only mode is flagged separately (F1). - const baseUrl = `http://127.0.0.1:${port}` - const wsOnlyTransport = Boolean(announcedToken) - return { baseUrl, mode: 'local', @@ -12695,7 +12598,6 @@ async function startHermes() { authMode: 'token', token: authToken, wsUrl, - wsOnlyTransport, logs: hermesLog.slice(-80), ...getWindowState() } diff --git a/apps/shared/src/json-rpc-gateway-replay.test.ts b/apps/shared/src/json-rpc-gateway-replay.test.ts index af1b58f20f..a587573830 100644 --- a/apps/shared/src/json-rpc-gateway-replay.test.ts +++ b/apps/shared/src/json-rpc-gateway-replay.test.ts @@ -41,6 +41,7 @@ class FakeWebSocket extends EventTarget { lastRequest(): { id: string; method: string; params: Record } { const last = this.sent[this.sent.length - 1] + return JSON.parse(last ?? '{}') } } @@ -54,6 +55,7 @@ const makeClient = () => { heartbeatDeadlineMs: 0, connectTimeoutMs: 1000 }) + return client } @@ -193,16 +195,58 @@ describe('JsonRpcGatewayClient event-seq tracking + replay resume', () => { client.close() }) - it('does not re-dispatch replayed events already applied live during the replay window', async () => { + it('rejects envelope-shaped replay elements (the #94219 server-shape bug)', async () => { const client = makeClient() - const seen: number[] = [] - client.on('tool.complete', e => seen.push((e.payload as { n: number }).n)) + const seen: string[] = [] + client.on('message.delta', () => seen.push('delta')) const first = client.connect('ws://x') let sock = sockets[sockets.length - 1] sock.open() await first - sock.serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'tool.complete', session_id: 's1', seq: 3, payload: { n: 3 } } }) + sock.serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: 's1', seq: 1 } }) + expect(seen).toEqual(['delta']) // the pre-drop live frame + + client.invalidate('drop') + const second = client.connect('ws://x') + sock = sockets[sockets.length - 1] + sock.open() + await second + + await vi.waitFor(() => { + expect(sock.lastRequest().method).toBe('session.events.since') + }) + const req = sock.lastRequest() + // Pre-fix servers returned FULL JSON-RPC envelopes. The client must not + // dispatch those blindly — and this documents why the server now sends + // bare event objects. + sock.serverFrame({ + jsonrpc: '2.0', + id: req.id, + result: { + events: [{ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: 's1', seq: 2 } }], + latest_seq: 2, + truncated: false, + count: 1 + } + }) + await Promise.resolve() + // Envelope-shaped replay elements must add nothing beyond the live frame. + expect(seen).toEqual(['delta']) + client.close() + }) + + it('holds live frames racing the replay fetch — no double dispatch, no skipped gap', async () => { + const client = makeClient() + const seen: number[] = [] + client.on('message.delta', e => seen.push((e as unknown as { seq: number }).seq)) + + const first = client.connect('ws://x') + let sock = sockets[sockets.length - 1] + sock.open() + await first + sock.serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: 's1', seq: 2 } }) + expect(seen).toEqual([2]) // pre-drop live frame dispatches normally client.invalidate('drop') const second = client.connect('ws://x') @@ -214,41 +258,56 @@ describe('JsonRpcGatewayClient event-seq tracking + replay resume', () => { expect(sock.lastRequest().method).toBe('session.events.since') }) - // A live frame lands WHILE the replay RPC is in flight (watermark → 5)… - sock.serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'tool.complete', session_id: 's1', seq: 5, payload: { n: 5 } } }) + // LIVE frames 5 and 6 arrive while the replay (which carries 3,4,5) is + // still in flight. They must be parked, not dispatched ahead of the gap. + sock.serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: 's1', seq: 5 } }) + sock.serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: 's1', seq: 6 } }) + expect(seen).toEqual([2]) // still only the pre-drop frame — 5/6 parked - // …then the replay response arrives containing that same event (seq 5, - // emitted after our last_seen=3) plus one genuinely missed event (seq 4). const req = sock.lastRequest() sock.serverFrame({ jsonrpc: '2.0', id: req.id, result: { events: [ - { type: 'tool.complete', session_id: 's1', seq: 4, payload: { n: 4 } }, - { type: 'tool.complete', session_id: 's1', seq: 5, payload: { n: 5 } } + { type: 'message.delta', session_id: 's1', seq: 3 }, + { type: 'message.delta', session_id: 's1', seq: 4 }, + { type: 'message.delta', session_id: 's1', seq: 5 } ], latest_seq: 5, truncated: false, - count: 2 + count: 3 } }) - await Promise.resolve() - // seq 5 was already applied live — replay must NOT double-dispatch it. - expect(seen).toEqual([3, 5, 4]) + // In-order, exactly once: replayed 3,4,5 then the parked live 6 — + // the parked duplicate of 5 is seq-gated out. + await vi.waitFor(() => { + expect(seen).toEqual([2, 3, 4, 5, 6]) + }) + expect(client.getSeqWatermarks().s1).toBe(6) client.close() }) - it('realigns the watermark when the server reports a seq epoch reset', async () => { + it('clears stale watermarks when the backend epoch changes (restart poisoning)', async () => { const client = makeClient() + const first = client.connect('ws://x') let sock = sockets[sockets.length - 1] sock.open() await first - // Client observed a long-lived session (watermark 97) before the drop. + // Learn epoch A and a high watermark. + sock.serverFrame({ + jsonrpc: '2.0', + method: 'event', + params: { type: 'gateway.ready', payload: { replay_epoch: 'epoch-A' } } + }) sock.serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: 's1', seq: 97 } }) + expect(client.getSeqWatermarks()).toEqual({ s1: 97 }) + // Backend restarts: reconnect, replay under a NEW epoch returns nothing + // (fresh process, empty ring) — pre-fix the client kept watermark 97 and + // silently believed it missed nothing, forever. client.invalidate('drop') const second = client.connect('ws://x') sock = sockets[sockets.length - 1] @@ -258,60 +317,20 @@ describe('JsonRpcGatewayClient event-seq tracking + replay resume', () => { await vi.waitFor(() => { expect(sock.lastRequest().method).toBe('session.events.since') }) - // Gateway restarted: its counter is behind us. truncated + latest_seq < last_seen. const req = sock.lastRequest() sock.serverFrame({ jsonrpc: '2.0', id: req.id, - result: { events: [], latest_seq: 3, truncated: true, count: 0 } + result: { events: [], latest_seq: 0, truncated: false, count: 0, epoch: 'epoch-B' } }) + await vi.waitFor(() => { - // Watermark re-adopts the server's epoch so future replays work again. - expect(client.getSeqWatermarks().s1).toBe(3) + expect(client.getSeqWatermarks()).toEqual({}) }) - // Next live frame in the new epoch advances normally. - sock.serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: 's1', seq: 4 } }) - expect(client.getSeqWatermarks().s1).toBe(4) - client.close() - }) - - it('resets seq watermarks when gateway.ready arrives with a NEW epoch', async () => { - const client = makeClient() - const p = client.connect('ws://x') - sockets[0].open() - await p - - // First connect: epoch A adopted, watermarks accumulate. - sockets[0].serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'gateway.ready', payload: { epoch: 'aaaa1111' } } }) - sockets[0].serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.complete', session_id: 's1', seq: 40 } }) - expect(client.getSeqWatermarks()).toEqual({ s1: 40 }) - - // Same epoch re-announced (same process, reconnect): watermarks KEPT. - sockets[0].serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'gateway.ready', payload: { epoch: 'aaaa1111' } } }) - expect(client.getSeqWatermarks()).toEqual({ s1: 40 }) - - // New epoch (gateway restarted, seq namespace reset): stale watermark 40 - // would be AHEAD of the new counter and suppress replay forever — must - // be dropped. - sockets[0].serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'gateway.ready', payload: { epoch: 'bbbb2222' } } }) - expect(client.getSeqWatermarks()).toEqual({}) - - // Fresh seqs in the new namespace accumulate normally. - sockets[0].serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.start', session_id: 's1', seq: 1 } }) - expect(client.getSeqWatermarks()).toEqual({ s1: 1 }) - client.close() - }) - - it('ignores gateway.ready without an epoch (legacy backend)', async () => { - const client = makeClient() - const p = client.connect('ws://x') - sockets[0].open() - await p - - sockets[0].serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.complete', session_id: 's1', seq: 7 } }) - sockets[0].serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'gateway.ready', payload: { skin: {} } } }) - expect(client.getSeqWatermarks()).toEqual({ s1: 7 }) + // New-epoch events build fresh watermarks from scratch. + sock.serverFrame({ jsonrpc: '2.0', method: 'event', params: { type: 'message.delta', session_id: 's1', seq: 3 } }) + expect(client.getSeqWatermarks()).toEqual({ s1: 3 }) client.close() }) }) diff --git a/apps/shared/src/json-rpc-gateway.ts b/apps/shared/src/json-rpc-gateway.ts index 50502bf4f0..406e471440 100644 --- a/apps/shared/src/json-rpc-gateway.ts +++ b/apps/shared/src/json-rpc-gateway.ts @@ -111,12 +111,22 @@ export class JsonRpcGatewayClient { private lastSeenSeq = new Map() /** Set while a post-reconnect replay fetch is in flight (dedup guard). */ private replayInFlight = false - /** Server boot epoch from gateway.ready — seq watermarks are only valid - * within one epoch; a change (gateway restart) resets them. */ - private serverEpoch: string | null = null - /** Seqs dispatched LIVE while a replay RPC is in flight, per session — - * the replay response overlaps with these and must not re-dispatch them. */ - private liveSeqsDuringReplay: Map> | null = null + /** + * While a replay fetch is in flight, live seq'd frames for the sessions + * being replayed are parked here instead of dispatching immediately. + * Without this hold, a live frame racing the replay response is dispatched + * twice (once live, once when the replay returns the same seq) or, worse, + * advances the watermark so the gap events the replay carries get skipped. + */ + private replayHold: Map | null = null + /** + * Server process identity for the replay contract (from gateway.ready / + * session.events.since). Seq counters are in-process on the backend, so a + * restart resets them while we still hold high watermarks — without this + * check events_since(sid, 97) returns [] + truncated=false forever and we + * silently believe nothing was missed. + */ + private replayEpoch: string | null = null private readonly eventHandlers = new Map void>>() private readonly stateHandlers = new Set<(state: ConnectionState) => void>() private readonly options: Required> & @@ -461,45 +471,25 @@ export class JsonRpcGatewayClient { } } - // Seq-namespace epoch: a new epoch means the gateway restarted and - // its per-session seq counters reset. Our stored watermarks are from - // the previous namespace — a stale HIGH watermark would make every - // replay return empty ("client ahead") and can suppress gap - // detection forever. Drop all watermarks so this connection starts - // fresh; the app layer re-hydrates state via session.resume anyway. - const payload = frame.params.payload as { epoch?: unknown } | undefined - const epoch = typeof payload?.epoch === 'string' ? payload.epoch : null + const epoch = (frame.params.payload as { replay_epoch?: unknown } | undefined)?.replay_epoch - if (epoch) { - if (this.serverEpoch && this.serverEpoch !== epoch) { - this.lastSeenSeq.clear() - } - - this.serverEpoch = epoch + if (typeof epoch === 'string' && epoch) { + this.adoptReplayEpoch(epoch) } } + const sid = frame.params.session_id + const seqValue = (frame.params as { seq?: unknown }).seq + + if (this.replayHold && sid && typeof seqValue === 'number' && this.replayHold.has(sid)) { + // Replay in flight for this session: park the frame; flushReplayHold + // dispatches it after the replayed gap, gated on seq. + this.replayHold.get(sid)?.push(frame.params) + + return + } + this.recordSeq(frame.params) - - // While a replay RPC is in flight, remember which seqs arrived live — - // the replay response overlaps with them (server returns everything - // > our pre-replay watermark) and must not re-dispatch those. - if (this.liveSeqsDuringReplay) { - const sid = frame.params.session_id - const seq = (frame.params as { seq?: unknown }).seq - - if (sid && typeof seq === 'number') { - let set = this.liveSeqsDuringReplay.get(sid) - - if (!set) { - set = new Set() - this.liveSeqsDuringReplay.set(sid, set) - } - - set.add(seq) - } - } - this.dispatchEvent(frame.params) } } @@ -531,11 +521,8 @@ export class JsonRpcGatewayClient { /** * After a reconnect, ask the gateway to replay every event newer than our * per-session watermarks. Replayed frames go through the SAME dispatchEvent - * path as live frames. Frames that arrived LIVE while the replay RPC was in - * flight are tracked and skipped here — the server returns everything past - * the pre-replay watermark, so those overlap, and re-dispatching them would - * double-append streamed text (message.delta has no identity; it's - * append-only downstream). + * path as live frames — dedupe happens naturally because recordSeq ignores + * non-increasing seqs and downstream stores key on event identity. * Best-effort: failures are swallowed (the next reconnect retries). */ private async fetchReplay(): Promise { @@ -544,61 +531,118 @@ export class JsonRpcGatewayClient { } this.replayInFlight = true - this.liveSeqsDuringReplay = new Map() + // Park live frames for the sessions we're about to replay so a frame + // racing the replay response can't dispatch ahead of (or duplicate) the + // gap events. Sessions without watermarks are unaffected. + const hold = new Map() + + for (const sid of this.lastSeenSeq.keys()) { + hold.set(sid, []) + } + + this.replayHold = hold try { const entries = Object.entries(this.getSeqWatermarks()) + // One RPC per known session keeps params flat; sessions are few (<20). - await Promise.allSettled( - entries.map(async ([sid, lastSeen]) => { - const result = await this.request<{ - events?: Array<{ type: string; session_id?: string; seq?: number; payload?: unknown }> - latest_seq?: number - truncated?: boolean - }>('session.events.since', { session_id: sid, last_seen: lastSeen }, REPLAY_REQUEST_TIMEOUT_MS) - - // Seq epoch reset (gateway restart / server-side eviction): the - // server's counter is now BEHIND our watermark, so replay could - // never deliver again. Re-adopt the server's epoch so future - // reconnects work; the gap itself is unrecoverable (truncated). - const latest = typeof result?.latest_seq === 'number' ? result.latest_seq : 0 - - if (result?.truncated && latest < lastSeen) { - if (latest > 0) { - this.lastSeenSeq.set(sid, latest) - } else { - this.lastSeenSeq.delete(sid) - } - - return - } - - if (!Array.isArray(result?.events)) { - return - } - - const liveSeqs = this.liveSeqsDuringReplay?.get(sid) - - for (const event of result.events) { - if (!event?.type) { - continue - } - - // Skip events already dispatched live during the replay window. - if (typeof event.seq === 'number' && liveSeqs?.has(event.seq)) { - continue - } - - this.recordSeq(event as GatewayEvent) - this.dispatchEvent(event as GatewayEvent) - } - }) + const results = await Promise.allSettled( + entries.map(([sid, lastSeen]) => + this.request<{ events?: Array<{ type: string; session_id?: string; seq?: number; payload?: unknown }> }>( + 'session.events.since', + { session_id: sid, last_seen: lastSeen }, + REPLAY_REQUEST_TIMEOUT_MS + ) + ) ) + + for (const result of results) { + if (result.status !== 'fulfilled' || !Array.isArray(result.value?.events)) { + continue + } + + const epoch = (result.value as { epoch?: unknown }).epoch + + if (typeof epoch === 'string' && epoch && this.replayEpoch && epoch !== this.replayEpoch) { + // Backend restarted: its seq numbering reset, so our watermarks — + // and this replay window — are meaningless. Drop them and start + // fresh under the new epoch. + this.adoptReplayEpoch(epoch) + + continue + } + + if (typeof epoch === 'string' && epoch && !this.replayEpoch) { + this.replayEpoch = epoch + } + + for (const event of result.value.events) { + if (!event?.type) { + continue + } + + this.dispatchIfNewer(event as GatewayEvent) + } + } } catch { // Replay is an optimization over lossy-reconnect; never surface errors. } finally { + this.flushReplayHold() this.replayInFlight = false - this.liveSeqsDuringReplay = null + } + } + + /** + * Dispatch an event only when its seq advances the session watermark. + * Seq-less events always dispatch (no ordering contract to violate). + */ + private dispatchIfNewer(event: GatewayEvent): void { + const sid = event.session_id + const seq = (event as { seq?: unknown }).seq + + if (sid && typeof seq === 'number' && Number.isFinite(seq)) { + const prev = this.lastSeenSeq.get(sid) ?? 0 + + if (seq <= prev) { + return + } + + this.lastSeenSeq.set(sid, seq) + } + + this.dispatchEvent(event) + } + + /** + * Record the server's replay epoch; on change (backend restart) the old + * seq watermarks describe a numbering that no longer exists — clear them + * so the next reconnect doesn't silently believe it missed nothing. + */ + private adoptReplayEpoch(epoch: string): void { + if (this.replayEpoch === epoch) { + return + } + + if (this.replayEpoch !== null) { + this.lastSeenSeq.clear() + } + + this.replayEpoch = epoch + } + + /** Release frames parked during a replay fetch, seq-gated against dupes. */ + private flushReplayHold(): void { + const hold = this.replayHold + this.replayHold = null + + if (!hold) { + return + } + + for (const parked of hold.values()) { + for (const event of parked) { + this.dispatchIfNewer(event) + } } } diff --git a/hermes_cli/main.py b/hermes_cli/main.py index 738b765f0c..778181c2eb 100644 --- a/hermes_cli/main.py +++ b/hermes_cli/main.py @@ -11970,29 +11970,6 @@ def cmd_dashboard(args): if _token_file and not _headless_backend: raise SystemExit("--ssh-session-token-file is only valid with hermes serve") - # ── Slim WS-only path ──────────────────────────────────────────── - # `--ws-only` skips the entire dashboard boot (FastAPI app, uvicorn, - # SPA build, auth gate, profile routing) and runs the bare WebSocket - # server directly. The desktop app opts into this to cut the - # FastAPI/uvicorn layer from the boot path — same dispatch surface, - # same RPCs, same events, zero HTTP framework. - if getattr(args, "ws_only", False): - from hermes_cli.resource_limits import apply_nofile_soft_limit - - apply_nofile_soft_limit() - # Resolve SSH session token (serve-only, same as the dashboard path). - if _token_file: - try: - _ssh_session_token = Path(_token_file).read_text(encoding="utf-8").strip() - except Exception: - pass - if _ssh_session_token: - os.environ["HERMES_DASHBOARD_SESSION_TOKEN"] = _ssh_session_token - from tui_gateway.entry_ws import run as run_ws_only - - run_ws_only(host=args.host or "127.0.0.1", port=args.port or 0) - return # run() blocks; returns only on shutdown - # ── Sanitize Desktop-inherited env that hijacks a standalone launch ─ # Desktop Electron spawns its backend with HERMES_DESKTOP=1 plus # HERMES_WEB_DIST= (and often diff --git a/hermes_cli/subcommands/dashboard.py b/hermes_cli/subcommands/dashboard.py index 6988f0f87f..0b695e076a 100644 --- a/hermes_cli/subcommands/dashboard.py +++ b/hermes_cli/subcommands/dashboard.py @@ -167,18 +167,6 @@ def build_dashboard_parser( # JSON-RPC/WS, so `serve` skips the web UI build AND never serves the SPA # (cmd_dashboard exports HERMES_SERVE_HEADLESS=1). `dashboard` leaves it # unset and serves the browser UI as before. - # `--ws-only` bypasses the FastAPI/uvicorn dashboard entirely: a bare - # `websockets` server that calls `handle_ws` directly. This is the lean - # desktop path — no HTTP framework, no SPA, no route resolution, no - # middleware stack. The desktop app opts into it; human callers get the - # full `serve` (which is still headless, just through uvicorn). - serve_parser.add_argument( - "--ws-only", - dest="ws_only", - action="store_true", - default=False, - help=argparse.SUPPRESS, - ) serve_parser.set_defaults(func=cmd_dashboard, no_open=True, headless_backend=True) # `hermes dashboard register` — register a self-hosted dashboard OAuth diff --git a/tests/test_tui_gateway_entry_ws.py b/tests/test_tui_gateway_entry_ws.py deleted file mode 100644 index d139a45286..0000000000 --- a/tests/test_tui_gateway_entry_ws.py +++ /dev/null @@ -1,118 +0,0 @@ -"""Tests for tui_gateway.entry_ws — the slim WS-only gateway server. - -The auth/path handling MUST be exercised against the real, repo-pinned -``websockets`` package (15.x): its asyncio server passes a single -``ServerConnection`` to the handler (no second ``path`` argument), and the -request path lives on ``connection.request.path``. A fake adapter would have -masked exactly the signature mismatch that review F2 on PR #94245 flagged. -""" - -from __future__ import annotations - -import asyncio -import contextlib -import threading - -import pytest - -websockets = pytest.importorskip("websockets") - -from tui_gateway import entry_ws - - -@pytest.fixture() -def ws_server(): - """Real websockets server running entry_ws._handle_connection on loopback. - - Runs in a dedicated event-loop thread so tests can drive real client - connections with the synchronous websockets client API. - """ - loop = asyncio.new_event_loop() - started = threading.Event() - state: dict = {} - - async def _serve(): - server = await websockets.serve( - entry_ws._handle_connection, "127.0.0.1", 0 - ) - state["port"] = server.sockets[0].getsockname()[1] - state["server"] = server - started.set() - await asyncio.Event().wait() - - def _run(): - asyncio.set_event_loop(loop) - with contextlib.suppress(asyncio.CancelledError): - loop.run_until_complete(_serve()) - - thread = threading.Thread(target=_run, daemon=True) - thread.start() - assert started.wait(timeout=10), "ws server failed to start" - yield state - loop.call_soon_threadsafe(loop.stop) - thread.join(timeout=5) - - -def _connect(port: int, query: str): - """Open a real client connection; returns (ok, close_code). - - Deterministic accept/reject probe: an ACCEPTED connection is handed to - handle_ws, which emits the ``gateway.ready`` event; a REJECTED one is - closed with 4401. Waiting on recv() observes whichever happens — no race - against the close frame (a ping can win that race and false-pass). - """ - from websockets.sync.client import connect as sync_connect - - uri = f"ws://127.0.0.1:{port}/{query}" - try: - with sync_connect(uri, open_timeout=5, close_timeout=5) as client: - first = client.recv(timeout=10) - return "gateway.ready" in str(first), None - except Exception as exc: # ConnectionClosed carries the close code - code = getattr(getattr(exc, "rcvd", None), "code", None) - return False, code - - -def test_handshake_rejects_missing_token(ws_server): - ok, code = _connect(ws_server["port"], "") - assert not ok - assert code == 4401 - - -def test_handshake_rejects_wrong_token(ws_server): - ok, code = _connect(ws_server["port"], "?token=definitely-wrong") - assert not ok - assert code == 4401 - - -def test_handshake_accepts_valid_token_reading_path_from_v15_connection(ws_server): - """The regression F2 guards against: under websockets 15.x the handler - receives only a ServerConnection — the ?token= query MUST still be read - (from ws.request.path). If the handler fell back to path='/', this valid - token would be rejected exactly like the missing-token case above.""" - token = entry_ws.get_session_token() - ok, code = _connect(ws_server["port"], f"?token={token}") - assert ok, f"valid token rejected (close code {code}) — path not read from v15 connection" - - -def test_run_refuses_non_loopback_host(): - """F3: --ws-only uses static-token auth and must fail closed off-loopback.""" - with pytest.raises(SystemExit, match="loopback"): - entry_ws.run(host="0.0.0.0", port=0) - with pytest.raises(SystemExit, match="loopback"): - entry_ws.run(host="192.168.1.10", port=0) - - -def test_run_allows_loopback_hosts_past_the_guard(monkeypatch): - """Loopback names/addresses pass the guard (asyncio.run is stubbed so the - server itself never starts).""" - sentinel = RuntimeError("guard passed — reached server start") - - def _boom(_coro): - _coro.close() - raise sentinel - - monkeypatch.setattr(entry_ws.asyncio, "run", _boom) - for host in ("127.0.0.1", "::1", "localhost"): - with pytest.raises(RuntimeError, match="guard passed"): - entry_ws.run(host=host, port=0) diff --git a/tests/test_tui_gateway_event_replay.py b/tests/test_tui_gateway_event_replay.py index f0b434dd81..5e431c29dd 100644 --- a/tests/test_tui_gateway_event_replay.py +++ b/tests/test_tui_gateway_event_replay.py @@ -6,11 +6,10 @@ import pytest from tui_gateway import event_replay from tui_gateway.event_replay import ( - events_since, latest_seq, - replay_stats, reset_replay_state, - stamp_event, + events_since, + replay_stats, ) @@ -21,9 +20,7 @@ def _clean(): reset_replay_state() -def _frame(sid, etype="status.update"): - # status.update is a DURABLE event type (buffered for replay). - # message.delta / thinking.delta are transient — stamped, never buffered. +def _frame(sid, etype="message.delta"): return { "jsonrpc": "2.0", "method": "event", @@ -36,9 +33,9 @@ def test_stamp_adds_monotonic_seq_per_session(): f2 = _frame("s1") other = _frame("s2") - stamp_event(f1) - stamp_event(other) - stamp_event(f2) + event_replay._stamp_event(f1) + event_replay._stamp_event(other) + event_replay._stamp_event(f2) assert f1["params"]["seq"] == 1 assert f2["params"]["seq"] == 2 # per-session counter, unaffected by s2 @@ -49,164 +46,70 @@ def test_stamp_ignores_non_event_and_sessionless_frames(): rpc = {"jsonrpc": "2.0", "id": 1, "result": {}} no_sid = {"jsonrpc": "2.0", "method": "event", "params": {"type": "skin.changed"}} - stamp_event(rpc) - stamp_event(no_sid) + event_replay._stamp_event(rpc) + event_replay._stamp_event(no_sid) assert "seq" not in rpc assert "seq" not in no_sid["params"] assert replay_stats()["events"] == 0 -def test_events_since_returns_bare_params_only_newer_in_order(): +def test_events_since_returns_only_newer_frames_in_order(): frames = [_frame("s1") for _ in range(5)] for f in frames: - stamp_event(f) + event_replay._stamp_event(f) - got, latest, truncated = events_since("s1", 3) + got = events_since("s1", 3) assert [e["seq"] for e in got] == [4, 5] - # Replay returns the bare event params (what live dispatch sees), NOT the - # full JSON-RPC frame envelope — the client reads event.type at top level. - assert all("jsonrpc" not in e and e["type"] == "status.update" for e in got) - assert latest == 5 - assert truncated is False - - all_events, _, _ = events_since("s1", 0) - assert all_events == [f["params"] for f in frames] - assert events_since("s1", 5) == ([], 5, False) + assert events_since("s1", 0) == [f["params"] for f in frames] + assert events_since("s1", 99) == [] assert latest_seq("s1") == 5 +def test_events_since_returns_client_dispatchable_event_objects(): + """Cross-language contract: the client's replay loop dispatches an element + only when it has a TOP-LEVEL ``type`` (json-rpc-gateway.ts fetchReplay: + ``if (!event?.type) continue``). Returning full JSON-RPC envelopes here + makes every replayed event silently droppable — the original #94219 bug. + """ + event_replay._stamp_event(_frame("s1")) + (event,) = events_since("s1", 0) + + # Bare event object, not an envelope. + assert event["type"] == "message.delta" + assert event["session_id"] == "s1" + assert event["seq"] == 1 + assert "jsonrpc" not in event + assert "method" not in event + assert "params" not in event + + def test_unknown_session_returns_empty(): - assert events_since("nope", 0) == ([], 0, False) + assert events_since("nope", 0) == [] assert latest_seq("nope") == 0 -def test_transient_deltas_stamped_but_not_buffered(): - """Streaming token deltas get seqs (live wire ordering) but never enter - the replay ring — one streaming turn must not evict durable control - events (message.start/complete, session.info) from replay coverage.""" - start = _frame("s1", "message.start") - deltas = [_frame("s1", "message.delta") for _ in range(100)] - thinking = [_frame("s1", "thinking.delta") for _ in range(50)] - complete = _frame("s1", "message.complete") - - stamp_event(start) - for f in deltas: - stamp_event(f) - for f in thinking: - stamp_event(f) - stamp_event(complete) - - # All frames got seqs, in one monotonic namespace. - assert start["params"]["seq"] == 1 - assert complete["params"]["seq"] == 152 - assert deltas[0]["params"]["seq"] == 2 - - # But only the 2 durable frames are buffered. - assert replay_stats()["events"] == 2 - - # Replay from 0 returns just the durable frames — and the delta-only - # gaps between them are NOT truncation (nothing recoverable was lost). - got, latest, truncated = events_since("s1", 0) - assert [e["type"] for e in got] == ["message.start", "message.complete"] - assert latest == 152 - assert truncated is False - - # A client that saw the whole live stream is fully covered too. - assert events_since("s1", 152) == ([], 152, False) - - -def test_ring_buffer_is_bounded_and_reports_truncation(): - for _ in range(event_replay._REPLAY_BUFFER_MAX + 50): - stamp_event(_frame("s1")) +def test_ring_buffer_is_bounded(): + for i in range(event_replay._REPLAY_BUFFER_MAX + 50): + event_replay._stamp_event(_frame("s1")) stats = replay_stats() assert stats["events"] == event_replay._REPLAY_BUFFER_MAX - - # Client that saw nothing (last_seen=0) has a gap older than the ring — - # durable frames were evicted, so the gap is real truncation. - got, latest, truncated = events_since("s1", 0) - assert truncated is True - assert latest == event_replay._REPLAY_BUFFER_MAX + 50 - assert len(got) == event_replay._REPLAY_BUFFER_MAX - - # Client aligned with the buffer start: fully covered, not truncated. - oldest = got[0]["seq"] - covered, _, covered_truncated = events_since("s1", oldest - 1) - assert covered_truncated is False - assert len(covered) == event_replay._REPLAY_BUFFER_MAX + # Oldest evicted: last_seen=0 must report truncation via the RPC contract. + buf = event_replay._replay_buffers["s1"] + assert buf[0][0] > 1 -def test_eviction_of_durable_frames_is_precise(): - """Truncation is keyed to the highest EVICTED durable seq, not the buffer - floor: a client whose last_seen covers everything evicted is not - truncated even when younger frames were dropped for other clients.""" - overflow = 50 - for _ in range(event_replay._REPLAY_BUFFER_MAX + overflow): - stamp_event(_frame("s1")) - - # The first `overflow` durable frames (seq 1..50) were evicted. - # A client at last_seen=overflow saw all of them — not truncated. - _, _, at_boundary = events_since("s1", overflow) - assert at_boundary is False - - # A client one behind the boundary lost seq=overflow — truncated. - _, _, behind = events_since("s1", overflow - 1) - assert behind is True - - -def test_epoch_reset_reports_truncated(): - """Client watermark AHEAD of the server (gateway restart) must be flagged. - - Without this, a restarted server returns [] / truncated=False and the - client's stuck watermark silently kills replay forever. - """ - for _ in range(3): - stamp_event(_frame("s1")) - - got, latest, truncated = events_since("s1", 97) - assert got == [] - assert latest == 3 - assert truncated is True - - # Session the server has never seen but the client has a watermark for. - assert events_since("gone", 42) == ([], 0, True) - - -def test_epoch_constant_is_stable_and_shaped(): - """EPOCH identifies this process's seq namespace: 8 hex chars, constant - for the process lifetime (clients compare it across reconnects).""" - assert isinstance(event_replay.EPOCH, str) - assert len(event_replay.EPOCH) == 8 - int(event_replay.EPOCH, 16) # hex - assert event_replay.EPOCH == event_replay.EPOCH # stable reference - - -def test_session_count_bounded_with_lru_eviction(): +def test_session_count_bounded_with_fifo_eviction(): for i in range(event_replay._REPLAY_SESSIONS_MAX + 10): - stamp_event(_frame(f"s{i}")) + event_replay._stamp_event(_frame(f"s{i}")) stats = replay_stats() assert stats["sessions"] == event_replay._REPLAY_SESSIONS_MAX - assert events_since("s0", 0)[0] == [] # oldest session fully evicted + assert events_since("s0", 0) == [] # oldest session fully evicted assert latest_seq(f"s{event_replay._REPLAY_SESSIONS_MAX + 9}") == 1 -def test_active_session_survives_eviction_lru(): - """Eviction is least-recently-ACTIVE, not first-created: a session that - keeps streaming must outlive idle sessions created after it.""" - stamp_event(_frame("active")) - for i in range(event_replay._REPLAY_SESSIONS_MAX - 1): - stamp_event(_frame(f"idle{i}")) - - # "active" is now the oldest by creation. Touch it, then overflow. - stamp_event(_frame("active")) - stamp_event(_frame("newcomer")) - - assert latest_seq("active") == 2 # survived — it was most recently active - assert events_since("idle0", 0) == ([], 0, False) # idle0 evicted instead - - def test_concurrent_stamping_never_drops_or_duplicates_seq(): errors = [] @@ -215,7 +118,7 @@ def test_concurrent_stamping_never_drops_or_duplicates_seq(): seen = set() for _ in range(200): f = _frame(sid) - stamp_event(f) + event_replay._stamp_event(f) seq = f["params"]["seq"] assert seq not in seen seen.add(seq) @@ -230,3 +133,22 @@ def test_concurrent_stamping_never_drops_or_duplicates_seq(): assert not errors assert replay_stats()["events"] == 8 * 200 + + +def test_truncation_detection_semantics(): + """The RPC handler's truncated flag: gap between last_seen and buffer start.""" + # Overflow the ring so the oldest events are genuinely evicted. + for _ in range(event_replay._REPLAY_BUFFER_MAX + 10): + event_replay._stamp_event(_frame("s1")) + + with event_replay._replay_lock: + oldest = event_replay._replay_buffers["s1"][0][0] + + assert oldest > 1 # eviction happened + + # Client saw everything up to just before the buffer → NOT truncated. + assert not event_replay.is_truncated("s1", oldest - 1) + # Client saw seq 5, buffer starts later → truncated. + assert event_replay.is_truncated("s1", 5) + # Unknown session: nothing evicted, nothing truncated. + assert not event_replay.is_truncated("nope", 0) diff --git a/tui_gateway/entry_ws.py b/tui_gateway/entry_ws.py deleted file mode 100644 index d3232068a3..0000000000 --- a/tui_gateway/entry_ws.py +++ /dev/null @@ -1,260 +0,0 @@ -"""Slim WebSocket-only gateway server for desktop / headless clients. - -Eliminates the FastAPI/uvicorn/dashboard layer from the desktop boot path. -The desktop app spawns this instead of ``hermes serve`` when ``--ws-only`` is -passed; the renderer connects to the same ``handle_ws`` dispatch surface, -gets the same 158 RPC methods, the same event stream — without importing or -initializing a 19.8K-line web framework. - -Wire protocol is identical to the dashboard's ``/api/ws`` route: the URL -carries ``?token=`` for auth, then -newline-delimited JSON-RPC flows in both directions. ``gateway.ready`` is -emitted on connect, identical to the dashboard path. - -Why not reuse uvicorn? The desktop client is a single WS peer on loopback; -spinning up an ASGI server + HTTP router + middleware stack to serve one -WebSocket is the exact "nightmare of layers" this eliminates. The -``websockets`` library gives us a bare TCP listener with per-connection async -handlers — no HTTP framework, no route resolution, no SPA, no CORS. - -Auth: constant-time ``hmac.compare_digest`` against ``_SESSION_TOKEN`` -(read from ``HERMES_DASHBOARD_SESSION_TOKEN`` or auto-generated, matching -``hermes_cli.web_server._resolve_session_token``). The token is passed as a -``?token=`` query parameter, identical to the dashboard path the desktop -already uses. -""" - -from __future__ import annotations - -import asyncio -import hmac -import logging -import os -import secrets -import sys -from typing import Any -from urllib.parse import parse_qs, urlparse - -_log = logging.getLogger(__name__) - -# ── Token resolution (mirrors hermes_cli.web_server._resolve_session_token) ── - -def _resolve_session_token() -> str: - """Return the session token for WS auth. - - Matches the dashboard's resolution: an explicit - ``HERMES_DASHBOARD_SESSION_TOKEN`` env var wins; otherwise we generate - one so the server is never unauthenticated on loopback. The desktop's - Electron main process reads the token from the backend's stdout - sentinel (``HERMES_BACKEND_READY token=``) and passes it to the - renderer, which includes it in the WS URL as ``?token=``. - """ - return os.environ.get("HERMES_DASHBOARD_SESSION_TOKEN") or secrets.token_urlsafe(32) - - -_SESSION_TOKEN = _resolve_session_token() - - -def get_session_token() -> str: - """Expose the token for the entrypoint to print as a ready sentinel.""" - return _SESSION_TOKEN - - -# ── ASGI-compatible WebSocket shim ──────────────────────────────────────── -# -# handle_ws expects an object with: accept(), receive_text(), send_text(), -# close(), and a .scope dict. The `websockets` library's ServerConnection -# (protocol v13) provides recv() and send() but NOT accept()/receive_text()/ -# send_text() (those are starlette/FastAPI names). This thin adapter wraps -# the websockets connection to expose the interface handle_ws expects. - - -class _WSAdapter: - """Adapt a ``websockets`` connection to the interface ``handle_ws`` expects.""" - - def __init__(self, ws: Any, path: str = "/", query_string: str = "") -> None: - self._ws = ws - # starlette's WebSocket has a .scope dict; handle_ws reads - # ws.scope["extensions"]["transport"] for TCP_NODELAY + keepalive. - peer = ws.remote_address if hasattr(ws, "remote_address") else None - self.scope: dict[str, Any] = { - "type": "websocket", - "query_string": query_string, - "path": path, - "client": peer, - "extensions": {}, - } - # Expose the underlying socket for _disable_nagle's TCP_NODELAY. - try: - transport = ws.transport # websockets >=14 - self.scope["extensions"]["transport"] = transport - except AttributeError: - pass - - async def accept(self, subprotocol: str | None = None) -> None: - # The websockets library auto-accepts on connect; nothing to do. - # If a subprotocol is requested, it's handled at the handshake level. - pass - - async def receive_text(self) -> str: - data = await self._ws.recv() - if isinstance(data, bytes): - return data.decode("utf-8") - return str(data) - - async def send_text(self, text: str) -> None: - await self._ws.send(text) - - async def close(self, code: int = 1000, reason: str = "") -> None: - await self._ws.close(code, reason) - - -class _WebSocketDisconnect(Exception): - """Mirrors starlette's WebSocketDisconnect for handle_ws's except clause.""" - - def __init__(self, code: int = 1000, reason: str = "") -> None: - self.code = code - self.reason = reason - super().__init__(f"code={code} reason={reason}") - - -# ── Auth ────────────────────────────────────────────────────────────────── - -def _check_token(query_string: str) -> bool: - """Constant-time token check against the ``?token=`` query parameter.""" - if not _SESSION_TOKEN: - return False - params = parse_qs(query_string) - presented = params.get("token", [None])[0] - if not presented: - return False - return hmac.compare_digest(presented, _SESSION_TOKEN) - - -# ── Connection handler ─────────────────────────────────────────────────── - -async def _handle_connection(ws: Any) -> None: - """Per-connection handler for the websockets server. - - Auth → adapter → ``handle_ws`` (the same function the dashboard's - ``/api/ws`` route calls). Events flow identically. - """ - # websockets >=13 passes the path on the ws object, not as a second arg. - path = getattr(ws, "path", None) or getattr(getattr(ws, "request", None), "path", "/") - - parsed = urlparse(path) - query_string = parsed.query - - if not _check_token(query_string): - _log.warning("ws-only: rejecting connection — token mismatch") - await ws.close(code=4401, reason="unauthorized") - return - - # Install our disconnect type so handle_ws's except clause catches it. - import tui_gateway.ws as ws_mod - ws_mod._WebSocketDisconnect = _WebSocketDisconnect - - adapter = _WSAdapter(ws, path=path, query_string=query_string) - from tui_gateway.ws import handle_ws - - try: - await handle_ws(adapter) - except _WebSocketDisconnect: - pass - except Exception: - _log.exception("ws-only: connection handler crashed") - finally: - try: - await ws.close() - except Exception: - pass - - -# ── Server ──────────────────────────────────────────────────────────────── - -def run(host: str = "127.0.0.1", port: int = 0) -> None: - """Start the slim WS server. Blocks until interrupted. - - Prints ``HERMES_BACKEND_READY port= token=`` to stdout so - the desktop's Electron main process can discover the port (when - ``--port 0``) and the token (when auto-generated), identical to how it - parses the dashboard's ready sentinel. - - Loopback-only: this server authenticates with a static per-process - session token — the loopback shortcut the desktop uses. It deliberately - does NOT implement the dashboard's gated one-time-ticket auth - (``should_require_dashboard_auth``), so binding it to a non-loopback - interface would expose a weaker auth posture than every other Hermes - surface. Fail closed instead of serving. - """ - import ipaddress - import signal - - try: - addr = ipaddress.ip_address(host) - is_loopback = addr.is_loopback - except ValueError: - # Hostnames: allow the conventional loopback name only. - is_loopback = host == "localhost" - if not is_loopback: - raise SystemExit( - f"--ws-only only binds loopback interfaces (got {host!r}). " - "It uses the desktop's static session-token auth, which is not " - "hardened for network exposure. Use `hermes serve` (the full " - "dashboard server with gated auth) for non-loopback hosts." - ) - - # Capture the real stdout BEFORE importing tui_gateway.server — - # server.py redirects sys.stdout → sys.stderr at module level - # (line 392: `sys.stdout = sys.stderr`) to keep JSON-RPC over stdio - # clean. Our ready sentinel must go to the pipe the desktop reads. - _real_stdout = sys.stdout - - # tui_gateway.server does NOT import fastapi/starlette/uvicorn — it's - # pure stdlib + agent modules. Importing it here is what boots the - # dispatch surface (session store, thread pool, skin resolver, etc). - from tui_gateway import server # noqa: F401 — import side effects initialize the dispatch surface - - # Register all RPC method modules onto the server. - from tui_gateway.methods_session import register as _reg_session - _reg_session(server) - # methods_complete, methods_prompt, etc. are auto-registered via - # tui_gateway.server's own import-time @method() decorator. - - async def _main() -> None: - import websockets - - ws_server = await websockets.serve( - _handle_connection, - host, - port, - # Match the dashboard's keepalive behaviour. - ping_interval=20, - ping_timeout=20, - ) - - # Discover the actual port (when port=0, the OS assigns one). - sockets = ws_server.sockets - actual_port = sockets[0].getsockname()[1] if sockets else port - - # Ready sentinel — the desktop's main.cjs parses this from stdout. - # Use the captured _real_stdout because tui_gateway.server redirects - # sys.stdout → sys.stderr at module level. - _real_stdout.write( - f"HERMES_BACKEND_READY port={actual_port} token={_SESSION_TOKEN}\n" - ) - _real_stdout.flush() - _log.info("ws-only gateway listening on %s:%d", host, actual_port) - - # Block forever. - await asyncio.Event().wait() - - try: - asyncio.run(_main()) - except KeyboardInterrupt: - _log.info("ws-only gateway shutting down") - - -if __name__ == "__main__": - logging.basicConfig(level=logging.INFO) - run() diff --git a/tui_gateway/event_replay.py b/tui_gateway/event_replay.py index 3b9e2c0d90..0ec4e2a86b 100644 --- a/tui_gateway/event_replay.py +++ b/tui_gateway/event_replay.py @@ -9,12 +9,11 @@ replays everything newer from the buffer, then live events resume seamlessly. Design constraints honored: - stdio TUI path unaffected: frames gain a ``seq`` field only on event frames; Ink ignores unknown params keys. -- Thread safety: a single module lock guards counters + buffers, so buffer - order always matches seq order. Wire order is enforced separately by the - per-transport write path; two racing writers can briefly invert seq order - on the wire, which the client tolerates (watermarks are monotonic-max). +- Thread safety: a single module lock guards counters + buffers; write_json + already serializes per-transport writes, so stamping under the lock cannot + reorder frames relative to each other. - Memory bound: _REPLAY_BUFFER_MAX events / _REPLAY_SESSIONS_MAX sessions, - least-recently-active session evicted first. + oldest session evicted FIFO. """ from __future__ import annotations @@ -23,47 +22,34 @@ import threading import uuid from collections import OrderedDict, deque -# Replay ring per session. Sized for control events only — streaming token -# deltas are transient (stamped but not buffered, see _TRANSIENT_EVENT_TYPES), -# so 512 slots cover hours of durable events instead of ~one streaming burst. +# Process identity for the replay contract. Seq counters live in-process, so +# a gateway restart silently resets them to 1 while clients still hold high +# watermarks — events_since(sid, 97) then returns [] with truncated=False and +# the client believes it missed nothing (and its stale watermark makes every +# future replay empty too). The epoch lets clients detect the restart and +# reset their watermarks. +_REPLAY_EPOCH = uuid.uuid4().hex + +# Replay ring per session. A long turn emits ~hundreds of token events; this +# covers several minutes of streaming plus all control events. _REPLAY_BUFFER_MAX = 512 # Distinct sessions remembered. Desktop users rarely exceed a dozen live chats. _REPLAY_SESSIONS_MAX = 64 -# Transient event types: delivered live but never buffered for replay. -# One streaming turn emits hundreds of per-token delta frames; buffering them -# would evict every durable control event from the ring (OpenHands makes the -# same split with StreamingDeltaEvent — published, never persisted). A -# reconnecting client recovers partial streamed text from the inflight -# snapshot in ``session.resume``, not from delta replay. These frames still -# get seqs (so ordering holds live), but replay skips them and gap detection -# ignores them via the durable-seq watermark below. -_TRANSIENT_EVENT_TYPES = frozenset({ - "message.delta", - "thinking.delta", -}) - -# Server boot epoch: lets a client detect that the seq namespace was reset -# (gateway restart) — a stale high watermark from the previous process must -# not suppress live events forever (Goose clamps; we reset via epoch). -EPOCH = uuid.uuid4().hex[:8] - _replay_lock = threading.Lock() -# sid -> deque of (seq, params_dict). params is the same dict written to the -# wire and already carries its stamped "seq" key — the replay RPC returns -# these bare event objects, matching what the client's live dispatch sees. -# Manual eviction (no maxlen): we must record the seq of every DURABLE frame -# we drop, so gap detection can distinguish "durable data lost" from "the -# missing seqs were transient deltas that were never replayable anyway". +# sid -> deque of (seq, event_object) where event_object is the frame's +# ``params`` dict (bare event: type/session_id/seq/payload) — the exact shape +# the client's dispatch path consumes. _replay_buffers: "OrderedDict[str, deque]" = OrderedDict() _replay_next_seq: dict[str, int] = {} -# sid -> highest DURABLE seq evicted from the ring (0 = nothing evicted). -# Precise truncation signal: durable data is lost for a client at last_seen -# iff an evicted durable frame had seq > last_seen. -_replay_evicted_seq: dict[str, int] = {} -def stamp_event(obj: dict) -> None: +def replay_epoch() -> str: + """Opaque token identifying this server process's seq numbering.""" + return _REPLAY_EPOCH + + +def _stamp_event(obj: dict) -> None: """Stamp one outgoing event frame (mutates obj in place) and record it.""" if obj.get("method") != "event": return @@ -75,59 +61,45 @@ def stamp_event(obj: dict) -> None: # Session-less global events (skin.changed etc.) are re-fetchable via # their own RPCs; no replay contract for them. return - transient = params.get("type") in _TRANSIENT_EVENT_TYPES with _replay_lock: seq = _replay_next_seq.get(sid, 0) + 1 _replay_next_seq[sid] = seq params["seq"] = seq - if transient: - # Live-only: seq stamped for wire ordering, never buffered. - return buf = _replay_buffers.get(sid) if buf is None: - buf = deque() + buf = deque(maxlen=_REPLAY_BUFFER_MAX) _replay_buffers[sid] = buf while len(_replay_buffers) > _REPLAY_SESSIONS_MAX: _oldest_sid, _oldest_buf = _replay_buffers.popitem(last=False) _replay_next_seq.pop(_oldest_sid, None) - _replay_evicted_seq.pop(_oldest_sid, None) - else: - # LRU, not insertion-FIFO: an actively streaming session must not - # be evicted just because it was created before idle newer ones. - _replay_buffers.move_to_end(sid) buf.append((seq, params)) - while len(buf) > _REPLAY_BUFFER_MAX: - dropped_seq, _dropped = buf.popleft() - _replay_evicted_seq[sid] = dropped_seq -def events_since(sid: str, last_seen: int) -> tuple[list[dict], int, bool]: - """Replay contract for one session, computed atomically. +def events_since(sid: str, last_seen: int) -> list[dict]: + """Return recorded EVENT OBJECTS with seq > last_seen for *sid*, in order. - Returns ``(events, latest_seq, truncated)``: - - - ``events``: bare event params dicts (``type``/``session_id``/``seq``/ - ``payload``) with ``seq > last_seen``, in seq order. Transient delta - frames are never included (they were never buffered). - - ``latest_seq``: current highest stamped seq (0 when unknown). - - ``truncated``: durable data the client has not seen is unrecoverable — - either a DURABLE frame with ``seq > last_seen`` was evicted from the - ring, or ``last_seen`` is AHEAD of ``latest_seq`` (seq namespace reset - after a gateway restart / session eviction — the client should compare - ``EPOCH`` and do a full state reload). Gaps consisting only of - transient delta seqs are NOT truncation: those frames were never - replayable and the resume snapshot covers their content. + Shape contract: each element is the frame's ``params`` dict — a bare event + object with top-level ``type`` / ``session_id`` / ``seq`` — because that is + exactly what the client's dispatch path consumes. Returning the full + JSON-RPC envelope here would make every replayed event fail the client's + ``event.type`` gate and be silently dropped. """ - sid = sid or "" with _replay_lock: - latest = _replay_next_seq.get(sid, 0) - evicted = _replay_evicted_seq.get(sid, 0) - buf = _replay_buffers.get(sid) + buf = _replay_buffers.get(sid or "") if not buf: - return [], latest, last_seen > latest or evicted > last_seen - frames = [params for seq, params in buf if seq > last_seen] - truncated = last_seen > latest or evicted > last_seen - return frames, latest, truncated + return [] + return [event for seq, event in buf if seq > last_seen] + + +def is_truncated(sid: str, last_seen: int) -> bool: + """True when events between *last_seen* and the ring's oldest retained + seq were evicted — the client must refetch history instead of trusting + the replay to be gap-free.""" + with _replay_lock: + buf = _replay_buffers.get(sid or "") + if not buf: + return False + return last_seen + 1 < buf[0][0] def latest_seq(sid: str) -> int: @@ -141,42 +113,13 @@ def reset_replay_state() -> None: with _replay_lock: _replay_buffers.clear() _replay_next_seq.clear() - _replay_evicted_seq.clear() - - -def replay_epoch() -> str: - """This process's seq-namespace epoch (see :data:`EPOCH`).""" - return EPOCH def replay_stats() -> dict: - """Telemetry: buffer occupancy + per-turn timing for the ops/debug surface.""" + """Telemetry: buffer occupancy for the ops/debug surface.""" with _replay_lock: - stats = { + return { "sessions": len(_replay_buffers), "events": sum(len(b) for b in _replay_buffers.values()), "max_per_session": _REPLAY_BUFFER_MAX, - "max_sessions": _REPLAY_SESSIONS_MAX, } - # Per-turn timing from the live session table (not under the replay lock — - # the session table has its own lock). - try: - import time as _time - - from tui_gateway.server import _sessions, _sessions_lock - - with _sessions_lock: - active_turns = [] - for sid, session in _sessions.items(): - inflight = session.get("inflight_turn") - if isinstance(inflight, dict) and inflight.get("started_at"): - active_turns.append({ - "session_id": sid, - "trace_id": inflight.get("trace_id"), - "elapsed_s": round(_time.time() - float(inflight["started_at"]), 2), - "streaming": inflight.get("streaming", False), - }) - stats["active_turns"] = active_turns - except Exception: - stats["active_turns"] = [] - return stats diff --git a/tui_gateway/methods_session.py b/tui_gateway/methods_session.py index f9162a5d9f..b82c4df717 100644 --- a/tui_gateway/methods_session.py +++ b/tui_gateway/methods_session.py @@ -3654,18 +3654,17 @@ def _(rid, params: dict) -> dict: return _err(rid, -32602, "invalid params: last_seen must be an integer") from tui_gateway import event_replay - frames, latest, truncated = event_replay.events_since(sid, last_seen) + frames = event_replay.events_since(sid, last_seen) return _ok(rid, { "events": frames, - "latest_seq": latest, - "truncated": truncated, + "latest_seq": event_replay.latest_seq(sid), + "truncated": event_replay.is_truncated(sid, last_seen), "count": len(frames), - # Server boot epoch: seq counters are in-process, so after a gateway + # Restart detection: seq counters are in-process, so after a gateway # restart a client's old high watermark would silently match nothing. # Clients compare this against the epoch they learned at gateway.ready - # and reset watermarks on mismatch, doing a full state reload instead - # of trusting replay. - "epoch": event_replay.EPOCH, + # and reset watermarks on mismatch. + "epoch": event_replay.replay_epoch(), }) diff --git a/tui_gateway/server.py b/tui_gateway/server.py index aef2b6bf99..04cbbe0cc2 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -39,7 +39,6 @@ from agent.compaction_display import project_compaction_message_for_display from agent.skill_commands import describe_skill_invocation from agent.conversation_loop import INTERRUPT_WAITING_FOR_MODEL_PREFIX from tui_gateway import git_probe -from tui_gateway.event_replay import stamp_event as _stamp_event from tui_gateway.turn_marker import ( clear_turn_marker, read_turn_marker, @@ -2043,9 +2042,13 @@ def write_json(obj: dict) -> bool: params = obj.get("params") sid = ((params or {}).get("session_id")) if isinstance(params, dict) else "" if sid and (t := (_sessions.get(sid) or {}).get("transport")) is not None: + from tui_gateway.event_replay import _stamp_event + _stamp_event(obj) return t.write(obj) + from tui_gateway.event_replay import _stamp_event + _stamp_event(obj) return (current_transport() or _stdio_transport).write(obj) @@ -2054,17 +2057,6 @@ def _event_frame(event: str, sid: str, payload: dict | None = None) -> dict: params: dict = {"type": event, "session_id": sid} if payload is not None: params["payload"] = payload - # Stamp the current turn's trace_id onto every event frame so the client - # can correlate the full turn lifecycle from one identifier. Looks up - # the inflight turn for this session; events outside a turn (session.info - # on idle, skin.changed) have no trace_id. - session = _sessions.get(sid) - if session is not None: - inflight = session.get("inflight_turn") - if isinstance(inflight, dict): - trace_id = inflight.get("trace_id") - if trace_id: - params["trace_id"] = trace_id return {"jsonrpc": "2.0", "method": "event", "params": params} @@ -8700,11 +8692,6 @@ def _start_inflight_turn(session: dict, text: Any) -> None: "streaming": True, "updated_at": now, "user": _inflight_text(text), - # Per-turn trace ID: stamped on every event frame emitted during this - # turn so a client can correlate the full lifecycle (dispatch → first - # token → tool calls → complete) from a single identifier. Cleared - # when the turn ends. - "trace_id": uuid.uuid4().hex[:12], } diff --git a/tui_gateway/ws.py b/tui_gateway/ws.py index dce70f57ba..ce0cbc7021 100644 --- a/tui_gateway/ws.py +++ b/tui_gateway/ws.py @@ -33,6 +33,7 @@ import time from typing import Any from tui_gateway import server +from tui_gateway.event_replay import replay_epoch _log = logging.getLogger(__name__) @@ -365,7 +366,6 @@ async def handle_ws( # (#60800). The skin payload is small (a dict of strings/arrays), # so the to_thread overhead is negligible. skin_payload = await asyncio.to_thread(server.resolve_skin) - from tui_gateway.event_replay import EPOCH as _replay_epoch ready_ok = await transport.write_async( { "jsonrpc": "2.0", @@ -379,12 +379,10 @@ async def handle_ws( "skin": skin_payload, "change_events": True, "heartbeat": True, - # Seq-namespace epoch: changes on every gateway boot. - # A client whose stored epoch differs must reset its - # per-session seq watermarks and reload state instead - # of trusting replay (stale high watermarks from the - # previous process would suppress live events). - "epoch": _replay_epoch, + # Replay-contract process identity: lets reconnecting + # clients detect a backend restart and reset their + # per-session seq watermarks (see event_replay). + "replay_epoch": replay_epoch(), }, }, }