fix(queue): reconcile active prompts after reconnect

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)}`);

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;

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_/),
+            })
+        );
+    });
 });