From 84c7941f1e151ad45f6d20b8cd8204f4e78f7bc9 Mon Sep 17 00:00:00 2001 From: Peter Steinberger Date: Wed, 15 Jul 2026 03:32:15 -0700 Subject: [PATCH] refactor: simplify recent async ownership fixes (#108164) * fix(device-pair): preserve concurrent notify state Co-authored-by: Alix-007 * refactor(ui): simplify logbook refresh ownership Co-authored-by: Alix-007 * fix(qa-lab): bound multipass retry window Co-authored-by: Alix-007 * chore: keep release notes in PR body Co-authored-by: Alix-007 * fix(ui): keep logbook controller state private Co-authored-by: Alix-007 * fix(ui): retire logbook client epochs Co-authored-by: Alix-007 * fix(ui): retire inactive logbook ownership Co-authored-by: Alix-007 * fix(device-pair): require atomic notify state Co-authored-by: Alix-007 --------- Co-authored-by: Alix-007 --- docs/plugins/sdk-runtime.md | 5 +- extensions/device-pair/notify-state.ts | 2 + extensions/device-pair/notify.test.ts | 225 +++++++++++++++- extensions/device-pair/notify.ts | 241 ++++++++--------- .../qa-lab/src/multipass.runtime.test.ts | 4 +- extensions/qa-lab/src/multipass.runtime.ts | 4 +- src/plugin-state/plugin-state-store.sqlite.ts | 34 +++ src/plugin-state/plugin-state-store.test.ts | 18 ++ src/plugin-state/plugin-state-store.ts | 21 ++ src/plugin-state/plugin-state-store.types.ts | 4 + .../pages/plugin/logbook-controller.test.ts | 231 +++++++++------- ui/src/pages/plugin/logbook-controller.ts | 246 ++++++++++-------- ui/src/pages/plugin/logbook-types.ts | 18 -- ui/src/pages/plugin/logbook-view.ts | 14 +- 14 files changed, 699 insertions(+), 368 deletions(-) diff --git a/docs/plugins/sdk-runtime.md b/docs/plugins/sdk-runtime.md index 1e48fd08942..81b05288ab1 100644 --- a/docs/plugins/sdk-runtime.md +++ b/docs/plugins/sdk-runtime.md @@ -703,13 +703,14 @@ two-party event loops that do not go through the shared inbound reply runner. await store.register("key-1", { value: "hello" }); const claimed = await store.registerIfAbsent("dedupe-key", { value: "first" }); const value = await store.lookup("key-1"); + await store.deleteIf?.("key-1", (current) => current.value === "hello"); await store.consume("key-1"); await store.clear(); ``` - Keyed stores survive restarts and are isolated by the runtime-bound plugin id. Use `registerIfAbsent(...)` for atomic dedupe claims: it returns `true` when the key was missing or expired and registered, or `false` when a live value already exists without overwriting its value, creation time, or TTL. Limits: `maxEntries` per namespace, 50,000 live rows per plugin, JSON values under 64KB, and optional TTL expiry. By default, a write at either row limit sheds the oldest live rows from the namespace being written; sibling namespaces are not evicted for that write, and the write still fails if the namespace cannot free enough rows. Set `overflowPolicy: "reject-new"` for durable ownership records that must never be evicted: new keys fail at either limit, while existing keys remain updateable. + Keyed stores survive restarts and are isolated by the runtime-bound plugin id. Use `registerIfAbsent(...)` for atomic dedupe claims: it returns `true` when the key was missing or expired and registered, or `false` when a live value already exists without overwriting its value, creation time, or TTL. Use `deleteIf(...)` when cleanup must remove only the value previously observed; its synchronous predicate and deletion run in one SQLite transaction. Limits: `maxEntries` per namespace, 50,000 live rows per plugin, JSON values under 64KB, and optional TTL expiry. By default, a write at either row limit sheds the oldest live rows from the namespace being written; sibling namespaces are not evicted for that write, and the write still fails if the namespace cannot free enough rows. Set `overflowPolicy: "reject-new"` for durable ownership records that must never be evicted: new keys fail at either limit, while existing keys remain updateable. - `openSyncKeyedStore(...)` returns the same store shape with synchronous methods (`register`, `registerIfAbsent`, `lookup`, `consume`, `clear` all return values directly instead of promises) for callers that cannot await. + `openSyncKeyedStore(...)` returns the same store shape with synchronous methods (`register`, `registerIfAbsent`, `deleteIf`, `lookup`, `consume`, `clear` all return values directly instead of promises) for callers that cannot await. `openChannelIngressQueue(...)` opens a persisted ingress queue scoped to the calling plugin, for buffering inbound events that need at-least-once processing across restarts. When stale-claim recovery uses `shouldRecover`, also provide `shouldRecoverCorrupt` if corrupt claimed payloads should be quarantined: its payload-independent claim identity lets the plugin preserve live owner and lane policy before the queue tombstones the row. diff --git a/extensions/device-pair/notify-state.ts b/extensions/device-pair/notify-state.ts index 0e0c9dc6e57..3c0b37bb2fb 100644 --- a/extensions/device-pair/notify-state.ts +++ b/extensions/device-pair/notify-state.ts @@ -15,6 +15,8 @@ export type NotifySubscription = { messageThreadId?: string | number; mode: "persistent" | "once"; addedAtMs: number; + /** Unique for new arms; absent only on subscriptions imported from legacy state. */ + armId?: string; }; export type NotifySeenRequest = { diff --git a/extensions/device-pair/notify.test.ts b/extensions/device-pair/notify.test.ts index ebc8d28a215..91bca82a0aa 100644 --- a/extensions/device-pair/notify.test.ts +++ b/extensions/device-pair/notify.test.ts @@ -3,7 +3,10 @@ import fs from "node:fs/promises"; import os from "node:os"; import path from "node:path"; import type { listDevicePairing as listDevicePairingFn } from "openclaw/plugin-sdk/device-bootstrap"; -import type { OpenKeyedStoreOptions } from "openclaw/plugin-sdk/plugin-state-runtime"; +import type { + OpenKeyedStoreOptions, + PluginStateKeyedStore, +} from "openclaw/plugin-sdk/plugin-state-runtime"; import { createPluginStateKeyedStoreForTests, resetPluginStateStoreForTests, @@ -67,12 +70,15 @@ describe("device-pair notify persistence", () => { }); } - function createApi(sendText?: ReturnType) { + function createApi( + sendText?: ReturnType, + openKeyedStore: (options: OpenKeyedStoreOptions) => PluginStateKeyedStore = openStore, + ) { return createTestPluginApi({ runtime: { state: { resolveStateDir: () => stateDir, - openKeyedStore: openStore, + openKeyedStore, }, channel: { outbound: { @@ -194,6 +200,219 @@ describe("device-pair notify persistence", () => { await service.stop?.({} as never); }); + it("preserves subscriber changes made while a notification is in flight", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const firstSend = createDeferred(); + const sendText = vi.fn(() => firstSend.promise); + const api = createApi(sendText); + await handleNotifyCommand({ + api, + ctx: { channel: "telegram", senderId: "old-chat" }, + action: "on", + }); + listDevicePairingMock.mockResolvedValueOnce({ pending: [], paired: [] }).mockResolvedValue({ + pending: [ + { + requestId: "request-1", + deviceId: "device-1", + publicKey: "public-key-1", + ts: 2_000, + }, + ], + paired: [], + }); + const service = createPairingNotifierService(api); + + await service.start({} as never); + await vi.advanceTimersByTimeAsync(10_000); + expect(sendText).toHaveBeenCalledTimes(1); + + await handleNotifyCommand({ + api, + ctx: { channel: "telegram", senderId: "old-chat" }, + action: "off", + }); + await handleNotifyCommand({ + api, + ctx: { channel: "telegram", senderId: "new-chat" }, + action: "on", + }); + firstSend.resolve({ channel: "telegram", to: "old-chat" }); + await vi.advanceTimersByTimeAsync(0); + + await expect(openSubscriberStore().entries()).resolves.toMatchObject([ + { + key: notifySubscriberStoreKey({ to: "new-chat" }), + value: { to: "new-chat", mode: "persistent" }, + }, + ]); + await service.stop?.({} as never); + }); + + it("preserves a one-shot subscription re-armed during its delivery", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const firstSend = createDeferred(); + const sendText = vi.fn(() => firstSend.promise); + const api = createApi(sendText); + await handleNotifyCommand({ + api, + ctx: { channel: "telegram", senderId: "chat-123" }, + action: "once", + }); + listDevicePairingMock.mockResolvedValueOnce({ pending: [], paired: [] }).mockResolvedValue({ + pending: [ + { + requestId: "request-1", + deviceId: "device-1", + publicKey: "public-key-1", + ts: 2_000, + }, + ], + paired: [], + }); + const service = createPairingNotifierService(api); + + await service.start({} as never); + await vi.advanceTimersByTimeAsync(10_000); + expect(sendText).toHaveBeenCalledTimes(1); + + await handleNotifyCommand({ + api, + ctx: { channel: "telegram", senderId: "chat-123" }, + action: "once", + }); + firstSend.resolve({ channel: "telegram", to: "chat-123" }); + await vi.advanceTimersByTimeAsync(0); + + await expect( + openSubscriberStore().lookup(notifySubscriberStoreKey({ to: "chat-123" })), + ).resolves.toMatchObject({ + to: "chat-123", + mode: "once", + addedAtMs: 11_000, + }); + await service.stop?.({} as never); + }); + + it("rejects missing conditional-delete support before one-shot delivery", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const subscriber: NotifySubscription = { + to: "chat-123", + mode: "once", + addedAtMs: 1_000, + armId: "arm-1", + }; + await openSubscriberStore().register(notifySubscriberStoreKey(subscriber), subscriber); + listDevicePairingMock.mockResolvedValue({ + pending: [ + { + requestId: "request-1", + deviceId: "device-1", + publicKey: "public-key-1", + ts: 2_000, + }, + ], + paired: [], + }); + const sendText = vi.fn(async () => ({ channel: "telegram", to: "chat-123" })); + const api = createApi(sendText, (options: OpenKeyedStoreOptions) => { + const { deleteIf: _deleteIf, ...store } = openStore(options); + return store; + }); + const service = createPairingNotifierService(api); + + await service.start({} as never); + + expect(sendText).not.toHaveBeenCalled(); + await expect( + openSubscriberStore().lookup(notifySubscriberStoreKey(subscriber)), + ).resolves.toEqual(subscriber); + await service.stop?.({} as never); + }); + + it("keeps the request boundary at the current millisecond when re-armed", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const sendText = vi.fn(async () => ({ channel: "telegram", to: "chat-123" })); + const api = createApi(sendText); + const command = { + api, + ctx: { channel: "telegram", senderId: "chat-123" }, + action: "once" as const, + }; + + await handleNotifyCommand(command); + const key = notifySubscriberStoreKey({ to: "chat-123" }); + const first = await openSubscriberStore().lookup(key); + await handleNotifyCommand(command); + const second = await openSubscriberStore().lookup(key); + + expect(first).toMatchObject({ addedAtMs: 1_000, armId: expect.any(String) }); + expect(second).toMatchObject({ addedAtMs: 1_000, armId: expect.any(String) }); + expect(second?.armId).not.toBe(first?.armId); + + listDevicePairingMock.mockResolvedValue({ + pending: [ + { + requestId: "request-same-ms", + deviceId: "device-1", + publicKey: "public-key-1", + ts: 1_000, + }, + ], + paired: [], + }); + const service = createPairingNotifierService(api); + await service.start({} as never); + + expect(sendText).toHaveBeenCalledWith( + expect.objectContaining({ text: expect.stringContaining("ID: request-same-ms") }), + ); + await service.stop?.({} as never); + }); + + it("delivers a one-shot subscription to only the first new request", async () => { + vi.useFakeTimers(); + vi.setSystemTime(1_000); + const sendText = vi.fn(async () => ({ channel: "telegram", to: "chat-123" })); + const api = createApi(sendText); + await handleNotifyCommand({ + api, + ctx: { channel: "telegram", senderId: "chat-123" }, + action: "once", + }); + listDevicePairingMock.mockResolvedValue({ + pending: [ + { + requestId: "request-1", + deviceId: "device-1", + publicKey: "public-key-1", + ts: 1_001, + }, + { + requestId: "request-2", + deviceId: "device-2", + publicKey: "public-key-2", + ts: 1_002, + }, + ], + paired: [], + }); + const service = createPairingNotifierService(api); + + await service.start({} as never); + + expect(sendText).toHaveBeenCalledTimes(1); + expect(sendText).toHaveBeenCalledWith( + expect.objectContaining({ text: expect.stringContaining("ID: request-1") }), + ); + await expect(openSubscriberStore().entries()).resolves.toStrictEqual([]); + await service.stop?.({} as never); + }); + it("matches persisted telegram thread ids across number and string roundtrips", async () => { const subscriber: NotifySubscription = { to: "chat-123", diff --git a/extensions/device-pair/notify.ts b/extensions/device-pair/notify.ts index 42846757c72..f7c7ea61b73 100644 --- a/extensions/device-pair/notify.ts +++ b/extensions/device-pair/notify.ts @@ -1,4 +1,5 @@ // Device Pair plugin module implements notify behavior. +import { randomUUID } from "node:crypto"; import type { OpenClawPluginService } from "openclaw/plugin-sdk/core"; import { listDevicePairing } from "openclaw/plugin-sdk/device-bootstrap"; import { formatErrorMessage } from "openclaw/plugin-sdk/error-runtime"; @@ -24,11 +25,6 @@ const NOTIFY_POLL_INTERVAL_MS = 10_000; // Keep one module-owned poll so the replacement service cannot race state or delivery. let notifyPollInFlight: Promise | null = null; -type NotifyStateFile = { - subscribers: NotifySubscription[]; - notifiedRequestIds: Record; -}; - type PendingPairingRequest = { requestId: string; deviceId: string; @@ -83,13 +79,21 @@ export function formatPendingRequests(pending: PendingPairingRequest[]): string return lines.join("\n"); } -function openNotifySubscriberStore( - api: OpenClawPluginApi, -): PluginStateKeyedStore { - return api.runtime.state.openKeyedStore({ +type NotifySubscriberStore = PluginStateKeyedStore & { + deleteIf: NonNullable["deleteIf"]>; +}; + +function openNotifySubscriberStore(api: OpenClawPluginApi): NotifySubscriberStore { + const store = api.runtime.state.openKeyedStore({ namespace: DEVICE_PAIR_NOTIFY_SUBSCRIBER_NAMESPACE, maxEntries: DEVICE_PAIR_NOTIFY_SUBSCRIBER_MAX_ENTRIES, }); + if (!store.deleteIf) { + throw new Error( + "device-pair notify requires a runtime with atomic plugin state conditional delete support", + ); + } + return store as NotifySubscriberStore; } function openNotifySeenRequestStore( @@ -102,63 +106,6 @@ function openNotifySeenRequestStore( }); } -async function readNotifyState(api: OpenClawPluginApi): Promise { - const subscriberStore = openNotifySubscriberStore(api); - const seenRequestStore = openNotifySeenRequestStore(api); - const [subscriberEntries, seenRequestEntries] = await Promise.all([ - subscriberStore.entries(), - seenRequestStore.entries(), - ]); - - const subscribers = subscriberEntries - .map((entry) => entry.value) - .toSorted((a, b) => a.addedAtMs - b.addedAtMs); - const notifiedRequestIds: Record = {}; - for (const entry of seenRequestEntries) { - const requestId = normalizeOptionalString(entry.value.requestId); - const notifiedAtMs = entry.value.notifiedAtMs; - if (!requestId || !Number.isFinite(notifiedAtMs) || notifiedAtMs <= 0) { - continue; - } - notifiedRequestIds[requestId] = Math.trunc(notifiedAtMs); - } - - return { subscribers, notifiedRequestIds }; -} - -async function writeNotifyState(api: OpenClawPluginApi, state: NotifyStateFile): Promise { - const subscriberStore = openNotifySubscriberStore(api); - const nextSubscribers = new Map( - state.subscribers.map((subscriber) => [notifySubscriberStoreKey(subscriber), subscriber]), - ); - for (const entry of await subscriberStore.entries()) { - if (!nextSubscribers.has(entry.key)) { - await subscriberStore.delete(entry.key); - } - } - for (const [key, subscriber] of nextSubscribers) { - await subscriberStore.register(key, subscriber); - } - - const seenRequestStore = openNotifySeenRequestStore(api); - const nextSeenRequests = new Map( - Object.entries(state.notifiedRequestIds).map(([requestId, notifiedAtMs]) => [ - notifyRequestStoreKey(requestId), - { requestId, notifiedAtMs }, - ]), - ); - for (const entry of await seenRequestStore.entries()) { - if (!nextSeenRequests.has(entry.key)) { - await seenRequestStore.delete(entry.key); - } - } - for (const [key, value] of nextSeenRequests) { - await seenRequestStore.register(key, value, { - ttlMs: DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS, - }); - } -} - type NotifyTarget = { to: string; accountId?: string; @@ -187,30 +134,51 @@ function resolveNotifyTarget(ctx: { }; } -function upsertNotifySubscriber( - subscribers: NotifySubscription[], +function nextNotifySubscription( target: NotifyTarget, mode: NotifySubscription["mode"], -): boolean { - const key = notifySubscriberKey(target); - const index = subscribers.findIndex((entry) => notifySubscriberKey(entry) === key); - const next: NotifySubscription = { +): NotifySubscription { + return { ...target, mode, addedAtMs: Date.now(), + armId: randomUUID(), }; - if (index === -1) { - subscribers.push(next); - return true; - } - const existing = subscribers[index]; - if (existing?.mode === mode) { +} + +async function registerNotifySubscriber(params: { + api: OpenClawPluginApi; + target: NotifyTarget; + mode: NotifySubscription["mode"]; + refresh: boolean; +}): Promise { + const store = openNotifySubscriberStore(params.api); + const key = notifySubscriberStoreKey(params.target); + const current = await store.lookup(key); + if (!params.refresh && current?.mode === params.mode) { return false; } - subscribers[index] = next; + await store.register(key, nextNotifySubscription(params.target, params.mode)); return true; } +function isSameNotifySubscription( + current: NotifySubscription, + expected: NotifySubscription, +): boolean { + if (expected.armId) { + return current.armId === expected.armId; + } + // Doctor-imported legacy subscriptions have no arm id. Their original + // fields remain the exact generation until a new arm replaces the row. + return ( + current.armId === undefined && + current.mode === expected.mode && + current.addedAtMs === expected.addedAtMs && + notifySubscriberKey(current) === notifySubscriberKey(expected) + ); +} + function buildPairingRequestNotificationText(request: PendingPairingRequest): string { const label = normalizeOptionalString(request.displayName) || request.deviceId; const platform = normalizeOptionalString(request.platform); @@ -289,30 +257,49 @@ async function notifySubscriber(params: { } async function notifyPendingPairingRequests(params: { api: OpenClawPluginApi }): Promise { - const state = await readNotifyState(params.api); - const pairing = await listDevicePairing(); + const subscriberStore = openNotifySubscriberStore(params.api); + const seenRequestStore = openNotifySeenRequestStore(params.api); + const [subscriberEntries, seenRequestEntries, pairing] = await Promise.all([ + subscriberStore.entries(), + seenRequestStore.entries(), + listDevicePairing(), + ]); + const subscribers = subscriberEntries.toSorted((a, b) => a.value.addedAtMs - b.value.addedAtMs); const pending: PendingPairingRequest[] = pairing.pending; const now = Date.now(); const pendingIds = new Set(pending.map((entry) => entry.requestId)); - let changed = false; + const notifiedRequestIds = new Set(); - for (const [requestId, ts] of Object.entries(state.notifiedRequestIds)) { - if (!pendingIds.has(requestId) || now - ts > DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS) { - delete state.notifiedRequestIds[requestId]; - changed = true; + for (const entry of seenRequestEntries) { + const requestId = normalizeOptionalString(entry.value.requestId); + const notifiedAtMs = entry.value.notifiedAtMs; + if ( + !requestId || + !Number.isFinite(notifiedAtMs) || + notifiedAtMs <= 0 || + !pendingIds.has(requestId) || + now - notifiedAtMs > DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS + ) { + await seenRequestStore.delete(entry.key); + continue; } + notifiedRequestIds.add(requestId); } - if (state.subscribers.length > 0) { - const oneShotDelivered = new Set(); + if (subscribers.length > 0) { + const deliveredOneShots = new Set(); for (const request of pending) { - if (state.notifiedRequestIds[request.requestId]) { + if (notifiedRequestIds.has(request.requestId)) { continue; } const text = buildPairingRequestNotificationText(request); let delivered = false; - for (const subscriber of state.subscribers) { + for (const entry of subscribers) { + const subscriber = entry.value; + if (subscriber.mode === "once" && deliveredOneShots.has(entry.key)) { + continue; + } if (!shouldNotifySubscriberForRequest(subscriber, request)) { continue; } @@ -323,28 +310,24 @@ async function notifyPendingPairingRequests(params: { api: OpenClawPluginApi }): }); delivered = delivered || sent; if (sent && subscriber.mode === "once") { - oneShotDelivered.add(notifySubscriberKey(subscriber)); + deliveredOneShots.add(entry.key); + // Delivery is fallible and uncancellable. Delete only the exact arm + // that was sent so an overlapping re-arm remains subscribed. + await subscriberStore.deleteIf(entry.key, (current) => + isSameNotifySubscription(current, subscriber), + ); } } if (delivered) { - state.notifiedRequestIds[request.requestId] = now; - changed = true; + await seenRequestStore.register( + notifyRequestStoreKey(request.requestId), + { requestId: request.requestId, notifiedAtMs: now }, + { ttlMs: DEVICE_PAIR_NOTIFY_MAX_SEEN_AGE_MS }, + ); + notifiedRequestIds.add(request.requestId); } } - if (oneShotDelivered.size > 0) { - const initialCount = state.subscribers.length; - state.subscribers = state.subscribers.filter( - (subscriber) => !oneShotDelivered.has(notifySubscriberKey(subscriber)), - ); - if (state.subscribers.length !== initialCount) { - changed = true; - } - } - } - - if (changed) { - await writeNotifyState(params.api, state); } } @@ -379,16 +362,12 @@ export async function armPairNotifyOnce(params: { return false; } - const state = await readNotifyState(params.api); - let changed = false; - - if (upsertNotifySubscriber(state.subscribers, target, "once")) { - changed = true; - } - - if (changed) { - await writeNotifyState(params.api, state); - } + await registerNotifySubscriber({ + api: params.api, + target, + mode: "once", + refresh: true, + }); return true; } @@ -413,14 +392,16 @@ export async function handleNotifyCommand(params: { return { text: "Could not resolve Telegram target for this chat." }; } - const state = await readNotifyState(params.api); - const targetKey = notifySubscriberKey(target); - const current = state.subscribers.find((entry) => notifySubscriberKey(entry) === targetKey); + const subscriberStore = openNotifySubscriberStore(params.api); + const targetStoreKey = notifySubscriberStoreKey(target); if (params.action === "on" || params.action === "enable") { - if (upsertNotifySubscriber(state.subscribers, target, "persistent")) { - await writeNotifyState(params.api, state); - } + await registerNotifySubscriber({ + api: params.api, + target, + mode: "persistent", + refresh: false, + }); return { text: "✅ Pair request notifications enabled for this Telegram chat.\n" + @@ -429,13 +410,7 @@ export async function handleNotifyCommand(params: { } if (params.action === "off" || params.action === "disable") { - const currentIndex = state.subscribers.findIndex( - (entry) => notifySubscriberKey(entry) === targetKey, - ); - if (currentIndex !== -1) { - state.subscribers.splice(currentIndex, 1); - await writeNotifyState(params.api, state); - } + await subscriberStore.delete(targetStoreKey); return { text: "✅ Pair request notifications disabled for this Telegram chat." }; } @@ -452,14 +427,18 @@ export async function handleNotifyCommand(params: { } if (params.action === "status" || params.action === "") { - const pending = await listDevicePairing(); + const [current, subscribers, pending] = await Promise.all([ + subscriberStore.lookup(targetStoreKey), + subscriberStore.entries(), + listDevicePairing(), + ]); const enabled = Boolean(current); const mode = current?.mode ?? "off"; return { text: [ `Pair request notifications: ${enabled ? "enabled" : "disabled"} for this chat.`, `Mode: ${mode}`, - `Subscribers: ${state.subscribers.length}`, + `Subscribers: ${subscribers.length}`, `Pending requests: ${pending.pending.length}`, "", "Use /pair notify on|off|once", diff --git a/extensions/qa-lab/src/multipass.runtime.test.ts b/extensions/qa-lab/src/multipass.runtime.test.ts index fdc79401eab..d6caad0fdd3 100644 --- a/extensions/qa-lab/src/multipass.runtime.test.ts +++ b/extensions/qa-lab/src/multipass.runtime.test.ts @@ -142,10 +142,10 @@ describe("qa multipass runtime", () => { expect(script).toContain("pnpm build"); expect(script).toContain("corepack prepare 'pnpm@"); expect(script).toContain( - 'curl -fsSL --connect-timeout 10 --max-time 120 --retry 2 --retry-delay 2 "${base_url}/SHASUMS256.txt" -o "${node_tmp_dir}/SHASUMS256.txt"', + 'curl -fsSL --connect-timeout 10 --max-time 120 --retry 2 --retry-delay 2 --retry-max-time 120 "${base_url}/SHASUMS256.txt" -o "${node_tmp_dir}/SHASUMS256.txt"', ); expect(script).toContain( - 'curl -fsSL --connect-timeout 10 --max-time 120 --retry 2 --retry-delay 2 "${base_url}/${tarball_name}" -o "${node_tmp_dir}/${tarball_name}"', + 'curl -fsSL --connect-timeout 10 --max-time 120 --retry 2 --retry-delay 2 --retry-max-time 120 "${base_url}/${tarball_name}" -o "${node_tmp_dir}/${tarball_name}"', ); expect(script).toContain("'pnpm' 'openclaw' 'qa' 'suite' '--transport' 'qa-channel'"); expect(script).toContain("'--provider-mode' 'live-frontier'"); diff --git a/extensions/qa-lab/src/multipass.runtime.ts b/extensions/qa-lab/src/multipass.runtime.ts index 7f562c7fd20..c7e74d155d7 100644 --- a/extensions/qa-lab/src/multipass.runtime.ts +++ b/extensions/qa-lab/src/multipass.runtime.ts @@ -410,10 +410,10 @@ function renderQaMultipassGuestScript( ' node_tmp_dir="$(mktemp -d)"', " trap 'rm -rf \"${node_tmp_dir}\"' RETURN", ' base_url="https://nodejs.org/dist/latest-v22.x"', - ' curl -fsSL --connect-timeout 10 --max-time 120 --retry 2 --retry-delay 2 "${base_url}/SHASUMS256.txt" -o "${node_tmp_dir}/SHASUMS256.txt" >>"$BOOTSTRAP_LOG" 2>&1', + ' curl -fsSL --connect-timeout 10 --max-time 120 --retry 2 --retry-delay 2 --retry-max-time 120 "${base_url}/SHASUMS256.txt" -o "${node_tmp_dir}/SHASUMS256.txt" >>"$BOOTSTRAP_LOG" 2>&1', ' tarball_name="$(awk \'/linux-\'"${node_arch}"\'\\.tar\\.xz$/ { print $2; exit }\' "${node_tmp_dir}/SHASUMS256.txt")"', ' [ -n "${tarball_name}" ] || { echo "unable to resolve node tarball for ${node_arch}" >&2; return 1; }', - ' curl -fsSL --connect-timeout 10 --max-time 120 --retry 2 --retry-delay 2 "${base_url}/${tarball_name}" -o "${node_tmp_dir}/${tarball_name}" >>"$BOOTSTRAP_LOG" 2>&1', + ' curl -fsSL --connect-timeout 10 --max-time 120 --retry 2 --retry-delay 2 --retry-max-time 120 "${base_url}/${tarball_name}" -o "${node_tmp_dir}/${tarball_name}" >>"$BOOTSTRAP_LOG" 2>&1', ' (cd "${node_tmp_dir}" && grep " ${tarball_name}$" SHASUMS256.txt | sha256sum -c -) >>"$BOOTSTRAP_LOG" 2>&1', ' extract_dir="${tarball_name%.tar.xz}"', ' sudo mkdir -p /usr/local/lib/nodejs >>"$BOOTSTRAP_LOG" 2>&1', diff --git a/src/plugin-state/plugin-state-store.sqlite.ts b/src/plugin-state/plugin-state-store.sqlite.ts index 15006cb8a19..4f6fa4e51be 100644 --- a/src/plugin-state/plugin-state-store.sqlite.ts +++ b/src/plugin-state/plugin-state-store.sqlite.ts @@ -792,6 +792,40 @@ export function pluginStateDelete(params: { } } +export function pluginStateDeleteIf(params: { + pluginId: string; + namespace: string; + key: string; + predicate: (current: unknown) => boolean; + env?: NodeJS.ProcessEnv; +}): boolean { + try { + return runWriteTransaction( + "delete", + ({ db }) => { + const row = selectPluginStateEntry(db, { + pluginId: params.pluginId, + namespace: params.namespace, + key: params.key, + now: Date.now(), + }); + if (!row || !params.predicate(parseStoredJson(row.value_json, "delete"))) { + return false; + } + return deletePluginStateEntry(db, params) > 0; + }, + envOptions(params.env), + ); + } catch (error) { + throw wrapPluginStateError( + error, + "delete", + "PLUGIN_STATE_WRITE_FAILED", + "Failed to conditionally delete plugin state entry.", + ); + } +} + export function pluginStateEntries(params: { pluginId: string; namespace: string; diff --git a/src/plugin-state/plugin-state-store.test.ts b/src/plugin-state/plugin-state-store.test.ts index a105e523ad7..3fb3bc44ae9 100644 --- a/src/plugin-state/plugin-state-store.test.ts +++ b/src/plugin-state/plugin-state-store.test.ts @@ -253,6 +253,24 @@ describe("plugin state keyed store", () => { }); }); + it("deletes an entry only when the current value matches", async () => { + await withPluginStateTestState(async () => { + const store = createPluginStateKeyedStore<{ version: number }>("device-pair", { + namespace: "notify-subscribers", + maxEntries: 10, + }); + await store.register("chat", { version: 1 }); + if (!store.deleteIf) { + throw new Error("plugin state conditional delete unavailable"); + } + + await expect(store.deleteIf("chat", (current) => current.version === 2)).resolves.toBe(false); + await expect(store.lookup("chat")).resolves.toEqual({ version: 1 }); + await expect(store.deleteIf("chat", (current) => current.version === 1)).resolves.toBe(true); + await expect(store.lookup("chat")).resolves.toBeUndefined(); + }); + }); + it("registerIfAbsent keeps plugin and namespace claims isolated", async () => { await withPluginStateTestState(async () => { const discordA = createPluginStateKeyedStore<{ owner: string }>("discord", { diff --git a/src/plugin-state/plugin-state-store.ts b/src/plugin-state/plugin-state-store.ts index ad02ce98d07..a767c0f1a00 100644 --- a/src/plugin-state/plugin-state-store.ts +++ b/src/plugin-state/plugin-state-store.ts @@ -7,6 +7,7 @@ import { pluginStateClear, pluginStateConsume, pluginStateDelete, + pluginStateDeleteIf, pluginStateEntries, pluginStateLookup, pluginStateRegister, @@ -320,6 +321,16 @@ function createKeyedStoreForPluginId( ...(env ? { env } : {}), }); }, + async deleteIf(key, predicate) { + const normalizedKey = validateKey(key, "delete"); + return pluginStateDeleteIf({ + pluginId, + namespace, + key: normalizedKey, + predicate: (current) => predicate(current as T), + ...(env ? { env } : {}), + }); + }, async lookup(key) { const normalizedKey = validateKey(key, "lookup"); return pluginStateLookup({ @@ -420,6 +431,16 @@ function createSyncKeyedStoreForPluginId( ...(env ? { env } : {}), }); }, + deleteIf(key, predicate) { + const normalizedKey = validateKey(key, "delete"); + return pluginStateDeleteIf({ + pluginId, + namespace, + key: normalizedKey, + predicate: (current) => predicate(current as T), + ...(env ? { env } : {}), + }); + }, lookup(key) { const normalizedKey = validateKey(key, "lookup"); return pluginStateLookup({ diff --git a/src/plugin-state/plugin-state-store.types.ts b/src/plugin-state/plugin-state-store.types.ts index abcac4be874..3f8ba31b99c 100644 --- a/src/plugin-state/plugin-state-store.types.ts +++ b/src/plugin-state/plugin-state-store.types.ts @@ -16,6 +16,8 @@ export type PluginStateKeyedStore = { updateValue: (current: T | undefined) => T | undefined, opts?: { ttlMs?: number }, ) => Promise; + /** Atomically deletes an existing entry when its current value matches. */ + deleteIf?: (key: string, predicate: (current: T) => boolean) => Promise; lookup(key: string): Promise; consume(key: string): Promise; delete(key: string): Promise; @@ -32,6 +34,8 @@ export type PluginStateSyncKeyedStore = { updateValue: (current: T | undefined) => T | undefined, opts?: { ttlMs?: number }, ) => boolean; + /** Atomically deletes an existing entry when its current value matches. */ + deleteIf?: (key: string, predicate: (current: T) => boolean) => boolean; lookup(key: string): T | undefined; consume(key: string): T | undefined; delete(key: string): boolean; diff --git a/ui/src/pages/plugin/logbook-controller.test.ts b/ui/src/pages/plugin/logbook-controller.test.ts index 8161c8e6e35..b9fc6df49dc 100644 --- a/ui/src/pages/plugin/logbook-controller.test.ts +++ b/ui/src/pages/plugin/logbook-controller.test.ts @@ -7,6 +7,7 @@ import { loadLogbook, loadLogbookStandup, runLogbookAnalysisNow, + setLogbookCapturePaused, stopLogbookPolling, } from "./logbook-controller.ts"; import type { LogbookStatusPayload } from "./logbook-types.ts"; @@ -71,22 +72,6 @@ describe("Logbook controller", () => { vi.useRealTimers(); }); - it("rebinds polling when the gateway client changes", async () => { - vi.useFakeTimers(); - const host = {}; - hosts.push(host); - const state = getLogbookState(host); - const firstRequest = vi.fn(async () => ({})); - const secondRequest = vi.fn(async () => ({})); - - configureLogbookPolling(state, clientWithRequest(firstRequest), true); - configureLogbookPolling(state, clientWithRequest(secondRequest), true); - await vi.advanceTimersByTimeAsync(30_000); - - expect(firstRequest).not.toHaveBeenCalled(); - expect(secondRequest).toHaveBeenCalled(); - }); - it("lets an in-flight load settle after polling stops", async () => { const host = {}; hosts.push(host); @@ -101,13 +86,12 @@ describe("Logbook controller", () => { ["logbook.days", days], ["logbook.timeline", timeline], ]); - const request = loadLogbook( - state, - clientWithRequest( - (method) => - responses.get(method)?.promise ?? Promise.reject(new Error(`Unexpected ${method}`)), - ), + const client = clientWithRequest( + (method) => + responses.get(method)?.promise ?? Promise.reject(new Error(`Unexpected ${method}`)), ); + configureLogbookPolling(state, client, true); + const request = loadLogbook(state, client); stopLogbookPolling(host); status.resolve(statusFor("2026-07-04")); @@ -170,53 +154,52 @@ describe("Logbook controller", () => { expect(state.timeline?.cards[0]?.title).toBe("Resumed poll"); }); - it("retires a pending poll refresh when the client changes", async () => { + it("retires silent refresh ownership while polling is inactive", async () => { vi.useFakeTimers(); const host = {}; hosts.push(host); const state = getLogbookState(host); state.day = "2026-07-04"; state.dayPinned = true; - const oldStatus = deferred(); - const oldDays = deferred(); - const oldTimeline = deferred(); - const oldResponses = new Map([ - ["logbook.status", oldStatus], - ["logbook.days", oldDays], - ["logbook.timeline", oldTimeline], + const staleStatus = deferred(); + const staleDays = deferred(); + const staleTimeline = deferred(); + const staleBatch = new Map([ + ["logbook.status", staleStatus], + ["logbook.days", staleDays], + ["logbook.timeline", staleTimeline], ]); - const oldRequest = vi.fn((method: string) => { - const response = oldResponses.get(method); - if (!response) { - throw new Error(`Unexpected old-client request: ${method}`); + const request = vi.fn((method: string) => { + const stale = staleBatch.get(method); + if (stale) { + staleBatch.delete(method); + return stale.promise; } - return response.promise; - }); - const newRequest = vi.fn(async (method: string) => { if (method === "logbook.status") { - return statusFor("2026-07-04"); + return Promise.resolve(statusFor("2026-07-04")); } if (method === "logbook.days") { - return { days: [] }; + return Promise.resolve({ days: [] }); } - return timelineFor("2026-07-04", "New client"); + return Promise.resolve(timelineFor("2026-07-04", "Reactivated poll")); }); + const client = clientWithRequest(request); - configureLogbookPolling(state, clientWithRequest(oldRequest), true); + configureLogbookPolling(state, client, true); await vi.advanceTimersByTimeAsync(30_000); - expect(oldRequest).toHaveBeenCalledTimes(3); + expect(request).toHaveBeenCalledTimes(3); - configureLogbookPolling(state, clientWithRequest(newRequest), true); + configureLogbookPolling(state, null, false); + configureLogbookPolling(state, client, true); await vi.advanceTimersByTimeAsync(30_000); - expect(newRequest).toHaveBeenCalledTimes(3); - expect(state.timeline?.cards[0]?.title).toBe("New client"); + expect(request).toHaveBeenCalledTimes(6); + expect(state.timeline?.cards[0]?.title).toBe("Reactivated poll"); - oldStatus.resolve(statusFor("2026-07-04")); - oldDays.resolve({ days: [] }); - oldTimeline.resolve(timelineFor("2026-07-04", "Retired client")); + staleStatus.resolve(statusFor("2026-07-04")); + staleDays.resolve({ days: [] }); + staleTimeline.resolve(timelineFor("2026-07-04", "Inactive poll")); await vi.advanceTimersByTimeAsync(0); - - expect(state.timeline?.cards[0]?.title).toBe("New client"); + expect(state.timeline?.cards[0]?.title).toBe("Reactivated poll"); }); it("shares the background refresh owner with analysis completion", async () => { @@ -271,6 +254,67 @@ describe("Logbook controller", () => { expect(state.timeline?.cards[0]?.title).toBe("Resumed poll"); }); + it("retires action ownership when the polling client changes", async () => { + const host = {}; + hosts.push(host); + const state = getLogbookState(host); + state.day = "2026-07-04"; + state.dayPinned = true; + const oldAnalysis = deferred(); + const newAnalysis = deferred(); + const oldClient = clientWithRequest(() => oldAnalysis.promise); + const newClient = clientWithRequest((method) => { + if (method === "logbook.analyze.now") { + return newAnalysis.promise; + } + if (method === "logbook.status") { + return Promise.resolve(statusFor("2026-07-04")); + } + if (method === "logbook.days") { + return Promise.resolve({ days: [] }); + } + return Promise.resolve(timelineFor("2026-07-04", "New client")); + }); + + configureLogbookPolling(state, oldClient, true); + const oldRequest = runLogbookAnalysisNow(state, oldClient); + expect(state.actionPending).toBe(true); + + configureLogbookPolling(state, newClient, true); + expect(state.actionPending).toBe(false); + const newRequest = runLogbookAnalysisNow(state, newClient); + expect(state.actionPending).toBe(true); + + oldAnalysis.resolve({ started: true }); + await oldRequest; + expect(state.actionPending).toBe(true); + + newAnalysis.resolve({ started: true }); + await newRequest; + expect(state.actionPending).toBe(false); + }); + + it("discards a capture result from a retired polling client", async () => { + const host = {}; + hosts.push(host); + const state = getLogbookState(host); + const oldStatus = deferred(); + const oldClient = clientWithRequest(() => oldStatus.promise); + const newStatus = { ...statusFor("2026-07-05"), capturePaused: true }; + const newClient = clientWithRequest(() => Promise.resolve(newStatus)); + + configureLogbookPolling(state, oldClient, true); + const oldRequest = setLogbookCapturePaused(state, oldClient, true); + configureLogbookPolling(state, newClient, true); + await setLogbookCapturePaused(state, newClient, true); + expect(state.status).toEqual(newStatus); + + oldStatus.resolve(statusFor("2026-07-04")); + await oldRequest; + expect(state.status).toEqual(newStatus); + expect(state.actionPending).toBe(false); + }); + it("queues an analysis refresh behind an in-flight poll", async () => { vi.useFakeTimers(); const host = {}; @@ -321,54 +365,48 @@ describe("Logbook controller", () => { expect(state.timeline?.cards[0]?.title).toBe("Post-analysis refresh"); }); - it("does not let a retired analysis action affect the new client", async () => { + it("drops a queued analysis refresh when polling stops", async () => { vi.useFakeTimers(); const host = {}; hosts.push(host); const state = getLogbookState(host); state.day = "2026-07-04"; state.dayPinned = true; - const oldAnalysis = deferred(); - const newAnalysis = deferred(); - const oldRequest = vi.fn((method: string) => { - if (method !== "logbook.analyze.now") { - throw new Error(`Unexpected retired-client request: ${method}`); - } - return oldAnalysis.promise; - }); - const newRequest = vi.fn(async (method: string) => { + const status = deferred(); + const days = deferred(); + const timeline = deferred(); + const pending = new Map([ + ["logbook.status", status], + ["logbook.days", days], + ["logbook.timeline", timeline], + ]); + const request = vi.fn((method: string) => { if (method === "logbook.analyze.now") { - return await newAnalysis.promise; + return Promise.resolve({ started: true }); } - if (method === "logbook.status") { - return statusFor("2026-07-04"); + const response = pending.get(method); + if (!response) { + throw new Error(`Unexpected refresh request: ${method}`); } - if (method === "logbook.days") { - return { days: [] }; - } - return timelineFor("2026-07-04", "New client"); + pending.delete(method); + return response.promise; }); - const oldClient = clientWithRequest(oldRequest); - const newClient = clientWithRequest(newRequest); + const client = clientWithRequest(request); - configureLogbookPolling(state, oldClient, true); - const oldAction = runLogbookAnalysisNow(state, oldClient); - configureLogbookPolling(state, newClient, true); - const newAction = runLogbookAnalysisNow(state, newClient); - expect(state.actionPending).toBe(true); + configureLogbookPolling(state, client, true); + await vi.advanceTimersByTimeAsync(30_000); + expect(request).toHaveBeenCalledTimes(3); + await runLogbookAnalysisNow(state, client); + expect(request).toHaveBeenCalledTimes(4); - oldAnalysis.resolve({ started: false, reason: "Retired analysis error" }); - await oldAction; - expect(oldRequest).toHaveBeenCalledTimes(1); - expect(state.actionPending).toBe(true); - expect(state.error).not.toBe("Retired analysis error"); - - newAnalysis.resolve({ started: true }); - await newAction; + stopLogbookPolling(host); + status.resolve(statusFor("2026-07-04")); + days.resolve({ days: [] }); + timeline.resolve(timelineFor("2026-07-04", "Detached host")); await vi.advanceTimersByTimeAsync(0); - expect(state.actionPending).toBe(false); - expect(newRequest).toHaveBeenCalledTimes(4); - expect(state.timeline?.cards[0]?.title).toBe("New client"); + + expect(request).toHaveBeenCalledTimes(4); + expect(state.timeline?.cards[0]?.title).toBe("Detached host"); }); it("does not let an older day load overwrite a newer selection", async () => { @@ -400,10 +438,14 @@ describe("Logbook controller", () => { return timelineFor("2026-07-05", "New day"); }); - const olderLoad = loadLogbook(state, clientWithRequest(oldRequest), { day: "2026-07-04" }); + const oldClient = clientWithRequest(oldRequest); + const newClient = clientWithRequest(newerRequest); + configureLogbookPolling(state, oldClient, true); + const olderLoad = loadLogbook(state, oldClient, { day: "2026-07-04" }); expect(oldRequest).toHaveBeenCalledWith("logbook.timeline", { day: "2026-07-04" }); - await loadLogbook(state, clientWithRequest(newerRequest), { day: "2026-07-05" }); + configureLogbookPolling(state, newClient, true); + await loadLogbook(state, newClient, { day: "2026-07-05" }); expect(newerRequest).toHaveBeenCalledWith("logbook.timeline", { day: "2026-07-05" }); expect(state.timeline?.cards[0]?.title).toBe("New day"); @@ -425,11 +467,9 @@ describe("Logbook controller", () => { const state = getLogbookState(host); state.day = "2026-07-04"; const pending = deferred(); - const request = loadLogbookStandup( - state, - clientWithRequest(() => pending.promise), - false, - ); + const client = clientWithRequest(() => pending.promise); + configureLogbookPolling(state, client, true); + const request = loadLogbookStandup(state, client, false); state.day = "2026-07-05"; pending.resolve({ day: "2026-07-04", text: "Old day", updatedMs: 1 }); @@ -445,10 +485,9 @@ describe("Logbook controller", () => { state.day = "2026-07-04"; state.askQuestion = "What did I do?"; const pending = deferred(); - const request = askLogbook( - state, - clientWithRequest(() => pending.promise), - ); + const client = clientWithRequest(() => pending.promise); + configureLogbookPolling(state, client, true); + const request = askLogbook(state, client); state.day = "2026-07-05"; pending.resolve({ answer: "Old day" }); diff --git a/ui/src/pages/plugin/logbook-controller.ts b/ui/src/pages/plugin/logbook-controller.ts index 0aaf534c405..738dbaf1618 100644 --- a/ui/src/pages/plugin/logbook-controller.ts +++ b/ui/src/pages/plugin/logbook-controller.ts @@ -1,7 +1,6 @@ // Control UI controller for the Logbook tab: state, gateway calls, polling. import type { GatewayBrowserClient } from "../../api/gateway.ts"; import type { - LogbookBackgroundRefresh, LogbookDaysPayload, LogbookStatusPayload, LogbookTimelinePayload, @@ -11,7 +10,22 @@ import type { const FRAME_PREVIEW_CACHE_LIMIT = 48; const POLL_INTERVAL_MS = 30_000; -const logbookStates = new WeakMap(); +type LogbookControllerState = LogbookUiState & { + // Client identity is the controller epoch. Rebinding retires every async + // owner so an old gateway cannot mutate or block the replacement view. + client: GatewayBrowserClient | null; + clientGeneration: number; + // Every load advances result ownership; foreground loading state has its own + // owner so a superseded request cannot clear a newer spinner. + loadGeneration: number; + loadingGeneration: number | null; + backgroundRefresh: Promise | null; + backgroundRefreshQueued: boolean; + pollTimer: ReturnType | null; + pollClient: GatewayBrowserClient | null; +}; + +const logbookStates = new WeakMap(); export function localDayKey(date = new Date()): string { const month = String(date.getMonth() + 1).padStart(2, "0"); @@ -25,7 +39,7 @@ export function shiftDay(day: string, deltaDays: number): string { return localDayKey(base); } -export function getLogbookState(host: object): LogbookUiState { +export function getLogbookState(host: object): LogbookControllerState { let state = logbookStates.get(host); if (!state) { state = { @@ -46,13 +60,12 @@ export function getLogbookState(host: object): LogbookUiState { askAnswer: null, askLoading: false, actionPending: false, - actionGeneration: 0, - actionPendingGeneration: null, + client: null, + clientGeneration: 0, loadGeneration: 0, loadingGeneration: null, - lifecycleGeneration: 0, backgroundRefresh: null, - backgroundRefreshQueued: null, + backgroundRefreshQueued: false, pollTimer: null, pollClient: null, requestUpdate: null, @@ -66,7 +79,39 @@ function notify(state: LogbookUiState): void { state.requestUpdate?.(); } -function resetDayView(state: LogbookUiState, day: string): void { +function ownsClient( + state: LogbookControllerState, + client: GatewayBrowserClient, + generation: number, +): boolean { + return state.client === client && state.clientGeneration === generation; +} + +function currentClientGeneration( + state: LogbookControllerState, + client: GatewayBrowserClient | null, +): number | null { + return client && state.client === client ? state.clientGeneration : null; +} + +function bindClient(state: LogbookControllerState, client: GatewayBrowserClient | null): void { + if (state.client === client) { + return; + } + state.client = client; + state.clientGeneration += 1; + state.loadGeneration += 1; + state.loadingGeneration = null; + state.loading = false; + state.backgroundRefresh = null; + state.backgroundRefreshQueued = false; + state.actionPending = false; + state.standupLoading = false; + state.askLoading = false; + state.frameLoads = new Set(); +} + +function resetDayView(state: LogbookControllerState, day: string): void { state.day = day; state.timeline = null; state.standup = null; @@ -75,11 +120,12 @@ function resetDayView(state: LogbookUiState, day: string): void { } export async function loadLogbook( - state: LogbookUiState, + state: LogbookControllerState, client: GatewayBrowserClient | null, opts?: { day?: string; today?: boolean; silent?: boolean }, ): Promise { - if (!client) { + const clientGeneration = currentClientGeneration(state, client); + if (!client || clientGeneration === null) { return; } if (opts?.day) { @@ -104,7 +150,11 @@ export async function loadLogbook( client.request("logbook.days", {}), client.request("logbook.timeline", { day: requestedDay }), ]); - if (generation !== state.loadGeneration || state.day !== requestedDay) { + if ( + !ownsClient(state, client, clientGeneration) || + generation !== state.loadGeneration || + state.day !== requestedDay + ) { return; } state.status = status; @@ -117,7 +167,11 @@ export async function loadLogbook( const todayTimeline = await client.request("logbook.timeline", { day: status.today, }); - if (generation !== state.loadGeneration || state.day !== status.today) { + if ( + !ownsClient(state, client, clientGeneration) || + generation !== state.loadGeneration || + state.day !== status.today + ) { return; } state.timeline = todayTimeline; @@ -126,11 +180,12 @@ export async function loadLogbook( } state.error = null; } catch (err) { - if (generation === state.loadGeneration) { + if (ownsClient(state, client, clientGeneration) && generation === state.loadGeneration) { state.error = err instanceof Error ? err.message : String(err); } } finally { - let shouldNotify = generation === state.loadGeneration; + let shouldNotify = + ownsClient(state, client, clientGeneration) && generation === state.loadGeneration; if (state.loadingGeneration === generation) { state.loadingGeneration = null; state.loading = false; @@ -143,68 +198,29 @@ export async function loadLogbook( } } -function retireLogbookLoads(state: LogbookUiState): void { - // A stopped or rebound view must not accept results from its retired client, - // and an abandoned background request must not block the next polling epoch. - state.loadGeneration += 1; - state.lifecycleGeneration += 1; - state.actionGeneration += 1; - state.actionPendingGeneration = null; - state.actionPending = false; - state.loadingGeneration = null; - state.loading = false; - state.backgroundRefresh = null; - state.backgroundRefreshQueued = null; -} - -function isLogbookActionCurrent( - state: LogbookUiState, - actionGeneration: number, - refresh: LogbookBackgroundRefresh, -): boolean { - return actionGeneration === state.actionGeneration && isLogbookRefreshCurrent(state, refresh); -} - -function isLogbookRefreshCurrent( - state: LogbookUiState, - refresh: LogbookBackgroundRefresh, -): boolean { - return ( - refresh.lifecycleGeneration === state.lifecycleGeneration && - (state.pollClient === null || state.pollClient === refresh.client) - ); -} - -function drainQueuedLogbookRefresh(state: LogbookUiState): void { - if (state.loading || state.backgroundRefresh) { +function drainQueuedLogbookRefresh(state: LogbookControllerState): void { + if (!state.backgroundRefreshQueued || state.loading || state.backgroundRefresh) { return; } - const queued = state.backgroundRefreshQueued; - state.backgroundRefreshQueued = null; - if (!queued || !isLogbookRefreshCurrent(state, queued)) { + state.backgroundRefreshQueued = false; + const client = state.pollClient; + if (!client) { return; } - void refreshLogbookSilently(state, queued.client, { - lifecycleGeneration: queued.lifecycleGeneration, - required: true, - }); + void refreshLogbookSilently(state, client, { required: true }); } function refreshLogbookSilently( - state: LogbookUiState, + state: LogbookControllerState, client: GatewayBrowserClient, - opts?: { lifecycleGeneration?: number; required?: boolean }, + opts?: { required?: boolean }, ): Promise { - const refreshRequest = { - client, - lifecycleGeneration: opts?.lifecycleGeneration ?? state.lifecycleGeneration, - }; - if (!isLogbookRefreshCurrent(state, refreshRequest)) { + if (state.pollClient !== client) { return Promise.resolve(); } if (state.loading || state.backgroundRefresh) { if (opts?.required) { - state.backgroundRefreshQueued = refreshRequest; + state.backgroundRefreshQueued = true; } return state.backgroundRefresh ?? Promise.resolve(); } @@ -228,13 +244,14 @@ export function stopLogbookPolling(host: object): void { } if (state) { state.pollClient = null; - // PluginPage replaces the host before calling stop. Let detached-host loads + state.backgroundRefreshQueued = false; + // PluginPage retires this host immediately after stop returns. Let its loads // settle; host identity keeps their results out of the replacement view. } } export function configureLogbookPolling( - state: LogbookUiState, + state: LogbookControllerState, client: GatewayBrowserClient | null, active: boolean, ): void { @@ -244,7 +261,10 @@ export function configureLogbookPolling( state.pollTimer = null; } state.pollClient = null; - retireLogbookLoads(state); + state.backgroundRefreshQueued = false; + // Unlike stopLogbookPolling's detached-host path, this state can render + // again after reconnect. Retire every old async owner before reuse. + bindClient(state, null); return; } if (state.pollTimer && state.pollClient === client) { @@ -253,7 +273,7 @@ export function configureLogbookPolling( if (state.pollTimer) { clearInterval(state.pollTimer); } - retireLogbookLoads(state); + bindClient(state, client); state.pollClient = client; state.pollTimer = setInterval(() => { // All background refresh sources share one owner so slow gateway responses @@ -263,12 +283,14 @@ export function configureLogbookPolling( } export async function loadLogbookFramePreview( - state: LogbookUiState, + state: LogbookControllerState, client: GatewayBrowserClient | null, frameId: number, ): Promise { + const clientGeneration = currentClientGeneration(state, client); if ( !client || + clientGeneration === null || state.framePreviews.has(frameId) || state.frameLoads.has(frameId) || state.framePreviewFailed.has(frameId) @@ -280,6 +302,9 @@ export async function loadLogbookFramePreview( const payload = await client.request<{ base64: string; format: string }>("logbook.frame", { frameId, }); + if (!ownsClient(state, client, clientGeneration)) { + return; + } if (state.framePreviews.size >= FRAME_PREVIEW_CACHE_LIMIT) { const oldest = state.framePreviews.keys().next().value; if (oldest !== undefined) { @@ -290,38 +315,39 @@ export async function loadLogbookFramePreview( } catch { // Preview loads are cosmetic, but a missing frame (e.g. pruned by // retention) must not re-fetch on every render, so remember the failure. - state.framePreviewFailed.add(frameId); + if (ownsClient(state, client, clientGeneration)) { + state.framePreviewFailed.add(frameId); + } } finally { - state.frameLoads.delete(frameId); - notify(state); + if (ownsClient(state, client, clientGeneration)) { + state.frameLoads.delete(frameId); + notify(state); + } } } export async function setLogbookCapturePaused( - state: LogbookUiState, + state: LogbookControllerState, client: GatewayBrowserClient | null, paused: boolean, ): Promise { - if (!client || state.actionPending) { + const clientGeneration = currentClientGeneration(state, client); + if (!client || clientGeneration === null || state.actionPending) { return; } - const actionGeneration = ++state.actionGeneration; - state.actionPendingGeneration = actionGeneration; state.actionPending = true; notify(state); - const refresh = { client, lifecycleGeneration: state.lifecycleGeneration }; try { const status = await client.request("logbook.capture.set", { paused }); - if (isLogbookActionCurrent(state, actionGeneration, refresh)) { + if (ownsClient(state, client, clientGeneration)) { state.status = status; } } catch (err) { - if (isLogbookActionCurrent(state, actionGeneration, refresh)) { + if (ownsClient(state, client, clientGeneration)) { state.error = err instanceof Error ? err.message : String(err); } } finally { - if (state.actionPendingGeneration === actionGeneration) { - state.actionPendingGeneration = null; + if (ownsClient(state, client, clientGeneration)) { state.actionPending = false; notify(state); } @@ -329,54 +355,43 @@ export async function setLogbookCapturePaused( } export async function runLogbookAnalysisNow( - state: LogbookUiState, + state: LogbookControllerState, client: GatewayBrowserClient | null, ): Promise { - if (!client || state.actionPending) { + const clientGeneration = currentClientGeneration(state, client); + if (!client || clientGeneration === null || state.actionPending) { return; } - const actionGeneration = ++state.actionGeneration; - state.actionPendingGeneration = actionGeneration; state.actionPending = true; notify(state); - const refresh = { client, lifecycleGeneration: state.lifecycleGeneration }; try { const result = await client.request<{ started: boolean; reason?: string }>( "logbook.analyze.now", {}, ); - if ( - isLogbookActionCurrent(state, actionGeneration, refresh) && - !result.started && - result.reason - ) { + if (ownsClient(state, client, clientGeneration) && !result.started && result.reason) { state.error = result.reason; } } catch (err) { - if (isLogbookActionCurrent(state, actionGeneration, refresh)) { + if (ownsClient(state, client, clientGeneration)) { state.error = err instanceof Error ? err.message : String(err); } } finally { - if (state.actionPendingGeneration === actionGeneration) { - state.actionPendingGeneration = null; + if (ownsClient(state, client, clientGeneration)) { state.actionPending = false; notify(state); - } - if (isLogbookActionCurrent(state, actionGeneration, refresh)) { - void refreshLogbookSilently(state, client, { - lifecycleGeneration: refresh.lifecycleGeneration, - required: true, - }); + void refreshLogbookSilently(state, client, { required: true }); } } } export async function loadLogbookStandup( - state: LogbookUiState, + state: LogbookControllerState, client: GatewayBrowserClient | null, refresh: boolean, ): Promise { - if (!client || state.standupLoading) { + const clientGeneration = currentClientGeneration(state, client); + if (!client || clientGeneration === null || state.standupLoading) { return; } state.standupLoading = true; @@ -387,23 +402,28 @@ export async function loadLogbookStandup( "logbook.standup", { day: requestedDay, refresh }, ); - if (state.day === requestedDay) { + if (ownsClient(state, client, clientGeneration) && state.day === requestedDay) { state.standup = standup; } } catch (err) { - state.error = err instanceof Error ? err.message : String(err); + if (ownsClient(state, client, clientGeneration)) { + state.error = err instanceof Error ? err.message : String(err); + } } finally { - state.standupLoading = false; - notify(state); + if (ownsClient(state, client, clientGeneration)) { + state.standupLoading = false; + notify(state); + } } } export async function askLogbook( - state: LogbookUiState, + state: LogbookControllerState, client: GatewayBrowserClient | null, ): Promise { const question = state.askQuestion.trim(); - if (!client || state.askLoading || question.length === 0) { + const clientGeneration = currentClientGeneration(state, client); + if (!client || clientGeneration === null || state.askLoading || question.length === 0) { return; } state.askLoading = true; @@ -415,13 +435,17 @@ export async function askLogbook( day: requestedDay, question, }); - if (state.day === requestedDay) { + if (ownsClient(state, client, clientGeneration) && state.day === requestedDay) { state.askAnswer = payload.answer; } } catch (err) { - state.error = err instanceof Error ? err.message : String(err); + if (ownsClient(state, client, clientGeneration)) { + state.error = err instanceof Error ? err.message : String(err); + } } finally { - state.askLoading = false; - notify(state); + if (ownsClient(state, client, clientGeneration)) { + state.askLoading = false; + notify(state); + } } } diff --git a/ui/src/pages/plugin/logbook-types.ts b/ui/src/pages/plugin/logbook-types.ts index b6275b38537..7a88bb31845 100644 --- a/ui/src/pages/plugin/logbook-types.ts +++ b/ui/src/pages/plugin/logbook-types.ts @@ -1,5 +1,3 @@ -import type { GatewayBrowserClient } from "../../api/gateway.ts"; - export type LogbookStatusPayload = { captureEnabled: boolean; capturePaused: boolean; @@ -54,11 +52,6 @@ export type LogbookDaysPayload = { days: Array<{ day: string; cards: number; firstMs: number; lastMs: number }>; }; -export type LogbookBackgroundRefresh = { - client: GatewayBrowserClient; - lifecycleGeneration: number; -}; - export type LogbookUiState = { day: string; /** True once the user navigated to a specific day; unpinned views follow the gateway's today. */ @@ -78,16 +71,5 @@ export type LogbookUiState = { askAnswer: string | null; askLoading: boolean; actionPending: boolean; - actionGeneration: number; - actionPendingGeneration: number | null; - // Every load advances result ownership; foreground loading state has its own - // owner so a superseded request cannot clear a newer spinner. - loadGeneration: number; - loadingGeneration: number | null; - lifecycleGeneration: number; - backgroundRefresh: Promise | null; - backgroundRefreshQueued: LogbookBackgroundRefresh | null; - pollTimer: ReturnType | null; - pollClient: GatewayBrowserClient | null; requestUpdate: (() => void) | null; }; diff --git a/ui/src/pages/plugin/logbook-view.ts b/ui/src/pages/plugin/logbook-view.ts index ff7943c9f53..293c0cd4c64 100644 --- a/ui/src/pages/plugin/logbook-view.ts +++ b/ui/src/pages/plugin/logbook-view.ts @@ -28,6 +28,8 @@ type LogbookProps = { onRequestUpdate?: () => void; }; +type LogbookControllerState = ReturnType; + function formatClock(ms: number, timeZone: string): string { return formatTimeMs(ms, { hour: "2-digit", minute: "2-digit", timeZone }, ""); } @@ -95,7 +97,7 @@ function renderStatusChips(status: LogbookStatusPayload): TemplateResult { } function renderCard( - state: LogbookUiState, + state: LogbookControllerState, client: GatewayBrowserClient | null, card: LogbookCardPayload, timeZone: string, @@ -246,7 +248,10 @@ function renderStats(state: LogbookUiState): TemplateResult | typeof nothing { `; } -function renderStandup(state: LogbookUiState, client: GatewayBrowserClient | null): TemplateResult { +function renderStandup( + state: LogbookControllerState, + client: GatewayBrowserClient | null, +): TemplateResult { return html`
@@ -273,7 +278,10 @@ function renderStandup(state: LogbookUiState, client: GatewayBrowserClient | nul `; } -function renderAsk(state: LogbookUiState, client: GatewayBrowserClient | null): TemplateResult { +function renderAsk( + state: LogbookControllerState, + client: GatewayBrowserClient | null, +): TemplateResult { return html`
${t("logbook.ask.title")}