fix(tui-gateway): adopt late compute-host compress acks instead of a false 120s timeout (#97948)

Manual /compress on a compute-host (turn_isolation) session blocked its RPC
waiter for a hard-coded 120s, answered error 5019, and then DROPPED the
host's late `control.ack`: HostSupervisor.control() popped the pending
queue in `finally`, so `_handle_host_frame` had nothing to deliver to. The
host kept compressing, succeeded minutes later, rotated the session — and
the gateway session never mirrored the new session_key/history_version and
the desktop never refreshed its transcript.

- host_supervisor: `control(..., on_late_ack=)` leaves a one-shot handler
  registered when the waiter times out; control.ack/control.error/error
  frames for that request_id fire it (bounded: 30min TTL, cap 64). A host
  crash fails outstanding handlers with a synthetic control.error.
- server: `_compute_host_compress_wait_seconds()` derives the wait from
  `compression.context_total_ceiling_seconds` (+30s slack, floor 120s,
  cap 630s) instead of the literal 120. `_adopt_late_compute_host_compress_ack`
  applies the metadata mirror and emits the same `session.info` a normal
  compress does plus the existing `status.update kind=compacted` edge; a
  late error goes out through the existing `error` event.
- session.compress / slash.compress (methods_tools + _mirror_slash_side_effects):
  on waiter timeout answer `status: pending` (not 5019) and register the
  late-ack handler.
- desktop: SESSION_COMPRESS_TIMEOUT_MS 120s -> 660s (above the gateway cap);
  `status: 'pending'` renders as an info notice, not `error:`; the
  `compacted` status edge rehydrates an idle active session's transcript
  (mid-turn compaction still defers to the turn settle path).

Minimal extraction of the design in #99630 by @vsd2807 (design trace by
@andrexibiza and @JoaoMarcos44 in the #97948 thread); no new DB tables,
modules, or polling protocol.

Refs #97948

Co-authored-by: VVV <vaibhavdahiya28@gmail.com>

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' })

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.

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<string, unknown>[] = []
 
     const requestGateway = vi.fn(async (method: string, _params?: Record<string, unknown>, _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'))

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

apps/desktop/src/app/types.ts

@@ -66,6 +66,10 @@ export interface SessionCompressResponse {
     usage?: Partial<UsageStats>
   }
   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?: {

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

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

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

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}")

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}")

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"}: