diff --git a/packages/workshop-backend/src/overseer.ts b/packages/workshop-backend/src/overseer.ts index 8fded60d7..96ce7a5fb 100644 --- a/packages/workshop-backend/src/overseer.ts +++ b/packages/workshop-backend/src/overseer.ts @@ -1,6 +1,6 @@ import { RpcCompatible, RpcStub, RpcTarget } from "capnweb"; import { validateRpc } from "capnweb-validate"; -import { Overseer, GadgetMetadata, UiBundle, WorkpieceId, WorkpieceSummary, WorkpiecesSubscriber, GadgetClient, GadgetBindingInfo, GatekeeperClient, ActionState, ActionLogEntry, ActionsSubscriber, ActionHistoryFilter, ActionHistoryPage, ChatGadgetPin, ChatCodeBase, ChatGadgetPinState, CodeChangeSubmission, CommitIdentity, CommitInfo, MergeChangesResult, AiChatMetadata, AiChatMessage, AiChatHistoryPage, AiChatSubscriber, AiChatAuthorInfo, AiModelConfig, AiChatMessageBody, AgentSpawnerConfig, ConsoleLogSubscriber, ConsoleLogEvent, CapsuleSpecifier, CollaboratorInfo, CollaboratorRole, AffectedCollaborator, ShareLinkInfo, GatekeeperCreationSpec, ObserverConfigCallback, ObserverBindingNeed, ObserverBindingFailure, BlueprintBindingAnnotation, BlueprintBinding, BlueprintMetadata, BlueprintOutput, MessageFormatRef, isOutputIcon, SpawnerEnvTarget, BlueprintGadgetSummary, AiChatStreamEvent, BlueprintScreenshotUpload, BLUEPRINT_SCREENSHOT_R2_PREFIX, blueprintScreenshotUrl, ChatAttachmentUpload, ChatAttachmentHandle, ChatAttachmentRef, BoundHookInfo, PreApprovableAction, PresenceParticipant, PresenceSubscriber, SlashCommandChoice, SlashCommandRequest, validateBindingName, createOpenGadgetError, OPEN_GADGET_ERROR_CODES, resolveSiteName } from '@gadgets/workshop-shared/api'; +import { Overseer, GadgetMetadata, UiBundle, WorkpieceId, WorkpieceSummary, WorkpiecesSubscriber, GadgetClient, GadgetBindingInfo, GatekeeperClient, ActionState, ActionLogEntry, ActionsSubscriber, ActionHistoryFilter, ActionHistoryPage, ChatGadgetPin, ChatCodeBase, ChatGadgetPinState, CodeChangeSubmission, CommitIdentity, CommitInfo, MergeChangesResult, AiChatMetadata, AiChatMessage, AiChatHistoryPage, AiChatSubscriber, AiChatAuthorInfo, AiModelConfig, AiChatMessageBody, AgentSpawnerConfig, ConsoleLogSubscriber, ConsoleLogEvent, CapsuleSpecifier, CollaboratorInfo, CollaboratorRole, AffectedCollaborator, ShareLinkInfo, GatekeeperCreationSpec, ObserverConfigCallback, ObserverBindingNeed, ObserverBindingFailure, BlueprintBindingAnnotation, BlueprintBinding, BlueprintMetadata, BlueprintOutput, MessageFormatRef, isOutputIcon, SpawnerEnvTarget, BlueprintGadgetSummary, AiChatStreamEvent, BlueprintScreenshotUpload, BLUEPRINT_SCREENSHOT_R2_PREFIX, blueprintScreenshotUrl, ChatAttachmentUpload, ChatAttachmentHandle, ChatAttachmentRef, BoundHookInfo, PreApprovableAction, PresenceParticipant, PresenceSubscriber, SlashCommandChoice, SlashCommandRequest, validateBindingName, createOpenGadgetError, OPEN_GADGET_ERROR_CODES, resolveSiteName, actionChangeTime } from '@gadgets/workshop-shared/api'; import { applyCodeChange, changedGadgets, composeCodeChange, diffFiles, transformCodeChange, validateCodeChangeContent, validateCodeChangeSchema, type CodeContent, type CodeChange } from "@gadgets/workshop-shared/code-change"; @@ -943,7 +943,7 @@ function stampBindHookAction(storage: OverseerStorage, actionId: number, enabled // frozen clock makes same-instant records routine. Every mutation path stamps appliedAt (apply, // reject, stampBindHookAction); one that doesn't would be missed by the resume replay. function actionLastChangedKey(record: ActionRecord): string { - return `${keyString((record.appliedAt ?? record.createdAt).valueOf())}.${keyString(record.id)}`; + return `${keyString(actionChangeTime(record).valueOf())}.${keyString(record.id)}`; } /** diff --git a/packages/workshop-frontend/src/Activity.tsx b/packages/workshop-frontend/src/Activity.tsx index 31cf4ee24..b8e1505f8 100644 --- a/packages/workshop-frontend/src/Activity.tsx +++ b/packages/workshop-frontend/src/Activity.tsx @@ -2,7 +2,7 @@ import { useEffect, useMemo, useRef, useState, type ReactNode } from 'react' import { Switch, useKumoToastManager } from '@cloudflare/kumo' import { CaretRight, Check, Eye, Lightning, ShieldCheck } from '@phosphor-icons/react' import { RpcStub } from 'capnweb' -import { ActionLogEntry, Overseer } from '@gadgets/workshop-shared/api' +import { ActionLogEntry, Overseer, actionChangeTime } from '@gadgets/workshop-shared/api' import { ActionKind } from '@gadgets/workshop-shared/gatekeeper' import { GatekeeperIcon } from './components/GatekeeperIcon' import { HookToggle } from './components/HookToggle' @@ -178,7 +178,7 @@ export default function Activity({ const historyGroups = useMemo(() => { const groups: { label: string; records: ActionLogEntry[] }[] = [] for (const record of history.entries) { - const label = dayLabel(record.appliedAt ?? record.createdAt) + const label = dayLabel(actionChangeTime(record)) const last = groups.at(-1) if (last?.label === label) last.records.push(record) else groups.push({ label, records: [record] }) @@ -676,7 +676,7 @@ function HistoryRow({ const resourceUrl = safeExternalUrl(record.resourceUrl) const resolvedBy = record.type === 'action' ? record.resolvedBy : undefined const autoApproved = record.type === 'action' && record.autoApproved === true - const at = record.appliedAt ?? record.createdAt + const at = actionChangeTime(record) const status = activityStatus(record) return ( diff --git a/packages/workshop-frontend/src/ChatInterface.actions.test.tsx b/packages/workshop-frontend/src/ChatInterface.actions.test.tsx new file mode 100644 index 000000000..cb8130f6d --- /dev/null +++ b/packages/workshop-frontend/src/ChatInterface.actions.test.tsx @@ -0,0 +1,141 @@ +// @vitest-environment jsdom +/* eslint-disable react/react-in-jsx-scope */ + +import { act } from 'react' +import { afterEach, describe, expect, it, vi } from 'vitest' +import type { RpcStub } from 'capnweb' +import type { AiChatMessage, AiChatSubscriber, Overseer } from '@gadgets/workshop-shared/api' + +vi.stubGlobal('ResizeObserver', class { + observe() {} + disconnect() {} +}) + +vi.mock('@cloudflare/kumo', async (importOriginal) => { + const actual = await importOriginal() as typeof import('@cloudflare/kumo') + const Pass = ({ children }: { children?: React.ReactNode }) => children ?? null + const Null = () => null + const parts = new Proxy(Pass, { + get: (_target, property) => property === 'Root' ? Null : Pass, + }) + const toasts = { add: vi.fn<(options: unknown) => void>() } + return { + ...actual, + Dialog: parts, + DropdownMenu: parts, + Popover: parts, + Tooltip: Pass, + useKumoToastManager: () => toasts, + } +}) + +vi.mock('./AuthContext', () => { + const context = { + authenticatedApi: { listGatekeeperVendors: async () => [] }, + currentUser: null, + } + return { + useAuthenticatedApi: () => context, + useOptionalAuthenticatedApi: () => null, + } +}) + +import { entry, makeOverseer, makeTestRoot } from './action-test-harness' +import ChatInterface from './ChatInterface' +import { linkActionLog } from './useActions' + +const testRoot = makeTestRoot() + +afterEach(() => { + testRoot.cleanup() + vi.restoreAllMocks() +}) + +function withChatApi( + server: ReturnType, + getChatMessage = vi.fn<(chatId: number, sequence: number) => Promise>(), +) { + let subscriber: AiChatSubscriber | undefined + Object.assign(server.overseer as object, { + getChatMessage, + listChats: async () => [], + listModels: async () => [], + onRpcBroken: () => {}, + subscribeToChat: (next: AiChatSubscriber) => { + subscriber = next + return { [Symbol.dispose]: () => {} } + }, + }) + return { + getChatMessage, + emitMessage(message: AiChatMessage) { + act(() => subscriber!.message(message)) + }, + } +} + +function renderChat(overseer: RpcStub) { + return testRoot.render( + {}} + pendingConsoleLogCount={0} + consoleLogPreview="" + consoleLogSeverity="info" + onConsumeConsoleLogs={() => ''} + onDiscardConsoleLogs={() => {}} + onOpenGadget={() => {}} + outputOfWorkpiece={() => undefined} + />, + ) +} + +const actionMessage = { + chatId: 1, + sequence: 0, + timestamp: new Date(), + author: { type: 'agent', id: 'model', name: 'Model' }, + type: 'action', + actionId: 1, + actionLog: entry(1), +} as AiChatMessage + +const resolvedMessage = + { ...actionMessage, actionLog: entry(1, { state: 'approved' }) } as AiChatMessage + +// Renders a first session that caches a pending action card, then settles it so a linked swap +// can resume. Pass a key to link the stub; unlinked sessions never park a watermark. +async function cachePendingCard(key?: string) { + const first = makeOverseer() + const firstChat = withChatApi(first) + if (key !== undefined) linkActionLog(first.overseer, key) + await renderChat(first.overseer) + await first.resolveSubscription() + await first.resolvePendingQuery({ entries: [entry(1)] }) + firstChat.emitMessage(actionMessage) +} + +describe('ChatInterface action refresh', () => { + it('refetches cached mutable cards when an unlinked stub swaps', async () => { + await cachePendingCard() + + const second = makeOverseer() + const secondChat = withChatApi(second, vi.fn(async () => resolvedMessage)) + await renderChat(second.overseer) + await vi.waitFor(() => expect(secondChat.getChatMessage).toHaveBeenCalledWith(1, 0)) + }) + + it('skips the cached-card refetch on a resumed linked stub swap', async () => { + await cachePendingCard('ws-chat-resume') + + const second = makeOverseer() + const secondChat = withChatApi(second, vi.fn(async () => resolvedMessage)) + linkActionLog(second.overseer, 'ws-chat-resume') + await renderChat(second.overseer) + await second.resolveSubscription() + await second.resolvePendingQuery({ entries: [entry(1)] }) + expect(secondChat.getChatMessage).not.toHaveBeenCalled() + }) +}) diff --git a/packages/workshop-frontend/src/ChatInterface.tsx b/packages/workshop-frontend/src/ChatInterface.tsx index 31e49fda3..c9d3a5176 100644 --- a/packages/workshop-frontend/src/ChatInterface.tsx +++ b/packages/workshop-frontend/src/ChatInterface.tsx @@ -119,7 +119,7 @@ import DeleteConfirmationDialog from "./components/DeleteConfirmationDialog"; import AutoApproveConfirmDialog from "./components/AutoApproveConfirmDialog"; import { AlwaysApproveButton, ResolveButton } from "./components/ResolveButton"; import { WorkshopButton, WorkshopIconButton, WorkshopInput } from "./components/WorkshopControls"; -import { useActionEntries } from "./useActions"; +import { actionLogResumed, useActionEntries } from "./useActions"; import { useAlwaysApproveTag } from "./useAlwaysApproveTag"; import { useResolveAction } from "./useResolveAction"; import { safeExternalUrl } from "./utils/safeExternalUrl"; @@ -5765,13 +5765,13 @@ function ChatInterface({ useActionEntries(overseer, (record) => { if (applyActionLogUpdateToCachedMessages(record)) scheduleUpdate(); }); - - // On (re)connect, re-fetch cached action cards whose log can still change: blank or pending - // cards (a resolution may have landed while we were away), and bindHook cards, which stay - // mutable after resolution (`enabled` toggles). The action subscription carries live deltas - // only, so changes from the gap never reach us through it. - // TODO: resubscribing with startAfter would replay the gap through the subscription instead. + // On a resumed reconnect the subscription replays the gap, so the entries above cover cached + // cards. Otherwise (cold open, or the prior session never settled) re-fetch cached action + // cards whose log can still change: blank or pending cards (a resolution may have landed + // while we were away), and bindHook cards, which stay mutable after resolution (`enabled` + // toggles). Runs after useActionEntries, whose effect creates the store and its resumed flag. useEffect(() => { + if (actionLogResumed(overseer)) return; let cancelled = false; const targets = [...cacheRef.current.actionMessages.values()].flatMap((locations) => { const location = locations.values().next().value; diff --git a/packages/workshop-frontend/src/useActionHistory.test.tsx b/packages/workshop-frontend/src/useActionHistory.test.tsx index af8d45305..56c0e572e 100644 --- a/packages/workshop-frontend/src/useActionHistory.test.tsx +++ b/packages/workshop-frontend/src/useActionHistory.test.tsx @@ -8,6 +8,7 @@ import type { ActionLogEntry, Overseer } from '@gadgets/workshop-shared/api' import { entry as pendingEntry, makeOverseer, makeTestRoot } from './action-test-harness' import { useActionHistory } from './useActionHistory' import type { HistoryViewFilter } from './useActionHistory' +import { linkActionLog } from './useActions' // Most history fixtures are resolved records; use the harness `pendingEntry` for pending ones. function entry(id: number, over: Partial> = {}): ActionLogEntry { @@ -115,7 +116,7 @@ describe('useActionHistory', () => { expect(latest.status).toBe('ready') }) - it('resets and refetches when the stub changes', async () => { + it('resets and refetches when an unlinked stub changes', async () => { const first = makeOverseer() await render(first.overseer, 'all', true) await first.resolvePage({ entries: [entry(30)], nextBeforeId: 10 }) @@ -135,6 +136,56 @@ describe('useActionHistory', () => { expect(latest.entries.map(e => e.id)).toEqual([40]) }) + it('keeps the loaded window across a resumed linked stub swap', async () => { + const first = makeOverseer() + linkActionLog(first.overseer, 'ws-hist-resume') + await render(first.overseer, 'all', true) + // Settle the shared store so the swap resumes; a pending record sets its watermark. + await first.resolveSubscription() + await first.resolvePendingQuery({ entries: [pendingEntry(1)] }) + await first.resolvePage({ entries: [entry(30), entry(20)], nextBeforeId: 10 }) + + const second = makeOverseer() + linkActionLog(second.overseer, 'ws-hist-resume') + await render(second.overseer, 'all', true) + expect(second.listCalls).toEqual([]) // no automatic refetch + expect(latest.entries.map(e => e.id)).toEqual([30, 20]) + expect(latest.status).toBe('ready') + + // A replayed entry patches an in-window record. + await second.emit(entry(30, { state: 'rejected' })) + expect(latest.entries.map(e => [e.id, e.state])) + .toEqual([[30, 'rejected'], [20, 'approved']]) + + // loadMore continues from the preserved frontier on the new stub. + act(() => latest.loadMore()) + expect(second.listCalls).toEqual([{ beforeId: 10, filter: 'all' }]) + await second.resolvePage({ entries: [entry(5)] }) + expect(latest.entries.map(e => e.id)).toEqual([30, 20, 5]) + expect(latest.hasMore).toBe(false) + }) + + it('drops an in-flight old-stub page after a resumed swap', async () => { + const first = makeOverseer() + linkActionLog(first.overseer, 'ws-hist-inflight') + await render(first.overseer, 'all', true) + await first.resolveSubscription() + await first.resolvePendingQuery({ entries: [pendingEntry(1)] }) + await first.resolvePage({ entries: [entry(30)], nextBeforeId: 10 }) + act(() => latest.loadMore()) // leave a second fetch in flight on the old stub + + const second = makeOverseer() + linkActionLog(second.overseer, 'ws-hist-inflight') + await render(second.overseer, 'all', true) + + await first.resolvePage({ entries: [entry(9)] }) + expect(latest.entries.map(e => e.id)).toEqual([30]) + expect(latest.isLoadingMore).toBe(false) + + act(() => latest.loadMore()) + expect(second.listCalls).toEqual([{ beforeId: 10, filter: 'all' }]) + }) + it('recovers from a failed first load on retry', async () => { vi.spyOn(console, 'error').mockImplementation(() => {}) const server = makeOverseer() diff --git a/packages/workshop-frontend/src/useActionHistory.ts b/packages/workshop-frontend/src/useActionHistory.ts index 617bb967d..f6b1d4e91 100644 --- a/packages/workshop-frontend/src/useActionHistory.ts +++ b/packages/workshop-frontend/src/useActionHistory.ts @@ -2,7 +2,7 @@ import { useCallback, useEffect, useMemo, useRef, useState } from 'react' import type { RpcStub } from 'capnweb' import { matchesActionHistoryFilter } from '@gadgets/workshop-shared/api' import type { ActionHistoryFilter, ActionLogEntry, Overseer } from '@gadgets/workshop-shared/api' -import { useActionEntries } from './useActions' +import { actionLogResumed, useActionEntries } from './useActions' export type ActionHistoryStatus = 'loading' | 'ready' | 'error' @@ -35,8 +35,9 @@ const INITIAL: HistoryState = { byId: new Map(), error: null } /** * Demand-loads action history (pending records included), one page at a time, newest first by id * (creation order). Nothing is fetched until `active` first becomes true; `loadMore()` continues - * from the server cursor, and `hasMore` is the termination signal. The overseer stub or filter - * changing resets everything (a reconnect hands out a fresh stub). + * from the server cursor, and `hasMore` is the termination signal. The filter changing resets + * everything; so does the overseer stub changing (a reconnect hands out a fresh stub), unless + * the shared store resumed — then the loaded window and cursor survive the swap. * * Live updates from the shared action subscription are merged in: a filter-matching record — * a fresh pending one or a resolution — patches in place or inserts if it falls inside the @@ -58,7 +59,7 @@ export function useActionHistory( useEffect(() => { sessionRef.current = createHistorySession() setState(INITIAL) - }, [overseer, filter]) + }, [filter]) const loadMore = useCallback(() => { const session = sessionRef.current @@ -106,6 +107,25 @@ export function useActionHistory( }) }) + // A stub swap resets everything — unless the shared store resumed, in which case the gap was + // replayed through the subscription above: keep the window and the frontier (a server-stable + // id cursor), rebuilding the session token so any in-flight old-stub page is dropped. Ordering + // matters: after useActionEntries (which creates the store, setting its resumed flag), and + // before the initial-load effect (so a reset refetches). + const prevOverseerRef = useRef(overseer) + useEffect(() => { + if (prevOverseerRef.current === overseer) return + prevOverseerRef.current = overseer + if (actionLogResumed(overseer)) { + const { frontier, hasLoadedPage } = sessionRef.current + sessionRef.current = { frontier, inFlight: false, hasLoadedPage } + setState(prev => ({ ...prev, error: null })) + } else { + sessionRef.current = createHistorySession() + setState(INITIAL) + } + }, [overseer]) + useEffect(() => { if (active && !sessionRef.current.hasLoadedPage) loadMore() }, [active, loadMore]) diff --git a/packages/workshop-frontend/src/useActions.test.tsx b/packages/workshop-frontend/src/useActions.test.tsx index 96cbdf706..b4de2fab2 100644 --- a/packages/workshop-frontend/src/useActions.test.tsx +++ b/packages/workshop-frontend/src/useActions.test.tsx @@ -5,7 +5,13 @@ import { afterEach, describe, expect, it, vi } from 'vitest' import type { RpcStub } from 'capnweb' import type { ActionLogEntry, Overseer } from '@gadgets/workshop-shared/api' import { entry, flushFrames, makeOverseer, makeTestRoot } from './action-test-harness' -import { useActionEntries, useActions, type ActionsState } from './useActions' +import { + actionLogResumed, + linkActionLog, + useActionEntries, + useActions, + type ActionsState, +} from './useActions' describe('useActions', () => { const view = makeTestRoot() @@ -118,7 +124,7 @@ describe('useActions', () => { expect(server.pendingQueryCalls).toHaveLength(1) }) - it('starts a fresh subscription when the stub changes', async () => { + it('starts a fresh subscription when an unlinked stub changes', async () => { const first = makeOverseer() await view.render() await first.emit(entry(1)) @@ -130,10 +136,96 @@ describe('useActions', () => { await view.render() expect(second.subscribeCalls).toHaveLength(1) expect(second.pendingQueryCalls).toHaveLength(1) + expect(actionLogResumed(second.overseer)).toBe(false) expect(latest.status).toBe('checking') expect(latest.pending).toEqual([]) }) + it('resubscribes with the settled watermark when a linked stub swaps', async () => { + const first = makeOverseer() + linkActionLog(first.overseer, 'ws-resume') + await view.render() + await first.resolveSubscription() + await first.resolvePendingQuery({ entries: [entry(1)] }) + const appliedAt = new Date(1700005000000) + await first.emit(entry(2, { state: 'approved', appliedAt })) + expect(latest.status).toBe('ready') + + const second = makeOverseer() + linkActionLog(second.overseer, 'ws-resume') + await view.render() + expect(second.ops).toEqual(['subscribe', 'listPending']) + expect(second.subscribeCalls).toEqual([[expect.anything(), appliedAt]]) + expect(actionLogResumed(second.overseer)).toBe(true) + + // The gap replays as entries ahead of the pages; a replayed resolution beats the page copy. + await second.emit(entry(1, { state: 'rejected', appliedAt: new Date(1700006000000) })) + await second.resolveSubscription() + await second.resolvePendingQuery({ entries: [entry(1)] }) + expect(latest.status).toBe('ready') + expect(latest.pending).toEqual([]) + }) + + it('does not resume from an unsettled session', async () => { + const first = makeOverseer() + linkActionLog(first.overseer, 'ws-unsettled') + await view.render() + await first.resolveSubscription() + await first.emit(entry(1)) + // The pending page never resolves — the session never settles. + + const second = makeOverseer() + linkActionLog(second.overseer, 'ws-unsettled') + await view.render() + expect(second.ops).toEqual(['subscribe', 'listPending']) + expect(second.subscribeCalls).toEqual([[expect.anything()]]) + expect(latest.pending).toEqual([]) + }) + + it('does not park a watermark while the subscribe call is still in flight', async () => { + const first = makeOverseer() + linkActionLog(first.overseer, 'ws-inflight') + await view.render() + await first.resolvePendingQuery({ entries: [entry(1)] }) + expect(latest.status).toBe('ready') // pages drained, but the replay may be undelivered + + const second = makeOverseer() + linkActionLog(second.overseer, 'ws-inflight') + await view.render() + expect(second.subscribeCalls).toEqual([[expect.anything()]]) + }) + + it('does not resume from an errored session', async () => { + vi.spyOn(console, 'error').mockImplementation(() => {}) + const first = makeOverseer() + linkActionLog(first.overseer, 'ws-errored') + await view.render() + await first.emit(entry(1)) + await first.rejectSubscription(new Error('DO overloaded')) + expect(latest.status).toBe('error') + + const second = makeOverseer() + linkActionLog(second.overseer, 'ws-errored') + await view.render() + expect(second.ops).toEqual(['subscribe', 'listPending']) + expect(second.subscribeCalls).toEqual([[expect.anything()]]) + }) + + it('resumes when the same linked stub is released and reacquired', async () => { + const server = makeOverseer() + linkActionLog(server.overseer, 'ws-reacquire') + await view.render() + await server.resolveSubscription() + await server.resolvePendingQuery({ entries: [entry(1)] }) + expect(latest.status).toBe('ready') + + view.unmount() + + // The paged (not just emitted) record's createdAt set the watermark. + await view.render() + expect(server.subscribeCalls[1]).toEqual([expect.anything(), entry(1).createdAt]) + }) + it('reports error but keeps gathered pendings when the subscribe call fails', async () => { vi.spyOn(console, 'error').mockImplementation(() => {}) const server = makeOverseer() diff --git a/packages/workshop-frontend/src/useActions.ts b/packages/workshop-frontend/src/useActions.ts index 5ed9b4a58..67b66a7ad 100644 --- a/packages/workshop-frontend/src/useActions.ts +++ b/packages/workshop-frontend/src/useActions.ts @@ -1,6 +1,6 @@ import { useCallback, useEffect, useRef, useSyncExternalStore } from 'react' import { RpcStub, RpcTarget } from 'capnweb' -import { ActionLogEntry, ActionsSubscriber, Overseer } from '@gadgets/workshop-shared/api' +import { ActionLogEntry, ActionsSubscriber, Overseer, actionChangeTime } from '@gadgets/workshop-shared/api' // One ref-counted store per Overseer stub, shared across consumers. On open the store initiates // the live subscription first, then pages the currently-pending set via @@ -9,6 +9,11 @@ import { ActionLogEntry, ActionsSubscriber, Overseer } from '@gadgets/workshop-s // everything that changes after arrives on the subscription. Pages fold with live-wins // semantics; the last page loading is the "settled" signal. Resolved history is demand-paged // separately (see useActionHistory). +// +// Stubs registered through linkActionLog() additionally park a resume watermark: a settled store +// records its last change time by workspace key on close, and the next store with the same key +// subscribes with startAfter — the server replays the gap (inclusive, as upserts) through the +// subscription, so per-record consumers are patched without refetching. export type ActionsState = { /** @@ -34,6 +39,11 @@ type Store = { subscription: RpcStub<{}> | null generation: number notifyScheduled: boolean + // Max change time (actionChangeTime, the server's index key) received this session, live or + // paged. Every change a settled session missed has change time ≥ its disconnect time ≥ this + // max, so an inclusive startAfter replay from it covers the gap. + lastChanged: Date | undefined + resumed: boolean } const EMPTY_STATE: ActionsState = { @@ -43,6 +53,29 @@ const EMPTY_STATE: ActionsState = { const stores = new WeakMap, Store>() +const storeKeys = new WeakMap, string>() +// Never deleted: a failed later session leaves the last good watermark in place, and resuming +// from an older watermark just replays more. +const watermarks = new Map() + +/** + * Give the stub a stable workspace identity so its store parks a resume watermark on close and + * a later stub with the same key replays the gap. Unlinked stubs always open blind. + */ +export function linkActionLog(overseer: RpcStub, key: string): void { + storeKeys.set(overseer, key) +} + +/** + * Whether the stub's store subscribed with startAfter — its gap is replayed as entries. + * Consumers may trust this without a failure path: a resumed session that later errors surfaces + * as store status 'error' and never parks a watermark, so the next stub swap replays its entire + * gap from the last good one. + */ +export function actionLogResumed(overseer: RpcStub | null): boolean { + return (overseer && stores.get(overseer)?.resumed) ?? false +} + function getStore(overseer: RpcStub): Store { let store = stores.get(overseer) if (!store) { @@ -56,6 +89,8 @@ function getStore(overseer: RpcStub): Store { subscription: null, generation: 0, notifyScheduled: false, + lastChanged: undefined, + resumed: false, } stores.set(overseer, store) } @@ -88,15 +123,26 @@ function resetSession(store: Store): number { store.stagedPending = new Map() store.stagedEntries = new Map() store.snapshot = EMPTY_STATE + store.lastChanged = undefined + store.resumed = false return store.generation } +function trackChange(store: Store, record: ActionLogEntry): void { + const changed = actionChangeTime(record) + if (!store.lastChanged || changed > store.lastChanged) store.lastChanged = changed +} + function openSubscription(overseer: RpcStub, store: Store) { const generation = resetSession(store) + const key = storeKeys.get(overseer) + const startAfter = key === undefined ? undefined : watermarks.get(key) + store.resumed = startAfter !== undefined class ActionsSubscriberImpl extends RpcTarget implements ActionsSubscriber { entry(record: ActionLogEntry): void { if (store.generation !== generation) return + trackChange(store, record) store.stagedEntries.set(record.id, record) let pendingChanged: boolean if (record.state === 'pending') { @@ -120,6 +166,7 @@ function openSubscription(overseer: RpcStub, store: Store) { // Settledness is signalled by the pending page loop draining, not by the subscription. ready(): void {} } + const subscriber = new ActionsSubscriberImpl() as unknown as RpcStub let failed = false const fail = (error: unknown) => { @@ -133,10 +180,12 @@ function openSubscription(overseer: RpcStub, store: Store) { } // Initiated first — the page loop below relies on capnweb e-order having registered the - // subscriber server-side before the first page reads. - overseer.subscribeToActions( - new ActionsSubscriberImpl() as unknown as RpcStub, - ).then(sub => { + // subscriber server-side before the first page reads. With a watermark the server also replays + // the gap (everything changed at/after it, as upserts) ahead of the pages. + const subscribed = startAfter + ? overseer.subscribeToActions(subscriber, startAfter) + : overseer.subscribeToActions(subscriber) + subscribed.then(sub => { if (store.generation !== generation) { sub[Symbol.dispose]() return @@ -155,6 +204,7 @@ function openSubscription(overseer: RpcStub, store: Store) { const page = await overseer.listActions({ filter: 'pending', beforeId }) if (store.generation !== generation) return for (const record of page.entries) { + trackChange(store, record) if (!store.stagedEntries.has(record.id)) store.stagedPending.set(record.id, record) } beforeId = page.nextBeforeId @@ -167,7 +217,16 @@ function openSubscription(overseer: RpcStub, store: Store) { })().catch(fail) } -function closeSubscription(store: Store) { +function closeSubscription(overseer: RpcStub, store: Store) { + const key = storeKeys.get(overseer) + // Only a cleanly settled session sets the watermark: pages drained AND subscribe resolved. The + // second condition is load-bearing — a page-only watermark is poison, because a pending + // record's createdAt can exceed a resolution the dead live stream never delivered, hiding it + // from every future replay. + if (key !== undefined && store.snapshot.status === 'ready' && store.subscription !== null && + store.lastChanged) { + watermarks.set(key, store.lastChanged) + } resetSession(store) store.subscription?.[Symbol.dispose]() store.subscription = null @@ -187,7 +246,7 @@ function release(overseer: RpcStub) { if (!store) return store.refCount-- if (store.refCount <= 0) { - closeSubscription(store) + closeSubscription(overseer, store) stores.delete(overseer) } } diff --git a/packages/workshop-frontend/src/useWorkspaceOpen.ts b/packages/workshop-frontend/src/useWorkspaceOpen.ts index 842fba8a5..79c00555b 100644 --- a/packages/workshop-frontend/src/useWorkspaceOpen.ts +++ b/packages/workshop-frontend/src/useWorkspaceOpen.ts @@ -9,6 +9,7 @@ import type { Overseer, } from '@gadgets/workshop-shared/api' import { reportIssue } from './errorReporting' +import { linkActionLog } from './useActions' import { useDocumentTitle } from './useDocumentTitle' import { classifyWorkspaceOpenFailure, @@ -116,6 +117,7 @@ export function useWorkspaceOpen({ configureObservers = new RpcStub(configureObserversTarget) overseerStub = authenticatedApi.openGadget(id, shareKey, configureObservers) + linkActionLog(overseerStub, id) setOverseer({ stub: overseerStub }) const resolvedSubscription = await overseerStub.subscribeToMetadata((nextMetadata) => { diff --git a/packages/workshop-shared/src/api.ts b/packages/workshop-shared/src/api.ts index 0d86e0bc7..1fdb84adf 100644 --- a/packages/workshop-shared/src/api.ts +++ b/packages/workshop-shared/src/api.ts @@ -2503,6 +2503,15 @@ export function matchesActionHistoryFilter( : filter === "all" || record.type === filter; } +/** + * A record's last state-change time: appliedAt once a mutation has stamped it, else createdAt. + * The server's byLastChanged resume index keys on this (actionLastChangedKey in overseer.ts) and + * the client's resume watermark must reproduce it exactly — derive it only through this helper. + */ +export function actionChangeTime(record: Pick): Date { + return record.appliedAt ?? record.createdAt; +} + /** One page of action history from listActions(). */ export type ActionHistoryPage = { /** Matching records, descending id (creation order, newest first). */