From 1ed7a0c0fb9da01b8324f6261a23ee95353c3843 Mon Sep 17 00:00:00 2001 From: ethernet Date: Tue, 21 Jul 2026 16:02:28 -0400 Subject: [PATCH] fix(desktop): preserve slash command and host compression semantics Keep commands whose CLI behavior exceeds their current RPC contracts on slash.exec. Propagate the full compression timeout through compute-host control, return structured host compression outcomes with metadata, and retain successful compression feedback in the desktop transcript. Add regressions for timeout forwarding, host aborts and metadata sync, structured host control responses, command routing parity, and numeric stop counts. --- .../hooks/use-prompt-actions/index.test.tsx | 4 +- .../session/hooks/use-prompt-actions/slash.ts | 13 ++- .../hooks/use-prompt-actions/utils.test.ts | 8 +- .../session/hooks/use-prompt-actions/utils.ts | 6 +- .../src/lib/desktop-slash-commands.test.ts | 38 +++----- .../desktop/src/lib/desktop-slash-commands.ts | 26 ++---- tests/test_tui_gateway_server.py | 88 ++++++++++++++++++- tests/tui_gateway/test_compute_host_phase1.py | 56 ++++++++++++ tui_gateway/compute_host.py | 38 ++++++++ tui_gateway/server.py | 12 +++ 10 files changed, 235 insertions(+), 54 deletions(-) diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx index b241e355c19..616b03b3256 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/index.test.tsx @@ -427,7 +427,9 @@ describe('usePromptActions /compress', () => { await handle!.submitText('/compress') - expect(renderedSeedTexts(seeds)).toEqual(expect.arrayContaining(['compute-host summary', 'compute-host answer'])) + const computeHostTexts = renderedSeedTexts(seeds) + expect(computeHostTexts).toEqual(expect.arrayContaining(['compute-host summary', 'compute-host answer'])) + expect(computeHostTexts.some(text => text.includes('Compressed 4 → 2 messages'))).toBe(true) expect($notifications.get()).toEqual( expect.arrayContaining([expect.objectContaining({ message: 'Compressed 4 → 2 messages' })]) ) diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts index 4ca7a728843..86c46cfd22e 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/slash.ts @@ -420,6 +420,14 @@ export function useSlashCommand(deps: SlashCommandDeps) { ) const aborted = result.status === 'aborted' || result.summary.aborted === true + + if (!aborted) { + // Keep a durable record of a successful manual compression in + // the chat after replacing it with the authoritative backend + // transcript. Errors remain transient: appending an error as a + // system message would look like a successful state change. + renderSlashOutput(lines.join('\n')) + } notify({ durationMs: 5_000, id: noticeId, kind: aborted ? 'error' : 'success', message: lines.join('\n') }) return @@ -428,17 +436,20 @@ export function useSlashCommand(deps: SlashCommandDeps) { const hostOutput = result?.host_ack?.output?.trim() if (hostOutput) { + renderSlashOutput(hostOutput) notify({ durationMs: 5_000, id: noticeId, kind: 'success', message: hostOutput }) return } const removed = result?.removed ?? 0 + const message = removed > 0 ? `compressed ${removed} messages` : 'nothing to compress' + renderSlashOutput(message) notify({ durationMs: 5_000, id: noticeId, kind: 'success', - message: removed > 0 ? `compressed ${removed} messages` : 'nothing to compress' + message }) } catch (err) { dismissNotification(noticeId) diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.test.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.test.ts index 4bf70e2fb54..061fd16210d 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.test.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.test.ts @@ -176,12 +176,12 @@ describe('renderRpcResult', () => { }) describe('process.stop', () => { - it('reports killed processes positively', () => { - expect(renderRpcResult({ killed: true }, 'stop')).toBe('Stopped all background processes.') + it('reports the numeric number of stopped processes', () => { + expect(renderRpcResult({ killed: 2 }, 'stop')).toBe('Stopped 2 background processes.') }) - it('reports nothing-to-stop when killed is false', () => { - expect(renderRpcResult({ killed: false }, 'stop')).toBe('No background processes to stop.') + it('reports nothing-to-stop when the numeric count is zero', () => { + expect(renderRpcResult({ killed: 0 }, 'stop')).toBe('No background processes to stop.') }) }) diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts index 29b151d0187..3da0f9c983b 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/utils.ts @@ -246,9 +246,9 @@ export function renderRpcResult(response: unknown, name: string): string { return 'Steer rejected — agent declined input' } - // process.stop — { killed: boolean } - if ('killed' in r && typeof r.killed === 'boolean') { - return r.killed ? 'Stopped all background processes.' : 'No background processes to stop.' + // process.stop — { killed: number } + if ('killed' in r && typeof r.killed === 'number') { + return r.killed > 0 ? `Stopped ${r.killed} background process${r.killed === 1 ? '' : 'es'}.` : 'No background processes to stop.' } // session.save — { file } diff --git a/apps/desktop/src/lib/desktop-slash-commands.test.ts b/apps/desktop/src/lib/desktop-slash-commands.test.ts index 54ce9374287..cbca758e7e6 100644 --- a/apps/desktop/src/lib/desktop-slash-commands.test.ts +++ b/apps/desktop/src/lib/desktop-slash-commands.test.ts @@ -95,19 +95,13 @@ describe('desktop slash command curation', () => { expect(isDesktopSlashSuggestion('/compact')).toBe(false) }) - it('routes commands with dedicated gateway RPCs to the rpc surface', () => { - const rpcNames = ['/agents', '/save', '/status', '/steer', '/stop', '/usage'] as const - + it('routes only stateless session commands through dedicated gateway RPCs', () => { const expected = { - '/agents': 'agents.list', '/save': 'session.save', - '/status': 'session.status', - '/steer': 'session.steer', - '/stop': 'process.stop', - '/usage': 'session.usage' + '/status': 'session.status' } as const - for (const name of rpcNames) { + for (const [name, rpcName] of Object.entries(expected)) { const surface = resolveDesktopCommand(name)?.surface expect(surface?.kind).toBe('rpc') @@ -115,22 +109,16 @@ describe('desktop slash command curation', () => { continue } - expect(surface.rpc).toBe(expected[name]) + expect(surface.rpc).toBe(rpcName) + expect(surface.buildParams({ arg: 'topic A', command: name, name: name.slice(1), sessionId: 's-1' })).toEqual({ + session_id: 's-1' + }) + } + }) - const params = surface.buildParams({ arg: 'topic A', command: name, name: name.slice(1), sessionId: 's-1' }) - - // process.stop doesn't take a session_id — kills ALL background - // processes. Other commands must echo the active session id. - if (name === '/stop') { - expect(params).not.toHaveProperty('session_id') - } else { - expect(params.session_id).toBe('s-1') - } - - // steer threads the typed arg through as `text`; others ignore it. - if (name === '/steer') { - expect(params.text).toBe('topic A') - } + it('keeps commands with richer CLI semantics on the slash worker', () => { + for (const name of ['/agents', '/steer', '/stop', '/usage']) { + expect(resolveDesktopCommand(name)?.surface).toEqual({ kind: 'exec' }) } }) @@ -285,7 +273,7 @@ describe('desktop slash command curation', () => { expect(resolveDesktopCommand('/new')?.surface).toEqual({ kind: 'action', action: 'new' }) expect(resolveDesktopCommand('/reset')?.surface).toEqual({ kind: 'action', action: 'new' }) expect(resolveDesktopCommand('/resume')?.surface).toEqual({ kind: 'picker', picker: 'session' }) - expect(resolveDesktopCommand('/usage')?.surface.kind).toBe('rpc') + expect(resolveDesktopCommand('/usage')?.surface).toEqual({ kind: 'exec' }) expect(resolveDesktopCommand('/clear')?.surface).toEqual({ kind: 'unavailable', reason: 'terminal' }) // Skill / quick commands aren't in the registry. expect(resolveDesktopCommand('/gif-search')).toBeNull() diff --git a/apps/desktop/src/lib/desktop-slash-commands.ts b/apps/desktop/src/lib/desktop-slash-commands.ts index ca5af91fb16..3b386e401b3 100644 --- a/apps/desktop/src/lib/desktop-slash-commands.ts +++ b/apps/desktop/src/lib/desktop-slash-commands.ts @@ -185,12 +185,12 @@ const DESKTOP_COMMAND_SPECS: readonly DesktopCommandSpec[] = [ // route to it directly via `rpc(...)` — bypassing slash.exec avoids the // slash-worker pipe timeout and the "not a quick/plugin/skill command" // fallback noise for commands the dispatcher doesn't handle inline. - { - name: '/agents', - description: 'Show active desktop sessions and running tasks', - aliases: ['/tasks'], - surface: rpc('agents.list', ctx => ({ session_id: ctx.sessionId })) - }, + // These commands have gateway RPCs, but their established desktop behavior + // carries richer CLI semantics: /agents includes delegations, /stop cancels + // them, /steer falls back to a next-turn prompt, and /usage is a formatted + // live report. Keep them on slash.exec until their RPC contracts are fully + // equivalent. + { name: '/agents', description: 'Show active desktop sessions and running tasks', aliases: ['/tasks'], surface: exec() }, { name: '/background', description: 'Run a prompt in the background', aliases: ['/bg', '/btw'], surface: exec() }, // /compress must be an action (session.compress RPC), not exec: the slash // worker route times out on large sessions (30s WS / 45s pipe) before the @@ -231,19 +231,11 @@ const DESKTOP_COMMAND_SPECS: readonly DesktopCommandSpec[] = [ description: 'Show current session status', surface: rpc('session.status', ctx => ({ session_id: ctx.sessionId })) }, - { - name: '/steer', - description: 'Steer the current run after the next tool call', - surface: rpc('session.steer', ctx => ({ session_id: ctx.sessionId, text: ctx.arg.trim() })) - }, - { name: '/stop', description: 'Stop running background processes', surface: rpc('process.stop', () => ({})) }, + { name: '/steer', description: 'Steer the current run after the next tool call', surface: exec(), args: true }, + { name: '/stop', description: 'Stop running background processes', surface: exec() }, { name: '/tools', description: 'List or toggle tools available to the agent', surface: exec(), args: true }, { name: '/undo', description: 'Remove the last user/assistant exchange', surface: exec() }, - { - name: '/usage', - description: 'Show token usage for this session', - surface: rpc('session.usage', ctx => ({ session_id: ctx.sessionId })) - }, + { name: '/usage', description: 'Show token usage for this session', surface: exec() }, { name: '/version', description: 'Show Hermes Agent version', surface: exec() }, // No desktop surface, but carry an alias (underscore spelling variants). diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index c5011d43d11..0ee06100870 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -2527,7 +2527,7 @@ def test_make_agent_passes_configured_fallback_chain(monkeypatch): assert agent.model == "gpt-5.5" assert captured["fallback_model"] == fallback_chain - assert captured["platform"] == "tui" + assert captured["platform"] == "desktop" def test_background_agent_kwargs_preserves_full_fallback_chain(monkeypatch): @@ -3811,7 +3811,7 @@ def test_ensure_session_db_row_persists_explicit_cwd(monkeypatch, tmp_path): server._ensure_session_db_row({"session_key": "k1", "cwd": str(tmp_path), "explicit_cwd": True}) assert created == [ - {"key": "k1", "source": "tui", "model": "test-model", "model_config": None, "cwd": str(tmp_path)} + {"key": "k1", "source": "desktop", "model": "test-model", "model_config": None, "cwd": str(tmp_path)} ] @@ -3851,7 +3851,7 @@ def test_ensure_session_db_row_defaults_to_no_workspace(monkeypatch, tmp_path): server._ensure_session_db_row({"session_key": "k1", "cwd": str(tmp_path)}) assert created == [ - {"key": "k1", "source": "tui", "model": "test-model", "model_config": None, "cwd": None} + {"key": "k1", "source": "desktop", "model": "test-model", "model_config": None, "cwd": None} ] @@ -5908,6 +5908,88 @@ def test_session_compress_returns_compute_host_history(monkeypatch): } +def test_session_compress_forwards_120_second_budget_to_compute_host(monkeypatch): + session = _session(agent=None, _compute_host_active=True) + server._sessions["sid"] = session + calls = [] + + def send_control(*args, **kwargs): + calls.append((args, kwargs)) + return { + "type": "control.ack", + "result": { + "status": "compressed", + "messages": [], + "removed": 0, + "summary": {"headline": "Already compressed", "noop": True}, + }, + } + + monkeypatch.setattr(server, "_session_uses_compute_host", lambda _session: True) + monkeypatch.setattr(server, "_send_compute_host_control", send_control) + + try: + resp = server.handle_request( + {"id": "1", "method": "session.compress", "params": {"session_id": "sid"}} + ) + finally: + server._sessions.pop("sid", None) + + assert resp["result"]["status"] == "compressed" + assert calls == [ + ( + ("sid",), + { + "route_name": "session.compress", + "command": "/compress", + "wait": True, + "timeout": 120.0, + }, + ) + ] + + +def test_session_compress_preserves_compute_host_aborted_summary(monkeypatch): + session = _session(agent=None, _compute_host_active=True) + server._sessions["sid"] = session + result = { + "status": "aborted", + "messages": [{"role": "user", "content": "preserved context"}], + "removed": 0, + "summary": { + "aborted": True, + "headline": "Compression aborted: 6 messages preserved", + "note": "No compression provider is configured.", + }, + } + monkeypatch.setattr(server, "_session_uses_compute_host", lambda _session: True) + monkeypatch.setattr( + server, + "_send_compute_host_control", + lambda *args, **kwargs: { + "type": "control.ack", + "result": result, + "session_key": "rotated-host-key", + "history_version": 7, + "message_count": 1, + "session_info": {"model": "host-model"}, + }, + ) + + try: + resp = server.handle_request( + {"id": "1", "method": "session.compress", "params": {"session_id": "sid"}} + ) + finally: + server._sessions.pop("sid", None) + + assert resp["result"] == {**result, "turn_isolation": True} + assert session["session_key"] == "rotated-host-key" + assert session["history_version"] == 7 + assert session["_metadata_message_count"] == 1 + assert session["_metadata_mirror"]["model"] == "host-model" + + def test_session_compress_reports_aborted_summary_without_success(monkeypatch): compression_state = types.SimpleNamespace( _last_compress_aborted=True, diff --git a/tests/tui_gateway/test_compute_host_phase1.py b/tests/tui_gateway/test_compute_host_phase1.py index 46c855401d0..fb1a0172898 100644 --- a/tests/tui_gateway/test_compute_host_phase1.py +++ b/tests/tui_gateway/test_compute_host_phase1.py @@ -252,6 +252,62 @@ def test_compute_host_compress_control_runs_identity_guard_in_host(monkeypatch): assert ack["session_info"]["model"] == "host-model" +def test_compute_host_session_compress_returns_structured_result(monkeypatch): + from tui_gateway import server + + out = io.StringIO() + host = ComputeHost(stdout=out, max_workers=1, heartbeat_secs=0) + session = { + "agent": None, + "session_key": "host-key", + "history": [{"role": "user", "content": "preserved"}], + "history_lock": threading.Lock(), + "history_version": 3, + "running": False, + } + calls: list[dict] = [] + + def compress_handler(_rid, params): + calls.append(params) + return { + "result": { + "status": "aborted", + "messages": [{"role": "user", "content": "preserved"}], + "summary": {"aborted": True, "headline": "Compression aborted"}, + } + } + + server._sessions["sid"] = session + monkeypatch.setitem(server._methods, "session.compress", compress_handler) + monkeypatch.setattr(server, "_session_info", lambda _agent, _session: {"model": "host-model"}) + + try: + host.handle_frame( + { + "type": "control", + "sid": "sid", + "request_id": "compress-structured", + "route_name": "session.compress", + "command": "/compress auth", + } + ) + ack = _wait_for_frame( + out, + lambda frame: frame.get("type") == "control.ack" and frame.get("request_id") == "compress-structured", + ) + finally: + server._sessions.pop("sid", None) + host.close() + + assert calls == [{"session_id": "sid", "focus_topic": "auth"}] + assert ack["result"]["status"] == "aborted" + assert ack["result"]["summary"]["aborted"] is True + assert ack["session_key"] == "host-key" + assert ack["history_version"] == 3 + assert ack["message_count"] == 1 + assert ack["session_info"] == {"model": "host-model"} + + def test_append_log_record_single_write_lines(tmp_path): path = tmp_path / "agent.log" diff --git a/tui_gateway/compute_host.py b/tui_gateway/compute_host.py index d95c769c87f..9ff3e100cb3 100644 --- a/tui_gateway/compute_host.py +++ b/tui_gateway/compute_host.py @@ -570,6 +570,44 @@ class ComputeHost: } ) return + if route_name == "session.compress": + command = str(frame.get("command") or "") + focus_topic = command.removeprefix("/compress").strip() + response = server._methods["session.compress"]( + request_id, + { + "session_id": sid, + **({"focus_topic": focus_topic} if focus_topic else {}), + }, + ) + if "error" in response: + self.emit( + { + "type": "control.error", + "sid": sid, + "request_id": request_id, + "message": str(response["error"].get("message") or "session compression failed"), + } + ) + return + with session["history_lock"]: + session_key = str(session.get("session_key") or "") + history_version = int(session.get("history_version", 0)) + message_count = len(session.get("history") or []) + self.emit( + { + "type": "control.ack", + "sid": sid, + "request_id": request_id, + "route_name": route_name, + "result": response.get("result") or {}, + "session_key": session_key, + "history_version": history_version, + "message_count": message_count, + "session_info": server._session_info(session.get("agent"), session), + } + ) + return command = str(frame.get("command") or "") output = "" if command: diff --git a/tui_gateway/server.py b/tui_gateway/server.py index ea61c3d4977..2f6c7488108 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -1381,6 +1381,7 @@ def _send_compute_host_control( command: str = "", payload: dict | None = None, wait: bool = True, + timeout: float = 30.0, ) -> dict: frame = dict(payload or {}) frame.setdefault("type", "control") @@ -1390,6 +1391,7 @@ def _send_compute_host_control( route_name=route_name, payload=frame, wait=wait, + timeout=timeout, ) @@ -8991,6 +8993,7 @@ def _(rid, params: dict) -> dict: session, err = _sess_nowait(params, rid) if err: return err + assert session is not None if _session_uses_compute_host(session): sid = str(params.get("session_id") or "") focus_topic = str(params.get("focus_topic", "") or "").strip() @@ -9001,12 +9004,21 @@ def _(rid, params: dict) -> dict: route_name="session.compress", command=command, wait=True, + timeout=120.0, ) except Exception as exc: return _err(rid, 5019, f"compute-host compress failed: {exc}") if ack.get("type") in {"control.error", "error"}: return _err(rid, 4009, str(ack.get("message") or "compute-host compress failed")) _apply_compute_host_metadata_mirror(session, ack) + host_result = ack.get("result") + if isinstance(host_result, dict): + # The host owns the isolated session's agent/history, so preserve + # its structured compression result verbatim. In particular this + # carries `status: aborted` and `summary.aborted`; flattening the + # old text-only acknowledgement made Desktop show aborted work as a + # success toast. + return _ok(rid, {**host_result, "turn_isolation": True}) host_info = ack.get("session_info") if isinstance(ack.get("session_info"), dict) else {} host_messages = ack.get("messages") if isinstance(ack.get("messages"), list) else [] # `messages` is returned at top level for the desktop transcript