diff --git a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts index cd5567fd0d97..6b29fedcba0e 100644 --- a/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts +++ b/apps/desktop/src/app/session/hooks/use-message-stream/gateway-event.ts @@ -599,6 +599,12 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { return state } + // Prefer the gateway-reported turn_started_at so the timer + // survives session switches and session.info heartbeats. + const gatewayTurnStartedAt = + typeof payload!.turn_started_at === 'number' && payload!.turn_started_at > 0 + ? payload!.turn_started_at * 1000 + : null return { ...state, busy, @@ -606,7 +612,7 @@ export function useGatewayEventHandler(deps: GatewayEventDeps) { // message.start (e.g. resuming an already-running session // that never replays its start event). turnLive: true, - turnStartedAt: state.turnStartedAt ?? Date.now() + turnStartedAt: state.turnStartedAt ?? gatewayTurnStartedAt ?? Date.now() } } diff --git a/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx b/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx index 02aecbd86b45..e48ee69f92a7 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx +++ b/apps/desktop/src/app/session/hooks/use-session-actions.test.tsx @@ -1,7 +1,8 @@ +import { useStore } from '@nanostores/react' import { act, cleanup, render, waitFor } from '@testing-library/react' import type { MutableRefObject } from 'react' -import { useEffect } from 'react' -import { afterEach, describe, expect, it, vi } from 'vitest' +import { useEffect, useRef } from 'react' +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' import { $terminalTakeover, setTerminalTakeover } from '@/app/right-sidebar/store' import { noteActiveTreeGroup, revealTreePane } from '@/components/pane-shell/tree/store' @@ -22,8 +23,11 @@ import { $newChatWorkspaceTarget, $resumeFailedSessionId, $selectedStoredSessionId, + $turnStartedAt, setActiveSessionId, setActiveSessionStoredIdRotation, + setAwaitingResponse, + setBusy, setCurrentCwd, setCurrentFastMode, setCurrentModel, @@ -33,14 +37,17 @@ import { setNewChatWorkspaceTarget, setResumeFailedSessionId, setSelectedStoredSessionId, - setSessions + setSessions, + setTurnStartedAt } from '@/store/session' import { $sessionTiles } from '@/store/session-states' +import sessionResumeActiveTurn from '../../../../../../tests/fixtures/session-resume-active-turn.json' import { sessionRoute } from '../../routes' import type { ClientSessionState } from '../../types' import { useSessionActions } from './use-session-actions' +import { useSessionStateCache } from './use-session-state-cache' vi.mock('@/hermes', async importOriginal => ({ ...(await importOriginal>()), @@ -631,8 +638,10 @@ function ResumeHarness({ selectedStoredSessionIdRef: ref(selectedStoredSessionId), sessionStateByRuntimeIdRef: stateMapRef, syncSessionStateToView: vi.fn(), - updateSessionState: (sessionId, updater) => { - const current = stateMapRef.current.get(sessionId) ?? ({} as ClientSessionState) + updateSessionState: (sessionId, updater, storedSessionId) => { + // Full default shape (not a bare {} cast) so seeded/derived fields like + // turnStartedAt behave as in production state updates. + const current = stateMapRef.current.get(sessionId) ?? createClientSessionState(storedSessionId ?? null) const next = updater(current) stateMapRef.current.set(sessionId, next) @@ -649,6 +658,49 @@ function ResumeHarness({ return null } +function ResumeTimerHarness({ + onReady, + requestGateway +}: { + onReady: (resume: (storedSessionId: string, replaceRoute?: boolean) => Promise) => void + requestGateway: (method: string, params?: Record) => Promise +}) { + const activeSessionId = useStore($activeSessionId) + const busyRef = useRef(false) + const cache = useSessionStateCache({ + activeSessionId, + busyRef, + selectedStoredSessionId: null, + setAwaitingResponse, + setBusy, + setMessages + }) + const actions = useSessionActions({ + activeSessionId, + activeSessionIdRef: cache.activeSessionIdRef, + busyRef, + creatingSessionRef: useRef(false), + ensureSessionState: cache.ensureSessionState, + getRouteToken: () => 'timer-contract', + navigate: vi.fn() as never, + requestGateway, + resetViewSync: cache.resetViewSync, + runtimeIdByStoredSessionIdRef: cache.runtimeIdByStoredSessionIdRef, + selectedStoredSessionId: null, + selectedStoredSessionIdRef: cache.selectedStoredSessionIdRef, + sessionStateByRuntimeIdRef: cache.sessionStateByRuntimeIdRef, + syncSessionStateToView: cache.syncSessionStateToView, + getRoutedStoredSessionId: () => null, + updateSessionState: cache.updateSessionState + }) + + useEffect(() => { + onReady(actions.resumeSession) + }, [actions.resumeSession, onReady]) + + return null +} + describe('resumeSession failure recovery', () => { afterEach(() => { cleanup() @@ -793,6 +845,7 @@ describe('resumeSession failure recovery', () => { message_count: compressedRuntimeMessages.length, messages: compressedRuntimeMessages, running: true, + turn_started_at: 1_700_000_000, inflight: { user: 'current prompt', assistant: 'partial answer', @@ -823,6 +876,7 @@ describe('resumeSession failure recovery', () => { expect(renderedMessages).toContain('current prompt') expect(renderedMessages).toContain('partial answer') expect(renderedMessages).toContain('newest prompt') + expect(resumedState?.turnStartedAt).toBe(1_700_000_000_000) }) it('preserves a runtime-cache delta that arrives while cold resume waits for REST', async () => { @@ -1108,6 +1162,85 @@ describe('resumeSession failure recovery', () => { }) }) +describe('session.resume turn timer contract', () => { + beforeEach(() => { + vi.spyOn(window, 'requestAnimationFrame').mockImplementation((callback: FrameRequestCallback) => { + callback(0) + + return null as unknown as number + }) + setActiveSessionId(null) + setAwaitingResponse(false) + setBusy(false) + setMessages([]) + setSessions([]) + setTurnStartedAt(null) + }) + + afterEach(() => { + cleanup() + setActiveSessionId(null) + setAwaitingResponse(false) + setBusy(false) + setMessages([]) + setSessions([]) + setTurnStartedAt(null) + vi.restoreAllMocks() + }) + + async function resumeFrom(response: unknown): Promise { + const requestGateway = vi.fn(async (method: string) => { + if (method === 'session.resume') { + // Model the JSON-RPC serialization/deserialization boundary. The shared + // fixture is asserted against the real gateway response in Python. + return JSON.parse(JSON.stringify(response)) as never + } + + return {} as never + }) + vi.mocked(getAllSessionMessages).mockResolvedValue({ messages: [], session_id: 'stored-running' } as never) + + let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise) | null = null + render( (resume = ready)} requestGateway={requestGateway} />) + await waitFor(() => expect(resume).not.toBeNull()) + await act(async () => { + await resume!('stored-running', true) + }) + } + + it('restores the canonical gateway turn timestamp in milliseconds', async () => { + await resumeFrom(sessionResumeActiveTurn) + + expect($turnStartedAt.get()).toBe(sessionResumeActiveTurn.turn_started_at * 1000) + }) + + it('clears a stale timer when the gateway response is not running', async () => { + setTurnStartedAt(1_600_000_000_000) + + await resumeFrom({ ...sessionResumeActiveTurn, running: false }) + + expect($turnStartedAt.get()).toBeNull() + }) + + it('clears a stale timer when the running gateway response omits its timestamp', async () => { + const missingTimestamp: Record = JSON.parse(JSON.stringify(sessionResumeActiveTurn)) + delete missingTimestamp.turn_started_at + setTurnStartedAt(1_600_000_000_000) + + await resumeFrom(missingTimestamp) + + expect($turnStartedAt.get()).toBeNull() + }) + + it('clears a stale timer when the running gateway response has a non-numeric timestamp', async () => { + setTurnStartedAt(1_600_000_000_000) + + await resumeFrom({ ...sessionResumeActiveTurn, turn_started_at: 'not-a-timestamp' }) + + expect($turnStartedAt.get()).toBeNull() + }) +}) + function BranchHarness({ activeSessionId = null, navigate = vi.fn(), @@ -1733,6 +1866,64 @@ describe('resumeSession warm-cache mapping integrity', () => { expect(resumedState?.messages[0]?.attachmentRefs).toEqual(['@image:/tmp/photo.png']) }) + it('restores the warm reconnect turn clock from session.activate', async () => { + const turnStartedAtSeconds = 1_700_000_123 + const runtimeIdByStoredSessionIdRef: MutableRefObject> = { + current: new Map([['stored-A', 'rt-A']]) + } + const cachedState = clientState('stored-A') + cachedState.busy = true + cachedState.turnStartedAt = null + const sessionStateByRuntimeIdRef: MutableRefObject> = { + current: new Map([['rt-A', cachedState]]) + } + + const requestGateway = vi.fn(async (method: string) => { + if (method === 'session.activate') { + return { + session_id: 'rt-A', + session_key: 'stored-A', + resumed: 'stored-A', + message_count: 0, + messages: [], + running: true, + turn_started_at: turnStartedAtSeconds, + inflight: { + user: 'current prompt', + assistant: 'partial answer', + streaming: true + }, + info: {} + } as never + } + + return {} as never + }) + + vi.mocked(getAllSessionMessages).mockResolvedValue({ messages: [], session_id: 'stored-A' } as never) + + let resumedState: ClientSessionState | undefined + let resume: ((storedSessionId: string, replaceRoute?: boolean) => Promise) | null = null + render( + (resume = ready)} + onStateUpdate={(_sessionId, state) => (resumedState = state)} + requestGateway={requestGateway} + runtimeIdByStoredSessionIdRef={runtimeIdByStoredSessionIdRef} + sessionStateByRuntimeIdRef={sessionStateByRuntimeIdRef} + /> + ) + await waitFor(() => expect(resume).not.toBeNull()) + await resume!('stored-A', true) + + expect(resumedState).toMatchObject({ + awaitingResponse: true, + busy: true, + turnStartedAt: turnStartedAtSeconds * 1000 + }) + expect(JSON.stringify(resumedState?.messages)).toContain('partial answer') + }) + it('repairs an idle warm cache from a divergent equal-length persisted transcript', async () => { const runtimeIdByStoredSessionIdRef: MutableRefObject> = { current: new Map([['stored-A', 'rt-A']]) diff --git a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts index 58c14205c555..85a53266d823 100644 --- a/apps/desktop/src/app/session/hooks/use-session-actions/index.ts +++ b/apps/desktop/src/app/session/hooks/use-session-actions/index.ts @@ -854,6 +854,11 @@ export function useSessionActions({ Boolean(sessionStateByRuntimeIdRef.current.get(cachedRuntimeId)?.busy) ) + const activatedTurnStartedAt = + typeof activated.turn_started_at === 'number' && activated.turn_started_at > 0 + ? activated.turn_started_at * 1000 + : null + // The persisted REST transcript is the display authority: a live // runtime may carry only the agent's compressed context projection, // which is intentionally smaller than the user-visible conversation. @@ -927,7 +932,8 @@ export function useSessionActions({ // Adopting someone else's turn: we'll stream its reply // without ever having received its prompt, so the settle // path must not take the "I saw it all" shortcut. - adoptedRunningTurn: state.adoptedRunningTurn || running + adoptedRunningTurn: state.adoptedRunningTurn || running, + turnStartedAt: running ? (activatedTurnStartedAt ?? state.turnStartedAt ?? Date.now()) : null }), storedSessionId ) @@ -1195,6 +1201,14 @@ export function useSessionActions({ patchSessionWorkspace(storedSessionId, runtimeInfo?.cwd) + // Preserve the turn-elapsed timer across cold resume: the gateway + // reports when the in-flight turn started so the desktop can restore + // the clock instead of resetting it to 0:00. + const resumedTurnStartedAt = + typeof resumed.turn_started_at === 'number' && resumed.turn_started_at > 0 + ? resumed.turn_started_at * 1000 + : null + updateSessionState( resumed.session_id, state => ({ @@ -1214,10 +1228,13 @@ export function useSessionActions({ // still mid-turn; a settled recovery keeps the stream idle. streamId: resumedRunning ? inFlightRecovery.streamId : null, turnStartedAt: resumedRunning - ? (inFlightRecovery.turnStartedAt ?? state.turnStartedAt ?? Date.now()) - : state.turnStartedAt + ? (inFlightRecovery.turnStartedAt ?? resumedTurnStartedAt) + : null } - : {}) + : { + turnStartedAt: + resumedRunning && resumedTurnStartedAt !== null ? resumedTurnStartedAt : null + }) }), storedSessionId ) diff --git a/apps/desktop/src/lib/chat-messages.ts b/apps/desktop/src/lib/chat-messages.ts index 8f26f8c00067..0a5b0554efad 100644 --- a/apps/desktop/src/lib/chat-messages.ts +++ b/apps/desktop/src/lib/chat-messages.ts @@ -71,6 +71,7 @@ export type GatewayEventPayload = { approval_mode?: string yolo?: boolean running?: boolean + turn_started_at?: number | null cwd?: string branch?: string terminal_backend?: string diff --git a/apps/desktop/src/types/hermes.ts b/apps/desktop/src/types/hermes.ts index 97b0039ec590..b6291e845539 100644 --- a/apps/desktop/src/types/hermes.ts +++ b/apps/desktop/src/types/hermes.ts @@ -658,6 +658,8 @@ export interface SessionResumeResponse { session_key?: string started_at?: number status?: string + /** Epoch seconds the current turn started, or null when idle. */ + turn_started_at?: number | null } export interface SessionRuntimeInfo { diff --git a/tests/fixtures/session-resume-active-turn.json b/tests/fixtures/session-resume-active-turn.json new file mode 100644 index 000000000000..048f30039f4b --- /dev/null +++ b/tests/fixtures/session-resume-active-turn.json @@ -0,0 +1,29 @@ +{ + "info": { + "cwd": "/workspace", + "lazy": false, + "model": "test/model", + "skills": {}, + "tools": {} + }, + "inflight": { + "assistant": "partial answer", + "streaming": true, + "user": "current prompt" + }, + "message_count": 1, + "messages": [ + { + "role": "user", + "text": "earlier prompt" + } + ], + "messages_omitted": false, + "resumed": "stored-running", + "running": true, + "session_id": "rt-running", + "session_key": "stored-running", + "started_at": 1700000000.0, + "status": "working", + "turn_started_at": 1700000123.5 +} diff --git a/tests/test_tui_gateway_server.py b/tests/test_tui_gateway_server.py index bfd1bfd66f8a..6e5b6c43af01 100644 --- a/tests/test_tui_gateway_server.py +++ b/tests/test_tui_gateway_server.py @@ -10206,6 +10206,21 @@ def test_session_info_reports_pending_model_switch(monkeypatch): assert server._session_info(agent, session)["model"] == "old/model" +def test_session_info_includes_turn_started_at(): + agent = types.SimpleNamespace(tools=[], model="", provider="") + session = { + "history": [], + "inflight_turn": {"started_at": 1_700_000_123.5}, + "running": True, + } + + assert server._session_info(agent, session)["turn_started_at"] == 1_700_000_123.5 + + session["inflight_turn"] = None + session["running"] = False + assert server._session_info(agent, session)["turn_started_at"] is None + + # --------------------------------------------------------------------------- # History-mutating commands must reject while session.running is True. # Without these guards, prompt.submit's post-run history write either @@ -13956,6 +13971,9 @@ def _emit(event, sid, payload=None): "streaming": True, "user": "write a long answer", } + turn_started_at = resp["result"]["turn_started_at"] + assert turn_started_at == server._sessions["sid-live"]["inflight_turn"]["started_at"] + assert turn_started_at > 0 assert resp["result"]["messages"] == [] release.set() @@ -13968,6 +13986,7 @@ def _emit(event, sid, payload=None): } ) assert completed["result"].get("inflight") is None + assert completed["result"]["turn_started_at"] is None assert completed["result"]["messages"] == [ {"role": "user", "text": "write a long answer"}, {"role": "assistant", "text": "partial answer complete"}, diff --git a/tests/tui_gateway/test_protocol.py b/tests/tui_gateway/test_protocol.py index b6b265c777ee..f04378084865 100644 --- a/tests/tui_gateway/test_protocol.py +++ b/tests/tui_gateway/test_protocol.py @@ -7,6 +7,7 @@ import time import types from unittest.mock import MagicMock, patch +from pathlib import Path import pytest @@ -564,6 +565,62 @@ def reopen_session(self, sid): assert reopened == ["transient-guard-session"] +def test_session_resume_active_turn_payload_matches_desktop_fixture(server, monkeypatch): + """A live resume serializes the exact timer payload consumed by Desktop.""" + fixture = json.loads( + (Path(__file__).parents[1] / "fixtures" / "session-resume-active-turn.json").read_text( + encoding="utf-8" + ) + ) + + class _DB: + def get_session(self, session_id): + return {"id": session_id} + + def get_session_by_title(self, _title): + return None + + def resolve_resume_session_id(self, session_id): + return session_id + + active_turn = { + "assistant": "partial answer", + "started_at": fixture["turn_started_at"], + "streaming": True, + "user": "current prompt", + } + server._sessions[fixture["session_id"]] = { + "agent": types.SimpleNamespace(session_id=fixture["session_key"]), + "created_at": fixture["started_at"], + "history": [{"content": "earlier prompt", "role": "user"}], + "history_lock": threading.Lock(), + "inflight_turn": active_turn, + "running": True, + "session_key": fixture["session_key"], + } + monkeypatch.setattr(server, "_get_db", lambda: _DB()) + monkeypatch.setattr(server, "_session_info", lambda _agent: fixture["info"]) + + # JSON round-trip the real RPC envelope: the desktop fixture must stay + # faithful to what the gateway actually serializes, not a copied shape. + response = json.loads( + json.dumps( + server.handle_request( + { + "id": "resume-running", + "method": "session.resume", + "params": {"session_id": fixture["session_key"]}, + } + ) + ) + ) + result = response["result"] + + assert result["running"] is True + assert result["turn_started_at"] == active_turn["started_at"] + assert result == fixture + + def test_enforce_session_cap_evicts_oldest_detached_only(server, monkeypatch): """The LRU cap frees the least-recently-active DETACHED sessions when over the limit, and never a live-transport / running / mid-build one.""" diff --git a/tui_gateway/server.py b/tui_gateway/server.py index 23b526652168..267a6839a57d 100644 --- a/tui_gateway/server.py +++ b/tui_gateway/server.py @@ -5450,6 +5450,16 @@ def _session_info(agent, session: dict | None = None) -> dict: pending_switch = (session or {}).get("pending_model_switch") or {} pending_model = str(pending_switch.get("display_model") or "").strip() pending_provider = str(pending_switch.get("display_provider") or "").strip() + # Epoch seconds the current turn started, or None when idle. Lets the + # desktop preserve the turn-elapsed timer across session switches (cold + # resume path) instead of resetting it to 0:00. + inflight = (session or {}).get("inflight_turn") + turn_started_at = ( + float(inflight["started_at"]) + if isinstance(inflight, dict) and inflight.get("started_at") + else None + ) + info: dict = { "model": pending_model or mirror.get("model", getattr(agent, "model", "")), "provider": pending_provider @@ -5467,6 +5477,7 @@ def _session_info(agent, session: dict | None = None) -> dict: "terminal_backend": _effective_terminal_backend(), "personality": str(personality or ""), "running": bool((session or {}).get("running")), + "turn_started_at": turn_started_at, "title": _session_live_title(session or {}, session_key) if session_key else "", "stored_session_id": session_key or "", "desktop_contract": DESKTOP_BACKEND_CONTRACT, @@ -8552,6 +8563,12 @@ def _live_session_payload( inflight = _inflight_snapshot(session) queued = _queued_prompt_snapshot(session) running = bool(session.get("running")) + inflight_turn = session.get("inflight_turn") + turn_started_at = ( + float(inflight_turn["started_at"]) + if isinstance(inflight_turn, dict) and inflight_turn.get("started_at") + else None + ) # Prefer the persisted display lineage (candidate-inclusive) so this payload # matches the eager session.resume + REST transcript. Use the session's # profile-aware DB (not launch ``_get_db()``): app-global remote profile @@ -8571,6 +8588,7 @@ def _live_session_payload( "messages": [] if omit_messages else _history_to_messages(history), "messages_omitted": omit_messages, "running": running, + "turn_started_at": turn_started_at, "session_id": sid, "session_key": _session_lookup_key(session, fallback=sid), "started_at": float(session.get("created_at") or time.time()),