From 5e6e6a91cf3e34cb2e7582d0da56c0c1f8739471 Mon Sep 17 00:00:00 2001 From: Bohdan Triapitsyn Date: Thu, 25 Dec 2025 02:38:38 +0200 Subject: [PATCH] feat: enhance session activity tracking with directory support and cooldown logic --- .../desktop/src-tauri/src/session_activity.rs | 286 ++++++++++++++---- packages/ui/src/hooks/useEventStream.ts | 200 +++++++++--- packages/ui/src/lib/opencode/client.ts | 17 +- packages/ui/src/stores/sessionStore.ts | 23 +- packages/web/server/index.js | 28 ++ 5 files changed, 439 insertions(+), 115 deletions(-) diff --git a/packages/desktop/src-tauri/src/session_activity.rs b/packages/desktop/src-tauri/src/session_activity.rs index dfedd9d0..68bbe703 100644 --- a/packages/desktop/src-tauri/src/session_activity.rs +++ b/packages/desktop/src-tauri/src/session_activity.rs @@ -24,6 +24,13 @@ struct EventEnvelope { properties: Value, } +#[derive(Deserialize)] +struct MultiplexedEventEnvelope { + #[serde(default)] + directory: Option, + payload: EventEnvelope, +} + #[derive(Clone, Debug, PartialEq)] pub enum ActivityPhase { Idle, @@ -31,6 +38,12 @@ pub enum ActivityPhase { Cooldown, } +#[derive(Clone, Debug)] +enum SseScope { + Global, + Directory(std::path::PathBuf), +} + pub fn spawn_session_activity_tracker( app: AppHandle, runtime: DesktopRuntime, @@ -85,39 +98,8 @@ async fn run_once( }; let prefix = opencode.api_prefix(); - let mut url = format!("http://127.0.0.1:{port}{}/event", prefix); - - if let Some(dir) = opencode.get_working_directory().to_str().map(|s| s.to_string()) { - let mut parsed = reqwest::Url::parse(&url)?; - parsed - .query_pairs_mut() - .append_pair("directory", &dir); - url = parsed.to_string(); - } - - debug!("[desktop:activity] Connecting SSE for activity phases: {url}"); - - let response = client - .get(&url) - .header("accept", "text/event-stream") - .header("accept-encoding", "identity") - .send() - .await?; - - debug!( - "[desktop:activity] SSE response status={} headers={:?}", - response.status(), - response.headers() - ); - - if !response.status().is_success() { - warn!( - "[desktop:activity] SSE connect failed with status {}", - response.status() - ); - tokio::time::sleep(Duration::from_secs(2)).await; - return Ok(()); - } + let base = format!("http://127.0.0.1:{port}{prefix}"); + let (response, scope) = connect_activity_sse(runtime, client, &base).await?; use tokio::io::AsyncBufReadExt; @@ -130,12 +112,28 @@ async fn run_once( loop { buf.clear(); - let bytes_read = match reader.read_until(b'\n', &mut buf).await { - Ok(n) => n, - Err(err) => { + let bytes_read = match tokio::time::timeout(Duration::from_secs(2), reader.read_until(b'\n', &mut buf)).await + { + Ok(Ok(n)) => n, + Ok(Err(err)) => { warn!("[desktop:activity] Read error in SSE stream: {err:?}"); return Err(err.into()); } + Err(_) => { + // No data received recently; if we are connected to a directory-scoped stream and the working directory + // has changed, reconnect so activity tracking follows the new directory. + if let SseScope::Directory(connected_dir) = &scope { + let current_dir = opencode.get_working_directory(); + if current_dir != *connected_dir { + debug!( + "[desktop:activity] Working directory changed; reconnecting activity SSE (from {:?} to {:?})", + connected_dir, current_dir + ); + return Ok(()); + } + } + continue; + } }; if bytes_read == 0 { break; @@ -156,12 +154,10 @@ async fn run_once( let raw = data_lines.join("\n"); data_lines.clear(); - match serde_json::from_str::(&raw) { - Ok(event) => handle_event(app, event, phases.clone(), cooldowns.clone()).await, - Err(err) => { - warn!("[desktop:activity] Failed to parse SSE data: {err}; raw={raw}"); - } - } + match parse_event_envelope(&raw) { + Ok((event, _directory)) => handle_event(app, event, phases.clone(), cooldowns.clone()).await, + Err(err) => warn!("[desktop:activity] Failed to parse SSE data: {err}; raw={raw}"), + }; continue; } @@ -173,6 +169,82 @@ async fn run_once( Ok(()) } +fn parse_event_envelope(raw: &str) -> Result<(EventEnvelope, Option)> { + if let Ok(event) = serde_json::from_str::(raw) { + return Ok((event, None)); + } + + let multiplexed = serde_json::from_str::(raw)?; + Ok((multiplexed.payload, multiplexed.directory)) +} + +async fn connect_activity_sse( + runtime: &DesktopRuntime, + client: &Client, + base: &str, +) -> Result<(reqwest::Response, SseScope)> { + let opencode = runtime.opencode_manager(); + + let global_url = format!("{base}/global/event"); + match try_connect_sse(client, &global_url, "[desktop:activity]").await { + Ok(response) => { + debug!("[desktop:activity] Using SSE endpoint: {global_url}"); + return Ok((response, SseScope::Global)); + } + Err(err) => { + debug!( + "[desktop:activity] SSE endpoint unavailable: {global_url} ({err:?}); falling back" + ); + } + } + + let event_url = format!("{base}/event"); + match try_connect_sse(client, &event_url, "[desktop:activity]").await { + Ok(response) => { + debug!("[desktop:activity] Using SSE endpoint: {event_url}"); + return Ok((response, SseScope::Global)); + } + Err(err) => { + debug!( + "[desktop:activity] SSE endpoint unavailable: {event_url} ({err:?}); falling back" + ); + } + } + + let working_dir = opencode.get_working_directory(); + let directory = working_dir.to_string_lossy().to_string(); + let mut parsed = reqwest::Url::parse(&event_url)?; + parsed.query_pairs_mut().append_pair("directory", &directory); + let directory_url = parsed.to_string(); + + let response = try_connect_sse(client, &directory_url, "[desktop:activity]").await?; + debug!("[desktop:activity] Using directory-scoped SSE endpoint: {directory_url}"); + Ok((response, SseScope::Directory(working_dir))) +} + +async fn try_connect_sse(client: &Client, url: &str, log_prefix: &str) -> Result { + debug!("{log_prefix} Connecting SSE: {url}"); + + let response = client + .get(url) + .header("accept", "text/event-stream") + .header("accept-encoding", "identity") + .send() + .await?; + + debug!( + "{log_prefix} SSE response status={} headers={:?}", + response.status(), + response.headers() + ); + + if !response.status().is_success() { + anyhow::bail!("SSE connect failed with status {}", response.status()); + } + + Ok(response) +} + async fn handle_event( app: &AppHandle, event: EventEnvelope, @@ -201,6 +273,16 @@ async fn handle_event( set_phase(app, &id, phase, phases.clone(), cooldowns.clone()).await; } } + "session.idle" => { + let session_id = event + .properties + .get("sessionID") + .and_then(Value::as_str) + .map(|s| s.to_string()); + if let Some(id) = session_id { + set_phase(app, &id, ActivityPhase::Idle, phases.clone(), cooldowns.clone()).await; + } + } "message.updated" => { if let Some(info) = event.properties.get("info") { let role = info.get("role").and_then(Value::as_str).unwrap_or_default(); @@ -219,37 +301,109 @@ async fn handle_event( .map(|s| s.to_string()); if let Some(id) = session_id { - // If current phase is busy, move to cooldown for 2s then idle - let current = { phases.lock().await.get(&id).cloned() }; - if matches!(current, Some(ActivityPhase::Busy)) { - set_phase(app, &id, ActivityPhase::Cooldown, phases.clone(), cooldowns.clone()).await; - - let app_clone = app.clone(); - let phases_clone = phases.clone(); - let cooldowns_clone = cooldowns.clone(); - let id_clone = id.clone(); - let handle = tauri::async_runtime::spawn(async move { - tokio::time::sleep(Duration::from_secs(2)).await; - let current = { phases_clone.lock().await.get(&id_clone).cloned() }; - if matches!(current, Some(ActivityPhase::Cooldown)) { - set_phase(&app_clone, &id_clone, ActivityPhase::Idle, phases_clone, cooldowns_clone).await; - } - }); - - // Store cooldown handle to cancel if phase changes earlier - let mut cd = cooldowns.lock().await; - if let Some(prev) = cd.remove(&id) { - prev.abort(); - } - cd.insert(id, handle); - } + enter_cooldown_if_busy(app, &id, phases.clone(), cooldowns.clone()).await; } } } + "message.part.updated" => { + let Some(info) = event.properties.get("info") else { + return; + }; + + let role = info.get("role").and_then(Value::as_str).unwrap_or_default(); + if role != "assistant" { + return; + } + + let session_id = info + .get("sessionID") + .and_then(Value::as_str) + .map(|s| s.to_string()); + + let Some(id) = session_id else { + return; + }; + + // Mark session busy when we see assistant parts streaming (covers cases where session.status is missing). + if is_streaming_assistant_part(&event.properties) { + set_phase(app, &id, ActivityPhase::Busy, phases.clone(), cooldowns.clone()).await; + } + + // Derive cooldown from "step-finish reason=stop" marker when present. + if is_stop_step_finish_part(&event.properties) { + enter_cooldown_if_busy(app, &id, phases.clone(), cooldowns.clone()).await; + } + } _ => {} } } +fn is_streaming_assistant_part(properties: &Value) -> bool { + let Some(part) = properties.get("part") else { + return false; + }; + let part_type = part.get("type").and_then(Value::as_str).unwrap_or_default(); + matches!( + part_type, + "step-start" | "text" | "tool" | "reasoning" | "file" | "patch" + ) +} + +fn is_stop_step_finish_part(properties: &Value) -> bool { + let Some(part) = properties.get("part") else { + return false; + }; + let part_type = part.get("type").and_then(Value::as_str); + let reason = part.get("reason").and_then(Value::as_str); + part_type == Some("step-finish") && reason == Some("stop") +} + +async fn enter_cooldown_if_busy( + app: &AppHandle, + session_id: &str, + phases: Arc>>, + cooldowns: Arc>>>, +) { + let current = { phases.lock().await.get(session_id).cloned() }; + if !matches!(current, Some(ActivityPhase::Busy)) { + return; + } + + set_phase( + app, + session_id, + ActivityPhase::Cooldown, + phases.clone(), + cooldowns.clone(), + ) + .await; + + let app_clone = app.clone(); + let phases_clone = phases.clone(); + let cooldowns_clone = cooldowns.clone(); + let id_clone = session_id.to_string(); + let handle = tauri::async_runtime::spawn(async move { + tokio::time::sleep(Duration::from_secs(2)).await; + let current = { phases_clone.lock().await.get(&id_clone).cloned() }; + if matches!(current, Some(ActivityPhase::Cooldown)) { + set_phase( + &app_clone, + &id_clone, + ActivityPhase::Idle, + phases_clone, + cooldowns_clone, + ) + .await; + } + }); + + let mut cd = cooldowns.lock().await; + if let Some(prev) = cd.remove(session_id) { + prev.abort(); + } + cd.insert(session_id.to_string(), handle); +} + async fn set_phase( app: &AppHandle, session_id: &str, diff --git a/packages/ui/src/hooks/useEventStream.ts b/packages/ui/src/hooks/useEventStream.ts index 9901cb1e..eff64bce 100644 --- a/packages/ui/src/hooks/useEventStream.ts +++ b/packages/ui/src/hooks/useEventStream.ts @@ -263,6 +263,8 @@ export const useEventStream = () => { const sessionCooldownTimersRef = React.useRef>(new Map()); const sessionActivityPhaseRef = React.useRef>(new Map()); + const sessionStatusLastRefreshAtRef = React.useRef(0); + const sessionStatusRefreshInFlightRef = React.useRef | null>(null); const currentSessionIdRef = React.useRef(currentSessionId); React.useEffect(() => { currentSessionIdRef.current = currentSessionId; @@ -317,8 +319,11 @@ export const useEventStream = () => { }, [loadSessions]); const updateSessionActivityPhase = React.useCallback((sessionId: string, phase: 'idle' | 'busy' | 'cooldown') => { - const currentPhase = sessionActivityPhaseRef.current.get(sessionId); - if (currentPhase === phase) return; + const storePhase = useSessionStore.getState().sessionActivityPhase?.get(sessionId); + if (storePhase === phase) { + sessionActivityPhaseRef.current = new Map(useSessionStore.getState().sessionActivityPhase ?? new Map()); + return; + } const existingTimer = sessionCooldownTimersRef.current.get(sessionId); if (existingTimer) { @@ -326,16 +331,20 @@ export const useEventStream = () => { sessionCooldownTimersRef.current.delete(sessionId); } - sessionActivityPhaseRef.current.set(sessionId, phase); - useSessionStore.setState({ sessionActivityPhase: new Map(sessionActivityPhaseRef.current) }); + const next = new Map(useSessionStore.getState().sessionActivityPhase ?? new Map()); + next.set(sessionId, phase); + sessionActivityPhaseRef.current = next; + useSessionStore.setState({ sessionActivityPhase: next }); if (phase === 'cooldown') { const timer = setTimeout(() => { sessionCooldownTimersRef.current.delete(sessionId); - const current = sessionActivityPhaseRef.current.get(sessionId); + const current = useSessionStore.getState().sessionActivityPhase?.get(sessionId); if (current === 'cooldown') { - sessionActivityPhaseRef.current.set(sessionId, 'idle'); - useSessionStore.setState({ sessionActivityPhase: new Map(sessionActivityPhaseRef.current) }); + const latest = new Map(useSessionStore.getState().sessionActivityPhase ?? new Map()); + latest.set(sessionId, 'idle'); + sessionActivityPhaseRef.current = latest; + useSessionStore.setState({ sessionActivityPhase: latest }); } }, 2000); sessionCooldownTimersRef.current.set(sessionId, timer); @@ -343,19 +352,94 @@ export const useEventStream = () => { }, []); const refreshSessionActivityStatus = React.useCallback(async () => { - try { - const statusMap = await opencodeClient.getSessionStatus(); - if (!statusMap) return; + const now = Date.now(); + if (sessionStatusRefreshInFlightRef.current) { + return sessionStatusRefreshInFlightRef.current; + } + if (now - sessionStatusLastRefreshAtRef.current < 1500) { + return; + } + sessionStatusLastRefreshAtRef.current = now; + const normalizeDirectory = (value: string | null | undefined): string | null => { + if (typeof value !== 'string') return null; + const trimmed = value.trim(); + if (!trimmed) return null; + const normalized = trimmed.replace(/\\/g, '/'); + return normalized.length > 1 ? normalized.replace(/\/+$/, '') : normalized; + }; + + const resolveSessionDirectoryForStatus = (sessionId: string): string | null => { + try { + const metadata = getWorktreeMetadata?.(sessionId); + const metaPath = normalizeDirectory(metadata?.path ?? null); + if (metaPath) return metaPath; + } catch { + // ignored + } + + const record = sessions.find((entry) => entry.id === sessionId); + const recordPath = normalizeDirectory((record as { directory?: string | null })?.directory ?? null); + return recordPath; + }; + + const applyStatusMap = (statusMap: Record) => { Object.entries(statusMap).forEach(([sessionId, raw]) => { if (!sessionId || !raw) return; - const status = raw as { type?: string }; - const phase: 'idle' | 'busy' = - status.type === 'busy' || status.type === 'retry' ? 'busy' : 'idle'; + const phase: 'idle' | 'busy' = + raw.type === 'busy' || raw.type === 'retry' ? 'busy' : 'idle'; updateSessionActivityPhase(sessionId, phase); }); - } catch { /* ignored */ } - }, [updateSessionActivityPhase]); + }; + + const task = (async (): Promise => { + try { + const globalStatusMap = await opencodeClient.getGlobalSessionStatus(); + if (globalStatusMap && Object.keys(globalStatusMap).length > 0) { + applyStatusMap(globalStatusMap); + return; + } + + const directories = new Set(); + sessions.forEach((session) => { + const directory = resolveSessionDirectoryForStatus(session.id); + if (directory) directories.add(directory); + }); + + const effective = normalizeDirectory(effectiveDirectory ?? null); + if (effective) directories.add(effective); + + const queries = Array.from(directories); + if (queries.length === 0) { + // Fall back to scoped status for whatever the OpenCode client currently tracks. + const scoped = await opencodeClient.getSessionStatus(); + if (scoped) { + applyStatusMap(scoped); + } + return; + } + + const results = await Promise.allSettled( + queries.map((directory) => opencodeClient.getSessionStatusForDirectory(directory)) + ); + + const merged: Record = {}; + results.forEach((result) => { + if (result.status !== 'fulfilled' || !result.value) return; + Object.assign(merged, result.value); + }); + + applyStatusMap(merged); + } catch { + // ignored + } + })().finally(() => { + sessionStatusRefreshInFlightRef.current = null; + }); + + sessionStatusRefreshInFlightRef.current = task; + return task; + }, [effectiveDirectory, getWorktreeMetadata, sessions, updateSessionActivityPhase]); const handleEvent = React.useCallback((event: EventData) => { lastEventTimestampRef.current = Date.now(); @@ -685,12 +769,21 @@ export const useEventStream = () => { } } - const messageTime = (message as { time?: { completed?: number } }).time; - const isCompleted = - (message as { role?: string }).role === 'assistant' && - (messageTime?.completed as number | undefined) !== undefined; + const messageTime = (message as { time?: { completed?: unknown } }).time; + const completedCandidate = (messageTime as { completed?: unknown } | undefined)?.completed; + const hasCompletedTimestamp = typeof completedCandidate === 'number' && Number.isFinite(completedCandidate); + const finishCandidate = (message as { finish?: unknown }).finish; + const finish = typeof finishCandidate === 'string' ? finishCandidate : null; - if (isCompleted && (message as { role?: string }).role === 'assistant') { + const stopMarkerPresent = partsArray.some( + (p) => p?.type === 'step-finish' && (p as { reason?: string }).reason === 'stop' + ) || existingStopMarker; + + const shouldFinalizeAssistantMessage = + (message as { role?: string }).role === 'assistant' && + (hasCompletedTimestamp || finish === 'stop' || stopMarkerPresent); + + if (shouldFinalizeAssistantMessage && (message as { role?: string }).role === 'assistant') { const storeState = useSessionStore.getState(); const sessionMessages = storeState.messages.get(sessionId) || []; @@ -707,20 +800,17 @@ export const useEventStream = () => { if (messageId !== latestAssistantMessageId) break; - const stopMarkerPresent = partsArray.some( - (p) => p?.type === 'step-finish' && (p as { reason?: string }).reason === 'stop' - ) || existingStopMarker; if (!stopMarkerPresent && isDesktopRuntimeRef.current) { trackMessage(messageId, 'desktop_completion_without_stop'); break; } const timeCompleted = - (messageTime?.completed as number | undefined) !== undefined - ? (messageTime?.completed as number) + hasCompletedTimestamp + ? (completedCandidate as number) : Date.now(); - if (!messageTime?.completed) { + if (!hasCompletedTimestamp) { updateMessageInfo(sessionId, messageId, { ...message, time: { ...(messageTime ?? {}), completed: timeCompleted }, @@ -801,29 +891,33 @@ export const useEventStream = () => { } } - completeStreamingMessage(sessionId, messageId); + completeStreamingMessage(sessionId, messageId); - // For web/vscode: trigger cooldown only when assistant message has finish === "stop" - // This matches the desktop backend logic in session_activity.rs - if (!isDesktopRuntimeRef.current) { - const finish = (message as { finish?: string }).finish; - if (finish === 'stop') { - const rawCompletedSessionId = (message as { sessionID?: string }).sessionID; - const completedSessionId: string = - typeof rawCompletedSessionId === 'string' && rawCompletedSessionId.length > 0 - ? rawCompletedSessionId - : sessionId; + // For web/vscode: trigger cooldown only when assistant message has finish === "stop" + // (or we can infer a stop marker) to match desktop backend semantics. + if (!isDesktopRuntimeRef.current) { + const finishCandidate = (message as { finish?: unknown }).finish; + const finish = typeof finishCandidate === 'string' ? finishCandidate : null; - const currentPhase = sessionActivityPhaseRef.current.get(completedSessionId); - if (currentPhase === 'busy') { - updateSessionActivityPhase(completedSessionId, 'cooldown'); - } - } - } + const inferredStopMarkerPresent = + Array.isArray(partsArray) && + partsArray.some((part) => { + if (!part || typeof part !== 'object') return false; + const partAny = part as { type?: string; reason?: string }; + return partAny.type === 'step-finish' && partAny.reason === 'stop'; + }); - const rawMessageSessionId = (message as { sessionID?: string }).sessionID; - const messageSessionId: string = - typeof rawMessageSessionId === 'string' && rawMessageSessionId.length > 0 + if (finish === 'stop' || inferredStopMarkerPresent) { + const currentPhase = useSessionStore.getState().sessionActivityPhase?.get(sessionId); + if (currentPhase === 'busy') { + updateSessionActivityPhase(sessionId, 'cooldown'); + } + } + } + + const rawMessageSessionId = (message as { sessionID?: string }).sessionID; + const messageSessionId: string = + typeof rawMessageSessionId === 'string' && rawMessageSessionId.length > 0 ? rawMessageSessionId : sessionId; requestSessionMetadataRefresh(messageSessionId); @@ -1007,6 +1101,13 @@ export const useEventStream = () => { publishStatus('connected', null); checkConnection(); + const hasBusySessions = Array.from(useSessionStore.getState().sessionActivityPhase?.values?.() ?? []).some( + (phase) => phase === 'busy' + ); + if (hasBusySessions) { + void refreshSessionActivityStatus(); + } + if (shouldRefresh) { void bootstrapState('sse_reconnected'); } else { @@ -1069,6 +1170,7 @@ export const useEventStream = () => { handleEvent, effectiveDirectory, updateSessionActivityPhase, + refreshSessionActivityStatus, waitForDesktopBridge, debugConnectionState, bootstrapState @@ -1238,6 +1340,12 @@ export const useEventStream = () => { if (!shouldHoldConnection()) return; const now = Date.now(); + const hasBusySessions = Array.from(useSessionStore.getState().sessionActivityPhase?.values?.() ?? []).some( + (phase) => phase === 'busy' + ); + if (hasBusySessions) { + void refreshSessionActivityStatus(); + } if (now - lastEventTimestampRef.current > 25000) { Promise.resolve().then(async () => { try { diff --git a/packages/ui/src/lib/opencode/client.ts b/packages/ui/src/lib/opencode/client.ts index 7012336b..f5f64ce1 100644 --- a/packages/ui/src/lib/opencode/client.ts +++ b/packages/ui/src/lib/opencode/client.ts @@ -500,12 +500,19 @@ class OpencodeService { async getSessionStatus(): Promise< Record > { + return this.getSessionStatusForDirectory(this.currentDirectory ?? null); + } + + async getSessionStatusForDirectory( + directory: string | null | undefined + ): Promise> { try { const base = this.baseUrl.replace(/\/$/, ""); const url = new URL(`${base}/session/status`); - if (this.currentDirectory && this.currentDirectory.length > 0) { - url.searchParams.set("directory", this.currentDirectory); + const trimmedDirectory = typeof directory === "string" ? directory.trim() : ""; + if (trimmedDirectory.length > 0) { + url.searchParams.set("directory", trimmedDirectory); } const response = await fetch(url.toString(), { @@ -533,6 +540,12 @@ class OpencodeService { } } + async getGlobalSessionStatus(): Promise< + Record + > { + return this.getSessionStatusForDirectory(null); + } + // Permissions async respondToPermission( sessionId: string, diff --git a/packages/ui/src/stores/sessionStore.ts b/packages/ui/src/stores/sessionStore.ts index fe84bb6a..33d9d19c 100644 --- a/packages/ui/src/stores/sessionStore.ts +++ b/packages/ui/src/stores/sessionStore.ts @@ -376,9 +376,30 @@ export const useSessionStore = create()( const validSessionIds = new Set(dedupedSessions.map((session) => session.id)); + const resolveSelectionDirectory = (sessionId: string | null): string | null => { + if (!sessionId) { + return null; + } + const sessionDir = getSessionDirectory(dedupedSessions, sessionId); + if (sessionDir) { + return sessionDir; + } + const persistedDir = getSessionDirectory(stateSnapshot.sessions, sessionId); + if (persistedDir) { + return persistedDir; + } + return null; + }; + + const selectionDirectoryKey = resolveSelectionDirectory(nextCurrentId) ?? normalizedProject ?? projectDirectory ?? null; + if (projectDirectory) { clearInvalidSessionSelection(projectDirectory, validSessionIds); - const storedSelection = getStoredSessionForDirectory(projectDirectory); + } + + if (selectionDirectoryKey) { + clearInvalidSessionSelection(selectionDirectoryKey, validSessionIds); + const storedSelection = getStoredSessionForDirectory(selectionDirectoryKey); if (storedSelection && validSessionIds.has(storedSelection)) { nextCurrentId = storedSelection; } diff --git a/packages/web/server/index.js b/packages/web/server/index.js index cf6300ee..11532a98 100644 --- a/packages/web/server/index.js +++ b/packages/web/server/index.js @@ -739,6 +739,34 @@ function deriveSessionActivity(payload) { } } + if (payload.type === 'message.updated') { + const info = payload.properties?.info; + const sessionId = info?.sessionID; + const role = info?.role; + const finish = info?.finish; + if (typeof sessionId === 'string' && sessionId.length > 0 && role === 'assistant' && finish === 'stop') { + return { sessionId, phase: 'cooldown' }; + } + } + + if (payload.type === 'message.part.updated') { + const info = payload.properties?.info; + const part = payload.properties?.part; + const sessionId = info?.sessionID ?? part?.sessionID ?? payload.properties?.sessionID; + const role = info?.role; + const partType = part?.type; + const reason = part?.reason; + if ( + typeof sessionId === 'string' && + sessionId.length > 0 && + role === 'assistant' && + partType === 'step-finish' && + reason === 'stop' + ) { + return { sessionId, phase: 'cooldown' }; + } + } + if (payload.type === 'session.idle') { const sessionId = payload.properties?.sessionID; if (typeof sessionId === 'string' && sessionId.length > 0) {