diff --git a/web/openclaw_api.js b/web/openclaw_api.js index 86c2cbc..623e832 100644 --- a/web/openclaw_api.js +++ b/web/openclaw_api.js @@ -250,6 +250,10 @@ export class OpenClawAPI { return { ...res, data: historyItem }; } + async getPromptQueue() { + return this.fetch("/queue"); + } + // R25: Trace timeline (optional) async getTrace(promptId) { return this.fetch(`${this._path("/trace")}/${encodeURIComponent(promptId)}`); diff --git a/web/openclaw_queue_monitor.js b/web/openclaw_queue_monitor.js index 3d7bd12..9bddf9f 100644 --- a/web/openclaw_queue_monitor.js +++ b/web/openclaw_queue_monitor.js @@ -1,5 +1,25 @@ import { openclawApi } from "./openclaw_api.js"; +function extractQueuePromptId(entry) { + if (!entry) return ""; + if (typeof entry === "string") return entry; + if (Array.isArray(entry)) return String(entry[1] || ""); + if (typeof entry === "object") { + return String(entry.prompt_id || entry.promptId || entry.id || entry.job_id || ""); + } + return ""; +} + +function extractActiveQueuePromptIds(queueData = {}) { + const entries = [ + ...(Array.isArray(queueData.queue_running) ? queueData.queue_running : []), + ...(Array.isArray(queueData.queue_pending) ? queueData.queue_pending : []), + ...(Array.isArray(queueData.Running) ? queueData.Running : []), + ...(Array.isArray(queueData.Pending) ? queueData.Pending : []), + ]; + return new Set(entries.map(extractQueuePromptId).filter(Boolean)); +} + /** * F48/F49: Queue Lifecycle Monitor. * Consumes R71 events (SSE) with polling fallback to show deduplicated status banners. @@ -22,6 +42,7 @@ export class QueueMonitor { this.disconnectFailures = 0; this.hasObservedHealthyBackend = false; this.disconnectAlertActive = false; + this.activePromptIds = new Set(); } start() { @@ -55,10 +76,12 @@ export class QueueMonitor { } const type = data.event_type; - const pid = data.prompt_id ? data.prompt_id.slice(0, 8) : "???"; + const promptId = data.prompt_id ? String(data.prompt_id) : ""; + const pid = promptId ? promptId.slice(0, 8) : "???"; switch (type) { case "queued": + if (promptId) this.activePromptIds.add(promptId); this.showBanner({ severity: "info", message: `\u23F3 Job ${pid} queued`, @@ -68,6 +91,7 @@ export class QueueMonitor { }); break; case "running": + if (promptId) this.activePromptIds.add(promptId); this.showBanner({ severity: "info", message: `\u25B6 Job ${pid} running...`, @@ -77,6 +101,7 @@ export class QueueMonitor { }); break; case "failed": + if (promptId) this.activePromptIds.delete(promptId); this.showBanner({ severity: "error", message: `\u274C Job ${pid} failed`, @@ -92,6 +117,7 @@ export class QueueMonitor { }); break; case "completed": + if (promptId) this.activePromptIds.delete(promptId); break; } } @@ -105,8 +131,10 @@ export class QueueMonitor { try { const res = await this.api.getHealth(); if (res.ok && res.data) { + const wasDisconnected = !this.isConnected; this._markHealthy(); - if (!this.isConnected) { + if (wasDisconnected) { + await this._reconcileActiveJobsAfterReconnect(); this.isConnected = true; this.showBanner({ severity: "success", @@ -144,6 +172,33 @@ export class QueueMonitor { } } + async _reconcileActiveJobsAfterReconnect() { + if (!this.activePromptIds.size || typeof this.api.getPromptQueue !== "function") { + return; + } + + const res = await this.api.getPromptQueue(); + if (!res?.ok) { + return; + } + + const activeQueueIds = extractActiveQueuePromptIds(res.data); + const staleIds = [...this.activePromptIds].filter((promptId) => !activeQueueIds.has(promptId)); + staleIds.forEach((promptId) => this.activePromptIds.delete(promptId)); + if (!staleIds.length) { + return; + } + + const first = staleIds[0].slice(0, 8) || "unknown"; + this.showBanner({ + severity: "info", + message: `Job ${first} no longer active after reconnect`, + id: `job_reconnect_cleared_${first}`, + ttl_ms: 3000, + source: "queue-monitor", + }); + } + _markHealthy() { this.hasObservedHealthyBackend = true; this.disconnectFailures = 0; diff --git a/web/tests/unit/openclaw_queue_monitor.test.js b/web/tests/unit/openclaw_queue_monitor.test.js index af7d5f8..131d80d 100644 --- a/web/tests/unit/openclaw_queue_monitor.test.js +++ b/web/tests/unit/openclaw_queue_monitor.test.js @@ -160,4 +160,72 @@ describe("QueueMonitor", () => { }) ); }); + + it("clears stale active prompt ids after reconnect when the queue snapshot no longer lists them", async () => { + const ui = { showBanner: vi.fn() }; + const closedStream = { readyState: 2, close: vi.fn() }; + const monitor = new QueueMonitor(ui, { + api: { + getHealth: vi.fn().mockResolvedValue({ + ok: true, + data: { stats: { observability: { total_dropped: 0 } } }, + }), + getPromptQueue: vi.fn().mockResolvedValue({ + ok: true, + data: { + queue_running: [], + queue_pending: [[1, "still-active"]], + }, + }), + subscribeEvents: vi.fn(() => ({ readyState: 1, close: vi.fn() })), + }, + setIntervalRef: vi.fn(), + }); + + monitor.handleEvent({ event_type: "running", prompt_id: "stale-job-1" }); + monitor.handleEvent({ event_type: "queued", prompt_id: "still-active" }); + monitor.isConnected = false; + monitor.es = closedStream; + + await monitor.checkHealth(); + + expect(monitor.activePromptIds.has("stale-job-1")).toBe(false); + expect(monitor.activePromptIds.has("still-active")).toBe(true); + expect(ui.showBanner).toHaveBeenCalledWith( + expect.objectContaining({ + id: "job_reconnect_cleared_stale-jo", + severity: "info", + }) + ); + }); + + it("preserves active prompt ids when reconnect queue refresh fails", async () => { + const ui = { showBanner: vi.fn() }; + const monitor = new QueueMonitor(ui, { + api: { + getHealth: vi.fn().mockResolvedValue({ + ok: true, + data: { stats: { observability: { total_dropped: 0 } } }, + }), + getPromptQueue: vi.fn().mockResolvedValue({ + ok: false, + error: "queue_unavailable", + }), + subscribeEvents: vi.fn(() => ({ readyState: 1, close: vi.fn() })), + }, + setIntervalRef: vi.fn(), + }); + + monitor.handleEvent({ event_type: "running", prompt_id: "active-job" }); + monitor.isConnected = false; + + await monitor.checkHealth(); + + expect(monitor.activePromptIds.has("active-job")).toBe(true); + expect(ui.showBanner).not.toHaveBeenCalledWith( + expect.objectContaining({ + id: expect.stringMatching(/^job_reconnect_cleared_/), + }) + ); + }); });