diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/compaction-event.test.tsx b/apps/desktop/src/app/session/hooks/use-message-stream/compaction-event.test.tsx index 7152d9eba5..6a72ea9705 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/compaction-event.test.tsx +++ b/apps/desktop/src/app/session/hooks/use-message-stream/compaction-event.test.tsx @@ -1,6 +1,7 @@ import { act, cleanup } from '@testing-library/react' import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' +import { createClientSessionState } from '@/lib/chat-runtime' import { $compactingSessions, setSessionCompacting } from '@/store/compaction' import type { RpcEvent } from '@/types/hermes' @@ -56,6 +57,31 @@ describe('useMessageStream compaction lifecycle', () => { expect($compactingSessions.get()).toEqual({ [OTHER_SID]: true }) }) + // #97948: a manual /compress whose RPC answered `pending` (the compute host + // outlived the gateway's wait) has no turn-end hydrate — the `compacted` + // edge is the only signal the transcript changed. + it('rehydrates the idle active session on the compacted edge', () => { + const hydrateFromStoredSession = vi.fn(async () => undefined) + const states = new Map([[SID, { ...createClientSessionState(), storedSessionId: 'stored-1' }]]) + + stream = renderMessageStream(SID, { hydrateFromStoredSession, states }) + + emit('status.update', { kind: 'compacted' }) + + expect(hydrateFromStoredSession).toHaveBeenCalledWith(3, 'stored-1', SID) + }) + + it('leaves the transcript to the turn settle path when compaction ends mid-turn', () => { + const hydrateFromStoredSession = vi.fn(async () => undefined) + const states = new Map([[SID, { ...createClientSessionState(), busy: true, storedSessionId: 'stored-1' }]]) + + stream = renderMessageStream(SID, { hydrateFromStoredSession, states }) + + emit('status.update', { kind: 'compacted' }) + + expect(hydrateFromStoredSession).not.toHaveBeenCalled() + }) + it('reconciles a reconnecting compaction only from trusted terminal server state', () => { mountStream() emit('status.update', { kind: 'compacting' }) diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/status.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/status.ts index 76b1b40a20..3fb0ff7e75 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/status.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event/status.ts @@ -21,7 +21,16 @@ import type { GatewayEventContext } from './types' * error — the status-and-notice tail of the dispatcher. */ export function handleStatusEvent(ctx: GatewayEventContext): boolean { const { deps, event, payload, sessionId, isActiveEvent, occurredAt } = ctx - const { compactedTurnRef, failAssistantMessage, flushQueuedDeltas, queryClient, updateSessionState } = deps + + const { + compactedTurnRef, + failAssistantMessage, + flushQueuedDeltas, + hydrateFromStoredSession, + queryClient, + sessionStateByRuntimeIdRef, + updateSessionState + } = deps if (event.type === 'status.update') { if (sessionId && payload?.kind === 'compacting') { @@ -30,6 +39,17 @@ export function handleStatusEvent(ctx: GatewayEventContext): boolean { } else if (sessionId && payload?.kind === 'compacted') { reconcileSessionCompacting(sessionId, 'terminal') compactedTurnRef.current.delete(sessionId) + + // A compress that finished with no live turn (manual /compress whose + // RPC answered `pending` because the compute host outlived the wait, + // #97948) has no turn-end hydrate to refresh the transcript — the + // summarized bubbles would stay on screen forever. Mid-turn compaction + // still defers to the turn's own settle path. + const state = sessionStateByRuntimeIdRef.current.get(sessionId) + + if (isActiveEvent && state && !state.busy && !state.awaitingResponse && !state.streamId) { + void hydrateFromStoredSession(3, state.storedSessionId, sessionId) + } } else if (sessionId && payload?.kind === 'process') { // The gateway's notification poller announces background process // completions / watch matches here — re-sync the status stack. 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 59e7182195..1b206f9eca 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 @@ -31,6 +31,7 @@ import { $wakeWord, resetWakeWordState } from '@/store/wake-word' import type { SessionInfo } from '@/types/hermes' import { clearSingleFlightSessionResumeState } from './single-flight-resume' +import { SESSION_COMPRESS_TIMEOUT_MS } from './slash' import type { SubmitTextOptions } from './utils' import { uploadComposerAttachment, usePromptActions } from '.' @@ -692,7 +693,7 @@ describe('usePromptActions /compress', () => { vi.restoreAllMocks() }) - it('routes through session.compress (not slash.exec) with a 120s timeout and renders the summary', async () => { + it('routes through session.compress (not slash.exec) with the compute-host ceiling timeout and renders the summary', async () => { const seeds: Record[] = [] const requestGateway = vi.fn(async (method: string, _params?: Record, _timeoutMs?: number) => { @@ -728,7 +729,7 @@ describe('usePromptActions /compress', () => { expect(requestGateway).toHaveBeenCalledWith( 'session.compress', expect.objectContaining({ session_id: RUNTIME_SESSION_ID }), - 120_000 + SESSION_COMPRESS_TIMEOUT_MS ) expect(requestGateway).not.toHaveBeenCalledWith('slash.exec', expect.anything()) expect(requestGateway).not.toHaveBeenCalledWith('command.dispatch', expect.anything()) @@ -862,7 +863,7 @@ describe('usePromptActions /compress', () => { expect(requestGateway).toHaveBeenCalledWith( 'session.compress', expect.objectContaining({ focus_topic: 'the auth refactor' }), - 120_000 + SESSION_COMPRESS_TIMEOUT_MS ) }) @@ -969,7 +970,7 @@ describe('usePromptActions /compress', () => { act(() => { submitted = handle!.submitTextRaw('/compress') }) - await waitFor(() => expect(requestGateway).toHaveBeenCalledWith('session.compress', expect.anything(), 120_000)) + await waitFor(() => expect(requestGateway).toHaveBeenCalledWith('session.compress', expect.anything(), SESSION_COMPRESS_TIMEOUT_MS)) // Switch to session B before compression resolves. activeSessionIdRef.current = RUNTIME_SESSION_B @@ -1028,7 +1029,7 @@ describe('usePromptActions /compress', () => { act(() => { submitted = handle!.submitTextRaw('/compress') }) - await waitFor(() => expect(requestGateway).toHaveBeenCalledWith('session.compress', expect.anything(), 120_000)) + await waitFor(() => expect(requestGateway).toHaveBeenCalledWith('session.compress', expect.anything(), SESSION_COMPRESS_TIMEOUT_MS)) activeSessionIdRef.current = RUNTIME_SESSION_B storedSessionIdRef.current = 'stored-b' rejectCompress(new Error('compression failed')) 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 770f02ae9b..a06b6b1643 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 @@ -74,9 +74,13 @@ import { } from './utils' // Manual compression is LLM-bound and routinely outlives the desktop's 30s -// default WS request timeout on large sessions — give it the TUI client's -// 120s RPC budget (HERMES_TUI_RPC_TIMEOUT_MS default) instead. -const SESSION_COMPRESS_TIMEOUT_MS = 120_000 +// default WS request timeout on large sessions. The gateway blocks its own +// compute-host wait for up to compression.context_total_ceiling_seconds + 30s +// (capped at 630s, tui_gateway/server.py _COMPUTE_HOST_COMPRESS_WAIT_CAP_SECS) +// and then answers `status: 'pending'` rather than an error, so this budget +// must sit above that cap or the desktop reports a false timeout while the +// host is still compressing (#97948). +export const SESSION_COMPRESS_TIMEOUT_MS = 660_000 const WAKE_START_TIMEOUT_MS = 180_000 const wakeDeviceLabel = (device?: WakeInputDeviceStatus): string => { @@ -666,6 +670,16 @@ export function useSlashCommand(deps: SlashCommandDeps) { sessionId = liveSessionId + // The gateway's compute-host wait expired but compression is still + // running there; it pushes session.info + a `compacted` status edge + // when the host finishes. Not an error (#97948). + if (result?.status === 'pending') { + const pendingMessage = result.message || 'compression still running in the background' + notify({ durationMs: 8_000, id: noticeId, kind: 'info', message: pendingMessage }) + + return + } + // Replace the transcript with the post-compress history so the // summarized bubbles actually disappear. `messages` is the same // shape session.resume returns (_history_to_messages), so diff --git a/apps/desktop/src/app/types.ts b/apps/desktop/src/app/types.ts index 0931ecf879..57e85a0a9c 100644 --- a/apps/desktop/src/app/types.ts +++ b/apps/desktop/src/app/types.ts @@ -66,6 +66,10 @@ export interface SessionCompressResponse { usage?: Partial } messages?: SessionMessage[] + /** Set with `status: 'pending'` when the gateway's compute-host wait expired + * while compression is still running; the transcript refreshes from the + * pushed session.info / `compacted` status edge (#97948). */ + message?: string removed?: number status?: string summary?: { diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index 647ea82618..fcb059da82 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -633,7 +633,7 @@ def test_slash_exec_compress_flag_on_applies_host_control_mirror(monkeypatch): def __init__(self): self.controls = [] - def control(self, sid, *, route_name, payload=None, wait=True, timeout=30.0): + def control(self, sid, *, route_name, payload=None, wait=True, timeout=30.0, on_late_ack=None): self.controls.append((sid, route_name, dict(payload or {}), wait)) return { "type": "control.ack", @@ -10500,7 +10500,7 @@ def test_session_compress_returns_compute_host_history(monkeypatch): } -def test_session_compress_forwards_120_second_budget_to_compute_host(monkeypatch): +def test_session_compress_forwards_config_ceiling_budget_to_compute_host(monkeypatch): session = _session(agent=None, _compute_host_active=True) server._sessions["sid"] = session calls = [] @@ -10519,6 +10519,9 @@ def test_session_compress_forwards_120_second_budget_to_compute_host(monkeypatch monkeypatch.setattr(server, "_session_uses_compute_host", lambda _session: True) monkeypatch.setattr(server, "_send_compute_host_control", send_control) + monkeypatch.setattr( + server, "_load_cfg", lambda: {"compression": {"context_total_ceiling_seconds": 300}} + ) try: resp = server.handle_request( @@ -10528,17 +10531,17 @@ def test_session_compress_forwards_120_second_budget_to_compute_host(monkeypatch server._sessions.pop("sid", None) assert resp["result"]["status"] == "compressed" - assert calls == [ - ( - ("sid",), - { - "route_name": "session.compress", - "command": "/compress", - "wait": True, - "timeout": 120.0, - }, - ) - ] + assert len(calls) == 1 + (sid_arg,), kwargs = calls[0] + assert sid_arg == "sid" + assert kwargs["route_name"] == "session.compress" + assert kwargs["command"] == "/compress" + assert kwargs["wait"] is True + # #97948: the waiter follows compression.context_total_ceiling_seconds + # (+30s slack) instead of a hard-coded 120s, and registers a late-ack + # handler so a compress that outlives it is still adopted. + assert kwargs["timeout"] == 330.0 + assert callable(kwargs["on_late_ack"]) def test_session_compress_preserves_compute_host_aborted_summary(monkeypatch): diff --git a/tests/tui_gateway/test_compute_host_late_compress_ack.py b/tests/tui_gateway/test_compute_host_late_compress_ack.py new file mode 100644 index 0000000000..3458ddbd10 --- /dev/null +++ b/tests/tui_gateway/test_compute_host_late_compress_ack.py @@ -0,0 +1,235 @@ +"""Regression tests for #97948 symptom A (salvaged from #99630). + +A manual /compress on a compute-host session used to block its RPC waiter for +a hard-coded 120s, return a 5019 timeout error, and then DROP the host's late +``control.ack`` — so the rotated session_key / history_version / session_info +never reached the gateway session and the desktop never refreshed. +""" + +import queue +import sys +import threading +import time +import types + +import pytest + +from tui_gateway import server +from tui_gateway.host_supervisor import HostSupervisor + + +def _supervisor() -> tuple[HostSupervisor, list]: + sup = HostSupervisor(argv=[sys.executable, "-c", ""], autostart=False) + sent: list = [] + sup._send_frame = lambda frame: sent.append(frame) + sup.start = lambda: None # never spawn a child + return sup, sent + + +def _session(**extra) -> dict: + return { + "agent": types.SimpleNamespace(), + "session_key": "old-session-key", + "history": [], + "history_lock": threading.Lock(), + "history_version": 3, + "running": False, + "attached_images": [], + "image_counter": 0, + "cols": 80, + "slash_worker": None, + "show_reasoning": False, + "tool_progress_mode": "all", + "_compute_host_active": True, + **extra, + } + + +# ── HostSupervisor: late-ack registration ─────────────────────────────────── + + +def test_control_timeout_registers_one_shot_late_ack_handler(): + sup, sent = _supervisor() + fired: list = [] + + with pytest.raises(queue.Empty): + sup.control("sid", route_name="session.compress", payload={"command": "/compress"}, + wait=True, timeout=0.05, on_late_ack=fired.append) + + request_id = sent[0]["request_id"] + assert request_id not in sup._pending_controls + assert request_id in sup._late_control_handlers + + late = {"type": "control.ack", "request_id": request_id, "result": {"status": "compressed"}} + sup._handle_host_frame(late) + assert fired == [late] + # One-shot: a duplicate ack for the same request is ignored. + sup._handle_host_frame(late) + assert fired == [late] + assert request_id not in sup._late_control_handlers + + +def test_control_timeout_without_handler_still_drops_late_ack(): + sup, sent = _supervisor() + with pytest.raises(queue.Empty): + sup.control("sid", route_name="session.compress", wait=True, timeout=0.05) + assert sup._late_control_handlers == {} + sup._handle_host_frame({"type": "control.ack", "request_id": sent[0]["request_id"]}) + + +def test_late_control_error_and_bare_error_frames_fire_handler(): + sup, sent = _supervisor() + fired: list = [] + for _ in range(2): + with pytest.raises(queue.Empty): + sup.control("sid", route_name="session.compress", wait=True, timeout=0.01, + on_late_ack=fired.append) + rid_a, rid_b = sent[0]["request_id"], sent[1]["request_id"] + sup._handle_host_frame({"type": "control.error", "request_id": rid_a, "message": "boom"}) + sup._handle_host_frame({"type": "error", "request_id": rid_b, "message": "bad frame"}) + assert [f["request_id"] for f in fired] == [rid_a, rid_b] + + +def test_late_ack_handlers_are_bounded_by_ttl_and_cap(monkeypatch): + from tui_gateway import host_supervisor as hs + + monkeypatch.setattr(hs, "_LATE_CONTROL_MAX", 3) + sup, _sent = _supervisor() + for i in range(5): + sup._register_late_control_handler(f"r{i}", lambda _f: None) + assert len(sup._late_control_handlers) == 3 + assert set(sup._late_control_handlers) == {"r2", "r3", "r4"} + + # TTL: an old registration is dropped on the next registration. + monkeypatch.setattr(hs, "_LATE_CONTROL_TTL_SECS", 0.0) + time.sleep(0.01) + sup._register_late_control_handler("fresh", lambda _f: None) + assert set(sup._late_control_handlers) == {"fresh"} + + +def test_host_crash_fails_outstanding_late_ack_handlers(): + sup, sent = _supervisor() + fired: list = [] + with pytest.raises(queue.Empty): + sup.control("sid", route_name="session.compress", wait=True, timeout=0.01, + on_late_ack=fired.append) + sup._fail_pending_turns(reason="crash", message="compute host exited with code 1") + assert len(fired) == 1 + assert fired[0]["type"] == "control.error" + assert fired[0]["request_id"] == sent[0]["request_id"] + assert sup._late_control_handlers == {} + + +# ── session.compress RPC: pending answer + late adoption ──────────────────── + + +@pytest.fixture +def compute_host_gateway(monkeypatch): + sup, sent = _supervisor() + emitted: list = [] + monkeypatch.setattr(server, "_compute_host_supervisor", sup) + monkeypatch.setattr(server, "_emit", lambda event, sid, payload=None: emitted.append((event, sid, payload))) + monkeypatch.setattr(server, "_session_uses_compute_host", lambda _s, cfg=None: True) + monkeypatch.setattr(server, "_compute_host_compress_wait_seconds", lambda cfg=None: 0.05) + monkeypatch.setattr(server, "_session_info", lambda _agent, _session=None: {"model": "mirrored"}) + session = _session() + server._sessions["sid"] = session + try: + yield sup, sent, emitted, session + finally: + server._sessions.pop("sid", None) + + +def _late_ack(request_id: str) -> dict: + return { + "type": "control.ack", + "sid": "sid", + "request_id": request_id, + "route_name": "session.compress", + "result": {"status": "compressed", "removed": 12, "summary": {"headline": "Compressed 14 → 2"}}, + "session_key": "rotated-session-key", + "history_version": 9, + "message_count": 2, + "session_info": {"model": "host-model", "usage": {"total": 111}}, + } + + +def test_session_compress_reports_pending_and_adopts_late_ack(compute_host_gateway): + sup, sent, emitted, session = compute_host_gateway + + resp = server.handle_request({"id": "1", "method": "session.compress", "params": {"session_id": "sid"}}) + + assert "error" not in resp, resp + assert resp["result"]["status"] == "pending" + assert resp["result"]["turn_isolation"] is True + assert "background" in resp["result"]["message"] + assert sent[0]["route_name"] == "session.compress" + # Nothing adopted yet, the host is still working. + assert session["session_key"] == "old-session-key" + assert emitted == [] + + sup._handle_host_frame(_late_ack(sent[0]["request_id"])) + + assert session["session_key"] == "rotated-session-key" + assert session["history_version"] == 9 + assert session["_metadata_message_count"] == 2 + assert session["_metadata_mirror"]["model"] == "host-model" + events = [(event, payload) for event, _sid, payload in emitted] + assert ("session.info", {"model": "mirrored"}) in events + assert ("status.update", {"kind": "compacted", "text": "✓ Context compression complete"}) in events + + +def test_session_compress_late_control_error_surfaces_as_error_event(compute_host_gateway): + sup, sent, emitted, session = compute_host_gateway + + resp = server.handle_request({"id": "1", "method": "session.compress", "params": {"session_id": "sid"}}) + assert resp["result"]["status"] == "pending" + + sup._handle_host_frame({"type": "control.error", "request_id": sent[0]["request_id"], "message": "provider down"}) + + assert session["session_key"] == "old-session-key" + assert ("error", "sid", {"message": "compression failed: provider down"}) in emitted + + +def test_session_compress_late_ack_ignored_after_session_closed(compute_host_gateway): + sup, sent, emitted, session = compute_host_gateway + server.handle_request({"id": "1", "method": "session.compress", "params": {"session_id": "sid"}}) + server._sessions.pop("sid") + + sup._handle_host_frame(_late_ack(sent[0]["request_id"])) + + assert session["session_key"] == "old-session-key" + assert emitted == [] + + +def test_slash_compress_route_reports_pending_and_adopts_late_ack(compute_host_gateway): + sup, sent, emitted, session = compute_host_gateway + + resp = server.handle_request( + {"id": "1", "method": "slash.exec", "params": {"session_id": "sid", "command": "/compress"}} + ) + + assert "error" not in resp, resp + assert "compression still running in the background" in resp["result"]["output"] + assert sent[0]["route_name"] == "slash.compress" + + sup._handle_host_frame({**_late_ack(sent[0]["request_id"]), "route_name": "slash.compress"}) + assert session["session_key"] == "rotated-session-key" + assert any(event == "session.info" for event, _sid, _p in emitted) + + +# ── wait budget follows compression.context_total_ceiling_seconds ─────────── + + +def test_compress_wait_budget_follows_config_ceiling(): + assert server._compute_host_compress_wait_seconds({"compression": {}}) == 630.0 + assert server._compute_host_compress_wait_seconds( + {"compression": {"context_total_ceiling_seconds": 200}} + ) == 230.0 + # Never below the historical 120s floor, never above the RPC-safe cap. + assert server._compute_host_compress_wait_seconds( + {"compression": {"context_total_ceiling_seconds": 10, "context_timeout_seconds": 0}} + ) == 120.0 + assert server._compute_host_compress_wait_seconds( + {"compression": {"context_total_ceiling_seconds": 99999}} + ) == server._COMPUTE_HOST_COMPRESS_WAIT_CAP_SECS diff --git a/tui_gateway/host_supervisor.py b/tui_gateway/host_supervisor.py index 0b826e4abe..9f8a7bd4a6 100644 --- a/tui_gateway/host_supervisor.py +++ b/tui_gateway/host_supervisor.py @@ -47,6 +47,11 @@ MUTATOR_ROUTE_TABLE: dict[str, str] = { _REGISTRY_NAME = "dashboard-compute-host.json" _RESPAWN_WINDOW_SECS = 300.0 _SHUTDOWN_TIMEOUT_SECS = 10.0 +# Late control-ack handlers (#97948): a compress that outlives its RPC waiter +# can legitimately run for the full compression ceiling plus a stall-fallback +# retry, so keep registrations around well past that — but bounded. +_LATE_CONTROL_TTL_SECS = 1800.0 +_LATE_CONTROL_MAX = 64 def append_log_record(path: str | Path, record: str) -> None: @@ -167,6 +172,11 @@ class HostSupervisor: self._restart_times: list[float] = [] self._pending_turns: dict[str, tuple[str, Callable[[dict], None] | None]] = {} self._pending_controls: dict[str, queue.Queue[dict]] = {} + # request_id -> (registered_at, handler) for control waiters that timed + # out but whose host work is still running (#97948). The host emits + # its control.ack whenever it finishes; without this the ack matched + # no queue and was silently dropped. + self._late_control_handlers: dict[str, tuple[float, Callable[[dict], None]]] = {} self._stderr_tail: list[str] = [] self._last_progress_counter = 0 @@ -307,7 +317,17 @@ class HostSupervisor: payload: dict[str, Any] | None = None, wait: bool = True, timeout: float = 30.0, + on_late_ack: Callable[[dict], None] | None = None, ) -> dict: + """Send a control frame; with ``wait`` block up to ``timeout`` for its ack. + + ``on_late_ack`` (only meaningful with ``wait``) keeps the request + adoptable after the waiter gives up: when the host's ``control.ack`` / + ``control.error`` / ``error`` for this ``request_id`` eventually + arrives, the handler fires once instead of the frame being dropped. + Registrations are bounded by ``_LATE_CONTROL_TTL_SECS`` / + ``_LATE_CONTROL_MAX``. + """ if route_name not in MUTATOR_ROUTE_TABLE: raise ValueError(f"unclassified host mutator route: {route_name}") self.start() @@ -327,10 +347,47 @@ class HostSupervisor: return {"status": "sent", "request_id": request_id} try: return q.get(timeout=timeout) + except queue.Empty: + if on_late_ack is not None: + self._register_late_control_handler(request_id, on_late_ack) + raise finally: with self._lock: self._pending_controls.pop(request_id, None) + def _register_late_control_handler(self, request_id: str, handler: Callable[[dict], None]) -> None: + now = time.monotonic() + with self._lock: + expired = [ + rid + for rid, (registered_at, _cb) in self._late_control_handlers.items() + if now - registered_at > _LATE_CONTROL_TTL_SECS + ] + for rid in expired: + self._late_control_handlers.pop(rid, None) + while len(self._late_control_handlers) >= _LATE_CONTROL_MAX: + oldest = min(self._late_control_handlers, key=lambda rid: self._late_control_handlers[rid][0]) + self._late_control_handlers.pop(oldest, None) + self._late_control_handlers[request_id] = (now, handler) + + def _deliver_control_frame(self, request_id: str, frame: dict[str, Any]) -> None: + with self._lock: + q = self._pending_controls.get(request_id) + late = None if q is not None else self._late_control_handlers.pop(request_id, None) + if q is not None: + try: + q.put_nowait(frame) + except queue.Full: + pass + return + if late is None: + return + _registered_at, handler = late + try: + handler(frame) + except Exception: + logger.exception("compute host late control ack handler failed (request_id=%s)", request_id) + def _spawn_locked(self, *, reason: str) -> None: if self._stopped_respawning: raise RuntimeError("compute host respawn disabled after crash loop") @@ -452,24 +509,10 @@ class HostSupervisor: self._complete_turn(frame) return if ftype in {"control.ack", "control.error", "respond.ack", "respond.error", "interrupt.ack", "reload_mcp.ack", "shutdown.ack"}: - request_id = str(frame.get("request_id") or "") - with self._lock: - q = self._pending_controls.get(request_id) - if q is not None: - try: - q.put_nowait(frame) - except queue.Full: - pass + self._deliver_control_frame(str(frame.get("request_id") or ""), frame) return if ftype == "error" and frame.get("request_id"): - request_id = str(frame.get("request_id") or "") - with self._lock: - q = self._pending_controls.get(request_id) - if q is not None: - try: - q.put_nowait(frame) - except queue.Full: - pass + self._deliver_control_frame(str(frame.get("request_id") or ""), frame) def _complete_turn(self, frame: dict[str, Any]) -> None: request_id = str(frame.get("request_id") or "") @@ -524,6 +567,17 @@ class HostSupervisor: cb(frame) except Exception: logger.exception("compute host error callback failed") + # A crashed host will never emit the late acks the timed-out control + # waiters are still expecting; fail them the same way so the client's + # "still running in the background" notice does not hang forever. + with self._lock: + late = self._late_control_handlers + self._late_control_handlers = {} + for request_id, (_registered_at, handler) in late.items(): + try: + handler({"type": "control.error", "request_id": request_id, "reason": reason, "message": message}) + except Exception: + logger.exception("compute host late control error handler failed") def _maybe_respawn_after_crash(self) -> None: now = time.monotonic() diff --git a/tui_gateway/methods_session.py b/tui_gateway/methods_session.py index dcceee5c7a..66ae830391 100644 --- a/tui_gateway/methods_session.py +++ b/tui_gateway/methods_session.py @@ -3006,13 +3006,37 @@ def _(rid, params: dict) -> dict: sid = str(params.get("session_id") or "") focus_topic = str(params.get("focus_topic", "") or "").strip() command = "/compress" + (f" {focus_topic}" if focus_topic else "") + _late_session = session + + def _on_late_ack(late: dict, _sid=sid) -> None: + _adopt_late_compute_host_compress_ack(_sid, _late_session, late, route_name="session.compress") + try: ack = _send_compute_host_control( sid, route_name="session.compress", command=command, wait=True, - timeout=120.0, + # Follows compression.context_total_ceiling_seconds instead of + # a fixed 120s: the host legitimately runs that long (#97948). + timeout=_compute_host_compress_wait_seconds(), + on_late_ack=_on_late_ack, + ) + except queue.Empty: + # The waiter gave up but the host is still compressing; the late + # ack handler adopts the rotated session and pushes session.info + # when it lands. Not an error — the old 5019 made Desktop/TUI + # report a timeout while compression later succeeded silently. + return _ok( + rid, + { + "status": "pending", + "turn_isolation": True, + "message": ( + "compression still running in the background; " + "the transcript will refresh when it finishes" + ), + }, ) except Exception as exc: return _err(rid, 5019, f"compute-host compress failed: {exc}") diff --git a/tui_gateway/methods_tools.py b/tui_gateway/methods_tools.py index 7d72f7e37b..fb9fd3fdcd 100644 --- a/tui_gateway/methods_tools.py +++ b/tui_gateway/methods_tools.py @@ -1057,12 +1057,31 @@ def _(rid, params: dict) -> dict: sid = params.get("session_id", "") if _session_uses_compute_host(session): command = f"/{name}" + (f" {arg}" if arg else "") + _late_session = session + + def _on_late_ack(late: dict, _sid=sid) -> None: + _adopt_late_compute_host_compress_ack(_sid, _late_session, late, route_name="slash.compress") + try: ack = _send_compute_host_control( sid, route_name="slash.compress", command=command, wait=True, + timeout=_compute_host_compress_wait_seconds(), + on_late_ack=_on_late_ack, + ) + except queue.Empty: + return _ok( + rid, + { + "type": "exec", + "status": "pending", + "output": ( + "compression still running in the background; " + "the transcript will refresh when it finishes" + ), + }, ) except Exception as exc: return _err(rid, 5019, f"compute-host slash.compress failed: {exc}") diff --git a/tui_gateway/server.py b/tui_gateway/server.py index fc982de9e1..9a519659eb 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -2656,6 +2656,11 @@ def _broadcast_global_event(event: str, payload: dict | None = None) -> None: _compute_host_supervisor = None _compute_host_supervisor_lock = threading.Lock() +# Hard cap on how long session.compress blocks its RPC waiting for the compute +# host (#97948). Must stay below the desktop's SESSION_COMPRESS_TIMEOUT_MS +# (660s) so the client receives the `pending` answer instead of its own +# timeout error; the late-ack path covers anything slower. +_COMPUTE_HOST_COMPRESS_WAIT_CAP_SECS = 630.0 def _inside_compute_host_child() -> bool: @@ -2931,6 +2936,7 @@ def _send_compute_host_control( payload: dict | None = None, wait: bool = True, timeout: float = 30.0, + on_late_ack=None, ) -> dict: frame = dict(payload or {}) frame.setdefault("type", "control") @@ -2941,9 +2947,68 @@ def _send_compute_host_control( payload=frame, wait=wait, timeout=timeout, + on_late_ack=on_late_ack, ) +def _compute_host_compress_wait_seconds(cfg: dict | None = None) -> float: + """RPC wait budget for a compute-host compress control (#97948). + + Manual compression legitimately runs up to the configured + ``compression.context_total_ceiling_seconds`` (default 600s), so a fixed + 120s waiter reported a false timeout while the host kept working. Follow + the ceiling with a little slack, but cap the blocking wait so it stays + below the desktop's own RPC timeout; anything longer is adopted through + the late-ack path instead of failing. + """ + from agent.conversation_compression import resolve_context_compression_timeouts + + try: + compression_cfg = (cfg if cfg is not None else _load_cfg()).get("compression", {}) + except Exception: + compression_cfg = {} + _idle, ceiling = resolve_context_compression_timeouts( + compression_cfg if isinstance(compression_cfg, dict) else {} + ) + return float(min(max(ceiling + 30.0, 120.0), _COMPUTE_HOST_COMPRESS_WAIT_CAP_SECS)) + + +def _announce_compute_host_compress_done(sid: str, session: dict, ack: dict) -> None: + """Mirror a compute-host compress ack and push the client-visible edges. + + Emits the same ``session.info`` the in-process /compress path does plus + the ``compacted`` status edge, so a client whose own RPC wait already + expired still learns the transcript changed. + """ + _apply_compute_host_metadata_mirror(session, ack) + try: + info = _session_info(session.get("agent"), session) + except TypeError: + info = _session_info(session.get("agent")) + _emit("session.info", sid, info) + _status_update(sid, "compacted", "✓ Context compression complete") + + +def _adopt_late_compute_host_compress_ack(sid: str, session: dict, ack: dict, *, route_name: str) -> None: + """Adopt a compute-host compress ack that arrived after its RPC waiter gave up. + + The RPC already answered ``status: pending``; this is the only place the + rotated session_key / history_version / session_info mirror can land, and + the only signal the client gets that the transcript changed. A late + ``control.error`` surfaces through the existing ``error`` event path. + """ + with _sessions_lock: + live = _sessions.get(sid) + if live is not session: + return + if not isinstance(ack, dict) or ack.get("type") in {"control.error", "error"}: + message = str((ack or {}).get("message") or f"compute-host {route_name} failed") + _emit("error", sid, {"message": f"compression failed: {message}"}) + _status_update(sid, "ready") + return + _announce_compute_host_compress_done(sid, session, ack) + + def _approval_request_payload(data: dict | None) -> dict: """Build the client-safe representation of a pending approval.""" payload = dict(data or {}) @@ -16708,13 +16773,28 @@ def _mirror_slash_side_effects(sid: str, session: dict, command: str) -> str: _MUTATES_WHILE_RUNNING = {"model", "personality", "prompt", "compress"} if _session_uses_compute_host(session) and name in _MUTATES_WHILE_RUNNING: route_name = f"slash.{name}" + is_compress = name == "compress" + _late_session = session + + def _on_late_ack(late: dict, _sid=sid) -> None: + _adopt_late_compute_host_compress_ack(_sid, _late_session, late, route_name=route_name) + try: ack = _send_compute_host_control( sid, route_name=route_name, command=command, wait=True, + **( + {"timeout": _compute_host_compress_wait_seconds(), "on_late_ack": _on_late_ack} + if is_compress + else {} + ), ) + except queue.Empty: + if is_compress: + return "compression still running in the background; the transcript will refresh when it finishes" + return f"compute-host {route_name} failed: timed out" except Exception as exc: return f"compute-host {route_name} failed: {exc}" if ack.get("type") in {"control.error", "error"}: