Resume the action stream across reconnects

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

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 (

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<typeof makeOverseer>,
+  getChatMessage = vi.fn<(chatId: number, sequence: number) => Promise<AiChatMessage | null>>(),
+) {
+  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<Overseer>) {
+  return testRoot.render(
+    <ChatInterface
+      workspaceId="workspace"
+      overseer={overseer}
+      selectedChatId={null}
+      onNavigateToChat={() => {}}
+      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()
+  })
+})

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;

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<Record<string, unknown>> = {}): 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()

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])