diff --git a/apps/desktop/src/lib/realtime.ts b/apps/desktop/src/lib/realtime.ts index 7ee1adc7..87e5eba4 100644 --- a/apps/desktop/src/lib/realtime.ts +++ b/apps/desktop/src/lib/realtime.ts @@ -1,5 +1,6 @@ import type { InvalidateQueryFilters, QueryClient } from "@tanstack/react-query"; -import { browserView, type AgentQuestion, type AutomationEvent, type BrowserProfile, type ClientEvent, type ConversationWithMessages, type DeviceGroup, type EntityName, type Message, type RemoteRunner, type ServerEvent, type Task, type TaskEvent, type Vm } from "@godmode/shared"; +import { browserView, type AgentQuestion, type AutomationEvent, type BrowserProfile, type ClientEvent, type ConversationWithMessages, type DeviceGroup, type EntityName, type Message, type RemoteRunner, type Run, type ServerEvent, type Task, type TaskEvent, type Vm } from "@godmode/shared"; +import { api } from "./api"; import { wsUrl } from "./core"; import { useLive } from "@/stores/live"; import { setChatMove, useChatMoves } from "@/stores/chatMoves"; @@ -13,6 +14,8 @@ let socket: WebSocket | null = null; let retry = 0; let stopped = false; let pingTimer: ReturnType | null = null; +let staleTimer: ReturnType | null = null; +let client: QueryClient | null = null; const pendingSends: ClientEvent[] = []; /** Subscribe to raw server events (returns unsubscribe). */ @@ -102,15 +105,53 @@ const ENTITY_KEYS: Record = { export function startRealtime(queryClient: QueryClient) { stopped = false; + client = queryClient; void connect(queryClient); + staleTimer ??= setInterval(() => void dropEndedRuns(), STALE_CHECK_MS); return () => { stopped = true; socket?.close(); socket = null; if (pingTimer) clearInterval(pingTimer); + if (staleTimer) clearInterval(staleTimer); + staleTimer = null; }; } +const ENDED: ReadonlySet = new Set(["succeeded", "failed", "cancelled"]); +const STALE_CHECK_MS = 60_000; +const STALE_AFTER_MS = 2 * 60_000; + +/** + * Asks for a run this window shows as working: one that ended without this window hearing of it (its end was told + * while its chat changed computers) goes. True when it was over. + */ +export async function settleRun(runId: string): Promise { + const run = await api.runs.get(runId).catch(() => null); + if (!run || !ENDED.has(run.status) || !useLive.getState().runs[runId]) return false; + if (client) handle(client, { type: "run.finished", run }); + else useLive.getState().runFinished(run); + return true; +} + +let sweeping = false; + +/** Runs that look busy for a while but aren't among the working ones are asked for one by one. */ +async function dropEndedRuns() { + if (sweeping || !useLive.getState().connected) return; + const old = Object.values(useLive.getState().runs).filter((r) => Date.now() - r.startedAt > STALE_AFTER_MS); + if (!old.length) return; + sweeping = true; + try { + const working = new Set((await api.runs.list({ status: "queued,running,paused", limit: 500 })).map((r) => r.id)); + for (const r of old) if (!working.has(r.runId)) await settleRun(r.runId); + } catch { + /* next round */ + } finally { + sweeping = false; + } +} + async function connect(queryClient: QueryClient) { if (stopped) return; let url: string; diff --git a/apps/desktop/src/pages/chat/chat-conversation.tsx b/apps/desktop/src/pages/chat/chat-conversation.tsx index 94f9a882..01204548 100644 --- a/apps/desktop/src/pages/chat/chat-conversation.tsx +++ b/apps/desktop/src/pages/chat/chat-conversation.tsx @@ -51,7 +51,7 @@ import { qk } from "@/lib/queryKeys"; import { useAllAgents, useBootstrap, useConversation, useRunners, useWorkspaces } from "@/lib/hooks"; import { ProjectChip } from "@/components/projects/project-chip"; import { effectiveProject, useProjectIndex } from "@/components/projects/project-utils"; -import { onServerEvent, viewConversation } from "@/lib/realtime"; +import { onServerEvent, settleRun, viewConversation } from "@/lib/realtime"; import { speak, useVoicePrefs, useVoiceSession } from "@/lib/voice"; import { useConversationLiveRun, type LiveRun } from "@/stores/live"; import { useUi } from "@/stores/ui"; @@ -426,8 +426,11 @@ function ConversationView({ conversationId }: { conversationId: string }) { }); const cancel = useMutation({ - mutationFn: (runId: string) => api.runs.cancel(runId), - onSuccess: () => toast("Stopping the agent…"), + mutationFn: async (runId: string) => { + await api.runs.cancel(runId); + return settleRun(runId); + }, + onSuccess: (ended) => toast(ended ? "That run had already ended" : "Stopping the agent…"), onError: (err) => toast.error("Couldn't stop the run", { description: errorMessage(err) }), }); diff --git a/apps/mobile/src/lib/api.ts b/apps/mobile/src/lib/api.ts index 27e6f68f..60c63876 100644 --- a/apps/mobile/src/lib/api.ts +++ b/apps/mobile/src/lib/api.ts @@ -65,6 +65,7 @@ import type { } from "@godmode/shared"; import { CloudErrorCode, LICENSE_REQUIRED, isPhoneUrlAllowed } from "@godmode/shared"; import type { PendingFile, UploadFile } from "./attachments"; +import { useLive } from "./live"; import { withPending } from "./pending-queue"; import { accepts, baseUrl, candidateOrder, computerOf, removedBy, type Candidate } from "./peers"; import { useSession, type Connection } from "./session"; @@ -629,7 +630,13 @@ export const api = { get: (id: string) => get(`/api/runs/${id}`), /** The run's raw stream-json log. */ log: (id: string) => get(`/api/runs/${id}/log`), - cancel: (id: string) => post<{ ok: true }>(`/api/runs/${id}/cancel`), + /** A run that had ended without the phone hearing of it (its chat changed computers) stops showing at once. */ + cancel: async (id: string) => { + const res = await post<{ ok: true }>(`/api/runs/${id}/cancel`); + const run = await get(`/api/runs/${id}`).catch(() => null); + if (run && (run.status === "succeeded" || run.status === "failed" || run.status === "cancelled")) useLive.getState().runFinished(run); + return res; + }, }, routines: {