diff --git a/docs/timeline-sync.md b/docs/timeline-sync.md index efbb0e9252..8264fe2683 100644 --- a/docs/timeline-sync.md +++ b/docs/timeline-sync.md @@ -86,6 +86,13 @@ offline, then Updating messages until authoritative catch-up completes. Socket c certify that the displayed conversation is current. The timeline owner publishes freshness; the view renders it without a toast timer or a separate resume workflow. +The draft-create handoff has the same lifetime: the viewed-timeline owner releases it when the sync +stops owing that chat a catch-up, not when the first authoritative page lands. Releasing it at the +first page leaves a chat this client just created looking hydrated but still catching up, which +shows Updating messages for a conversation that cannot be out of date. Closing the tab ends the +obligation too, so a reopened chat starts from authoritative state instead of a stale optimistic +one; disconnect and backgrounding keep it. + Foregrounding probes a nominally connected session immediately. A healthy response preserves the socket; a failed three-second probe starts reconnecting without waiting for the background heartbeat or retry backoff. This cannot keep a mobile socket alive after the operating system suspends it. diff --git a/packages/app/e2e/browser/agent-message-submission.spec.ts b/packages/app/e2e/browser/agent-message-submission.spec.ts index ee97d87adb..e7f87ed103 100644 --- a/packages/app/e2e/browser/agent-message-submission.spec.ts +++ b/packages/app/e2e/browser/agent-message-submission.spec.ts @@ -38,6 +38,7 @@ import { gotoAppShell, openSettings, selectModel } from "../support/helpers/app" import { observeTimelineSubscriptions } from "../support/helpers/timeline-delivery"; import { rememberTimelineRequestCounts } from "../support/helpers/timeline-resume"; import { + recordPanelToasts, switchWorkspaceViaSidebar, waitForWorkspaceInSidebar, workspaceDeckEntryLocator, @@ -1165,8 +1166,11 @@ test.describe("Agent message submission", () => { draftCreateScenario, }) => { test.setTimeout(120_000); + const toasts = await recordPanelToasts(page); const pending = await beginDraftCreateSubmission(page, draftCreateScenario); await completeDraftCreateSubmission(page, draftCreateScenario, pending); + // A chat this client just created is current by construction; it is never out of date. + await toasts.expectNeverShown("agent-updating-toast"); }); test("restores a rejected submission and accepts its retry", async ({ diff --git a/packages/app/e2e/support/helpers/workspace-ui.ts b/packages/app/e2e/support/helpers/workspace-ui.ts index 9cedfc5c79..fe03ec053c 100644 --- a/packages/app/e2e/support/helpers/workspace-ui.ts +++ b/packages/app/e2e/support/helpers/workspace-ui.ts @@ -111,6 +111,50 @@ export async function expectReconnectingToastGone( }); } +declare global { + interface Window { + __shownPanelToasts?: string[]; + } +} + +export interface PanelToastLog { + /** Test ids of every toast that entered the DOM since recording started, in order. */ + shown(): Promise; + expectNeverShown(testId: string): Promise; +} + +/** + * Record every toast that enters the DOM from the next navigation onward. A toast can appear + * and disappear within a few frames, so a settled `toHaveCount(0)` cannot prove one never showed. + */ +export async function recordPanelToasts(page: Page): Promise { + await page.addInitScript(() => { + const shown: string[] = []; + window.__shownPanelToasts = shown; + const record = (node: Element) => { + if (node.getAttribute("role") !== "alert") return; + shown.push(node.getAttribute("data-testid") ?? ""); + }; + new MutationObserver((mutations) => { + for (const mutation of mutations) { + for (const node of mutation.addedNodes) { + if (!(node instanceof Element)) continue; + record(node); + for (const descendant of node.querySelectorAll('[role="alert"]')) record(descendant); + } + } + }).observe(document, { childList: true, subtree: true }); + }); + + const shown = () => page.evaluate(() => window.__shownPanelToasts ?? []); + return { + shown, + async expectNeverShown(testId: string): Promise { + expect(await shown()).not.toContain(testId); + }, + }; +} + export async function expectHostConnectingOrOffline( page: Page, options?: { timeout?: number }, diff --git a/packages/app/src/stores/workspace-layout-store.test.ts b/packages/app/src/stores/workspace-layout-store.test.ts index b5ff97e9a4..d1a4768cb5 100644 --- a/packages/app/src/stores/workspace-layout-store.test.ts +++ b/packages/app/src/stores/workspace-layout-store.test.ts @@ -134,6 +134,7 @@ it("feeds restored layout to timeline sync as a release signal, not a subscripti readCursor: () => undefined, fetchPage: async () => ({ hasNewer: false, endCursor: null }), fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }), + onCatchUpEnded: () => undefined, reportError: () => undefined, schedule: () => () => undefined, }); diff --git a/packages/app/src/timeline/replica.test.ts b/packages/app/src/timeline/replica.test.ts index f77e95b8c6..fa7521e487 100644 --- a/packages/app/src/timeline/replica.test.ts +++ b/packages/app/src/timeline/replica.test.ts @@ -9,6 +9,7 @@ import { afterEach, describe, expect, it } from "vitest"; import type { AgentStreamEventPayload } from "@getpaseo/protocol/messages"; import type { CachedTimeline } from "@/runtime/replica-cache"; import { selectAgentTimelineState, useSessionStore } from "@/stores/session-store"; +import { useCreateFlowStore, type PendingCreateAttempt } from "@/stores/create-flow-store"; import type { StreamItem } from "@/types/stream"; import { createTimelineReplica, @@ -443,6 +444,119 @@ describe("viewed timeline persistence", () => { }); }); +describe("create handoff lifetime", () => { + const DRAFT_ID = "draft-1"; + + afterEach(() => useCreateFlowStore.getState().clearAll()); + + function sentCreateHandoff(): void { + useCreateFlowStore.getState().setPending({ + draftId: DRAFT_ID, + serverId: SERVER_ID, + agentId: AGENT_ID, + clientMessageId: "client-message-1", + text: "Create this chat.", + timestamp: 0, + }); + useCreateFlowStore.getState().markLifecycle({ draftId: DRAFT_ID, lifecycle: "sent" }); + } + + function createHandoff(): PendingCreateAttempt | undefined { + return useCreateFlowStore.getState().pendingByDraftId[DRAFT_ID]; + } + + function syncedTailPage(): Parameters[0] { + return { + requestId: "tail-1", + agentId: AGENT_ID, + agent: null, + direction: "tail", + projection: "projected", + reset: false, + epoch: "epoch-1", + window: { minSeq: 1, maxSeq: 0, nextSeq: 1 }, + startCursor: null, + endCursor: null, + entries: [], + error: null, + hasNewer: false, + hasOlder: false, + staleCursor: false, + gap: false, + }; + } + + interface HeldCatchUpOwner { + owner: ViewedTimelineOwner; + release: () => void; + } + + async function ownerWithHeldCatchUp(): Promise { + let releaseTailFetch!: () => void; + let tailFetchStarted = false; + const tailFetched = new Promise((resolve) => { + releaseTailFetch = resolve; + }); + const replica = createTimelineReplica({ + serverId: SERVER_ID, + storage: { readTimeline: async () => undefined, commitTimeline: () => undefined }, + prepareAgent: async () => undefined, + }); + const owner = createViewedTimelineOwner({ + serverId: SERVER_ID, + replica, + replaceDemandedAgentIds: () => undefined, + drainQueuedAgentMessage: () => undefined, + ports: { + observe: () => ({ ready: Promise.resolve(), release: async () => undefined }), + readCursor: () => undefined, + fetchPage: async () => { + tailFetchStarted = true; + await tailFetched; + return { hasNewer: false, endCursor: null }; + }, + fetchLatestTail: async () => ({ hasNewer: false, endCursor: null }), + reportError: () => undefined, + schedule: () => () => undefined, + }, + }); + + owner.setConnected(true); + owner.replaceVisibleAgentIds("test", [AGENT_ID]); + await expect.poll(() => tailFetchStarted).toBe(true); + return { owner, release: releaseTailFetch }; + } + + it("keeps the handoff through the first synchronized page and releases it when the chat is current", async () => { + useSessionStore.getState().initializeSession(SERVER_ID, null); + sentCreateHandoff(); + const { owner, release } = await ownerWithHeldCatchUp(); + + owner.applyTimelineResponse(syncedTailPage()); + expect( + selectAgentTimelineState(useSessionStore.getState().sessions[SERVER_ID], AGENT_ID), + ).toMatchObject({ status: "synced" }); + expect(createHandoff()).toMatchObject({ agentId: AGENT_ID, lifecycle: "sent" }); + + release(); + await expect.poll(() => createHandoff()).toBeUndefined(); + owner.dispose(); + }); + + it("releases the handoff when the chat's tab closes before catch-up completes", async () => { + useSessionStore.getState().initializeSession(SERVER_ID, null); + sentCreateHandoff(); + const { owner, release } = await ownerWithHeldCatchUp(); + + owner.replaceVisibleAgentIds("test", []); + owner.replaceOpenTabAgentIds([]); + + expect(createHandoff()).toBeUndefined(); + release(); + owner.dispose(); + }); +}); + function createSqliteCache() { const database = new DatabaseSync(":memory:"); let beforeRead = async () => {}; diff --git a/packages/app/src/timeline/viewed-timeline-sync.test.ts b/packages/app/src/timeline/viewed-timeline-sync.test.ts index e850621cbb..9bf6897bd9 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.test.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.test.ts @@ -6,6 +6,7 @@ import { consumeForcedTimelineTailReplacement, createViewedTimelineSync, type TimelineResponsePayload, + type ViewedTimelineStatus, type ViewedTimelineSyncPorts, } from "./viewed-timeline-sync"; @@ -31,6 +32,11 @@ interface MembershipRequest { fail(message: string): void; } +interface TimelineCurrentReport { + agentId: string; + status: ViewedTimelineStatus; +} + interface TimelineFetch { agentId: string; request: ProjectedTimelineForwardFetchPlan; @@ -126,9 +132,16 @@ class TimelineWorld { readonly cursors = new Map(); readonly cacheRequests: string[] = []; readonly forcedTimelineTailReplacements = new Set(); + readonly reportedCurrent: TimelineCurrentReport[] = []; cacheGate: Deferred | null = null; readonly sync = createViewedTimelineSync({ replaceDemandedAgentIds: () => undefined, + onCatchUpEnded: (agentId) => { + this.reportedCurrent.push({ + agentId, + status: this.sync.getAgentTimelineStatus(agentId), + }); + }, prepare: async (agentId) => { this.cacheRequests.push(agentId); this.cacheRequestWaiters.shift()?.(agentId); @@ -436,6 +449,60 @@ test("a gap absorbed by a running tail is recovered after the tail completes", a await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); }); +test("reports a chat current only once its parked follow-up page completes", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + (await world.nextMembership()).succeed(); + const tail = await world.nextFetch("agent-a"); + + world.sync.recoverGap("agent-a", { epoch: "epoch-agent-a", endSeq: 9 }); + tail.respond({ hasNewer: false }); + + const recovery = await world.nextFetch("agent-a"); + expect(world.reportedCurrent).toEqual([]); + expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("pending"); + + recovery.respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + expect(world.reportedCurrent).toEqual([{ agentId: "agent-a", status: "ready" }]); + world.sync.dispose(); +}); + +test("reports a chat current once the latest-tail fallback replaces an overflowing catch-up", async () => { + const world = new TimelineWorld(); + world.cursors.set("agent-a", { epoch: "epoch-agent-a", endSeq: 42 }); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]); + (await world.nextMembership()).succeed(); + + const probe = await world.nextFetch("agent-a"); + probe.respond({ hasNewer: true, seq: 82 }); + + const fallback = await world.nextFetch("agent-a"); + expect(world.reportedCurrent).toEqual([]); + + fallback.respond({ hasNewer: false }); + await vi.waitFor(() => expect(world.sync.getAgentTimelineStatus("agent-a")).toBe("ready")); + expect(world.reportedCurrent).toEqual([{ agentId: "agent-a", status: "ready" }]); + world.sync.dispose(); +}); + +test("stops owing a chat a catch-up when its tab closes mid-fetch", async () => { + const world = new TimelineWorld(); + world.sync.setConnected(true); + world.sync.replaceVisibleAgentIds("pane", ["agent-a"]); + (await world.nextMembership()).succeed(); + await world.nextFetch("agent-a"); + expect(world.reportedCurrent).toEqual([]); + + world.sync.replaceVisibleAgentIds("pane", []); + world.sync.replaceOpenTabAgentIds([]); + + expect(world.reportedCurrent).toEqual([{ agentId: "agent-a", status: "pending" }]); + world.sync.dispose(); +}); + test("unchanged visible-set publication does not cancel paged catch-up", async () => { const world = new TimelineWorld(); world.sync.setConnected(true); diff --git a/packages/app/src/timeline/viewed-timeline-sync.ts b/packages/app/src/timeline/viewed-timeline-sync.ts index 5f805a829b..5939feabce 100644 --- a/packages/app/src/timeline/viewed-timeline-sync.ts +++ b/packages/app/src/timeline/viewed-timeline-sync.ts @@ -247,9 +247,6 @@ function finalizeProcessedTimeline(input: { clearAgentInitializingFlag(input.serverId, input.agentId); } if (input.synchronized) { - useCreateFlowStore - .getState() - .clearByAgent({ serverId: input.serverId, agentId: input.agentId }); const session = useSessionStore.getState().sessions[input.serverId]; const agent = session?.agents.get(input.agentId) ?? session?.agentDetails.get(input.agentId); if (agent && agent.turn.phase === "idle") input.drainQueuedAgentMessage(input.agentId); @@ -317,6 +314,11 @@ export interface ViewedTimelineSyncPorts { request: ProjectedTimelineForwardFetchPlan, ): Promise; fetchLatestTail(agentId: string): Promise; + /** + * The sync no longer owes this chat a catch-up: it reached current, or it left the + * demanded set. Disconnect and backgrounding keep the obligation and do not report here. + */ + onCatchUpEnded(agentId: string): void; reportError(error: unknown): void; schedule(task: () => void, delayMs: number): () => void; } @@ -346,7 +348,7 @@ export interface ViewedTimelineSync extends ViewedTimelineUiBridge { export type ViewedTimelineOwnerPorts = Omit< ViewedTimelineSyncPorts, - "prepare" | "replaceDemandedAgentIds" + "prepare" | "replaceDemandedAgentIds" | "onCatchUpEnded" >; export interface ViewedTimelineOwner extends ViewedTimelineSync { @@ -376,6 +378,8 @@ export function createViewedTimelineOwner(input: { prepare: (agentId) => input.replica.prepare(agentId), readCursor: (agentId) => input.replica.readCursor(agentId) ?? input.ports.readCursor(agentId), replaceDemandedAgentIds: input.replaceDemandedAgentIds, + onCatchUpEnded: (agentId) => + useCreateFlowStore.getState().clearByAgent({ serverId: input.serverId, agentId }), }); const streamQueue = createSessionAgentStreamReducerQueue({ serverId: input.serverId, @@ -530,6 +534,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed const wasPending = visibilityCatchUpPending.delete(agentId); const hadError = visibilityCatchUpErrors.delete(agentId); const wasRetrying = manualRetries.delete(agentId); + ports.onCatchUpEnded(agentId); if (wasPending || hadError || wasRetrying) notifyListeners(); }; @@ -785,12 +790,14 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed return; } + const released: string[] = []; for (const agentId of desired) { if (!nextDesired.includes(agentId)) { cancelCatchUp(agentId); visibilityCatchUpPending.delete(agentId); visibilityCatchUpErrors.delete(agentId); manualRetries.delete(agentId); + released.push(agentId); } } for (const agentId of nextDesired) { @@ -806,6 +813,7 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed desired = nextDesired; ports.replaceDemandedAgentIds(desired); membershipGeneration += 1; + for (const agentId of released) ports.onCatchUpEnded(agentId); notifyListeners(); void reconcileMembership(); };