fix selective timeline compatibility races

packages/app/src/contexts/session-context.tsx

@@ -21,6 +21,7 @@ import {
 } from "@/timeline/session-stream-reducers";
 import { useCreateFlowStore } from "@/stores/create-flow-store";
 import { isTimelineCatchUpComplete } from "@/timeline/timeline-sync-plan";
+import { fetchAgentTimelineOnce } from "@/timeline/fetch-agent-timeline-once";
 import { createViewedTimelineSync, type ViewedTimelineSync } from "@/timeline/viewed-timeline-sync";
 import type { AgentAttachment, SessionOutboundMessage } from "@getpaseo/protocol/messages";
 import { parseServerInfoStatusPayload } from "@getpaseo/protocol/messages";
@@ -1071,7 +1072,8 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
         const initKey = getInitKey(serverId, agentId);
         if (session?.agentAuthoritativeHistoryApplied.get(agentId) !== true) {
           if (!getInitDeferred(initKey)) {
-            createInitDeferred(initKey, request.direction ?? "tail");
+            const deferred = createInitDeferred(initKey, request.direction ?? "tail");
+            void deferred.promise.catch(() => undefined);
           }
           refreshAgentInitializationTimeout({
             key: initKey,
@@ -1081,7 +1083,7 @@ function SessionProviderInternal({ children, serverId, client }: SessionProvider
           setAgentInitializing(agentId, true);
         }
         try {
-          const page = await client.fetchAgentTimeline(agentId, request);
+          const page = await fetchAgentTimelineOnce(client, agentId, request);
           if (getInitDeferred(initKey)) {
             refreshAgentInitializationTimeout({ key: initKey, agentId, setAgentInitializing });
           }

packages/app/src/hooks/use-agent-initialization.ts

@@ -10,6 +10,7 @@ import {
   rejectInitDeferred,
   refreshInitTimeout,
 } from "@/utils/agent-initialization";
+import { fetchAgentTimelineOnce } from "@/timeline/fetch-agent-timeline-once";
 import { planInitialAgentTimelineSync, planTimelineTailFetch } from "@/timeline/timeline-sync-plan";
 import { i18n } from "@/i18n/i18next";
 
@@ -68,7 +69,7 @@ export function ensureAgentIsInitialized(input: EnsureAgentIsInitializedInput):
     return deferred.promise;
   }
 
-  client.fetchAgentTimeline(agentId, timelineRequest).catch((error) => {
+  fetchAgentTimelineOnce(client, agentId, timelineRequest).catch((error) => {
     setAgentInitializing(agentId, false);
     rejectInitDeferred(key, error instanceof Error ? error : new Error(String(error)));
   });

packages/app/src/runtime/host-runtime.test.ts

@@ -2153,6 +2153,77 @@ describe("HostRuntimeStore", () => {
     useSessionStore.getState().clearSession(host.serverId);
   });
 
+  it("uses legacy GitHub attachments when draining a queue for an old daemon", async () => {
+    const host = makeHost({ serverId: "srv_legacy_queue_attachment" });
+    const fakeClient = new FakeDaemonClient();
+    const store = new HostRuntimeStore({
+      deps: {
+        createClient: () => fakeClient as unknown as DaemonClient,
+        connectToDaemon: async () => ({
+          client: fakeClient as unknown as DaemonClient,
+          serverId: host.serverId,
+          hostname: null,
+        }),
+        getClientId: async () => "cid_legacy_queue_attachment",
+      },
+    });
+    const sessionStore = useSessionStore.getState();
+    sessionStore.initializeSession(host.serverId, fakeClient as unknown as DaemonClient, 1);
+    sessionStore.updateSessionServerInfo(host.serverId, {
+      serverId: host.serverId,
+      hostname: null,
+      version: "0.1.105",
+      features: { forgeSearch: false },
+    });
+    sessionStore.setQueuedMessages(
+      host.serverId,
+      new Map([
+        [
+          "agent",
+          [
+            {
+              id: "queued-legacy-attachment",
+              text: "review this",
+              attachments: [
+                {
+                  kind: "github_pr" as const,
+                  item: {
+                    kind: "change_request" as const,
+                    number: 42,
+                    title: "Compatibility fix",
+                    url: "https://github.com/acme/repo/pull/42",
+                    state: "open" as const,
+                    body: "Details",
+                    labels: [],
+                    baseRefName: "main",
+                    headRefName: "fix",
+                  },
+                },
+              ],
+            },
+          ],
+        ],
+      ]),
+    );
+
+    store.drainQueuedAgentMessage(host.serverId, "agent");
+    await fakeClient.waitForSentMessages(1);
+
+    expect(fakeClient.sentAgentMessages[0]?.[2]?.attachments).toEqual([
+      {
+        type: "github_pr",
+        mimeType: "application/github-pr",
+        number: 42,
+        title: "Compatibility fix",
+        url: "https://github.com/acme/repo/pull/42",
+        body: "Details",
+        baseRefName: "main",
+        headRefName: "fix",
+      },
+    ]);
+    sessionStore.clearSession(host.serverId);
+  });
+
   it("applies buffered stale side effects from the accepted page agent", async () => {
     const host = makeHost({ serverId: "srv_buffered_stale_side_effects" });
     const fakeClient = new FakeDaemonClient();

packages/app/src/runtime/host-runtime.ts

@@ -62,7 +62,10 @@ import {
 import { mountBrowserAutomationDaemonClientHandler } from "@/browser-automation/handler";
 import { schedulesQueryBaseKey } from "@/schedules/aggregated-schedules";
 import { sendQueuedComposerMessageNow } from "@/composer/actions";
-import { splitComposerAttachmentsForSubmit } from "@/composer/attachments/submit";
+import {
+  resolveComposerAttachmentSubmitFormat,
+  splitComposerAttachmentsForSubmit,
+} from "@/composer/attachments/submit";
 import { encodeImages } from "@/utils/encode-images";
 
 export type HostRuntimeConnectionStatus = "idle" | "connecting" | "online" | "offline" | "error";
@@ -2141,7 +2144,11 @@ export class HostRuntimeStore {
         write: (update) => useSessionStore.getState().setQueuedMessages(serverId, update),
       },
       submitMessage: async ({ text, attachments }) => {
-        const wirePayload = splitComposerAttachmentsForSubmit(attachments);
+        const supportsForgeAttachments =
+          useSessionStore.getState().sessions[serverId]?.serverInfo?.features?.forgeSearch === true;
+        const wirePayload = splitComposerAttachmentsForSubmit(attachments, {
+          format: resolveComposerAttachmentSubmitFormat({ supportsForgeAttachments }),
+        });
         const images = await encodeImages(wirePayload.images);
         await client.sendAgentMessage(agentId, text, {
           messageId: next.id,

packages/app/src/timeline/fetch-agent-timeline-once.test.ts

@@ -0,0 +1,30 @@
+import { expect, test } from "vitest";
+import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
+import { fetchAgentTimelineOnce } from "./fetch-agent-timeline-once";
+
+type TimelinePage = Awaited<ReturnType<DaemonClient["fetchAgentTimeline"]>>;
+
+test("concurrent identical timeline reads share one request", async () => {
+  let resolvePage: (page: TimelinePage) => void = () => {};
+  const page = new Promise<TimelinePage>((resolve) => {
+    resolvePage = resolve;
+  });
+  const requests: Array<{ agentId: string; direction: string }> = [];
+  const client = {
+    fetchAgentTimeline: async (
+      agentId: string,
+      request: Parameters<DaemonClient["fetchAgentTimeline"]>[1],
+    ) => {
+      requests.push({ agentId, direction: request?.direction ?? "tail" });
+      return page;
+    },
+  };
+  const request = { direction: "tail" as const, limit: 100, projection: "projected" as const };
+
+  const first = fetchAgentTimelineOnce(client, "agent", request);
+  const second = fetchAgentTimelineOnce(client, "agent", request);
+  resolvePage({ hasNewer: false } as TimelinePage);
+
+  await expect(Promise.all([first, second])).resolves.toHaveLength(2);
+  expect(requests).toEqual([{ agentId: "agent", direction: "tail" }]);
+});

packages/app/src/timeline/fetch-agent-timeline-once.ts

@@ -0,0 +1,31 @@
+import type { DaemonClient } from "@getpaseo/client/internal/daemon-client";
+
+type TimelineClient = Pick<DaemonClient, "fetchAgentTimeline">;
+type TimelineRequest = Parameters<TimelineClient["fetchAgentTimeline"]>[1];
+type TimelinePage = Awaited<ReturnType<TimelineClient["fetchAgentTimeline"]>>;
+
+const inFlightByClient = new WeakMap<object, Map<string, Promise<TimelinePage>>>();
+
+export function fetchAgentTimelineOnce(
+  client: TimelineClient,
+  agentId: string,
+  request: TimelineRequest,
+): Promise<TimelinePage> {
+  let inFlight = inFlightByClient.get(client);
+  if (!inFlight) {
+    inFlight = new Map();
+    inFlightByClient.set(client, inFlight);
+  }
+
+  const key = `${agentId}:${JSON.stringify(request)}`;
+  const existing = inFlight.get(key);
+  if (existing) return existing;
+
+  const fetch = client.fetchAgentTimeline(agentId, request);
+  inFlight.set(key, fetch);
+  const clear = () => {
+    if (inFlight.get(key) === fetch) inFlight.delete(key);
+  };
+  void fetch.then(clear, clear);
+  return fetch;
+}

packages/app/src/timeline/viewed-timeline-sync.test.ts

@@ -345,6 +345,24 @@ test("gap recovery supersedes completed catch-up and pages through the current t
   ]);
 });
 
+test("repeated recovery for the same running gap reuses the in-flight fetch", async () => {
+  const world = new TimelineWorld();
+  world.sync.setConnected(true);
+  world.sync.replaceVisibleAgentIds("workspace", ["agent-a"]);
+  const membership = await world.nextMembership();
+  membership.succeed();
+  const initial = await world.nextFetch("agent-a");
+  initial.respond({ hasNewer: false });
+
+  const cursor = { epoch: "epoch-agent-a", endSeq: 10 };
+  world.sync.recoverGap("agent-a", cursor);
+  const gapPage = await world.nextFetch("agent-a");
+  world.sync.recoverGap("agent-a", cursor);
+
+  world.expectNoPendingFetch();
+  gapPage.respond({ hasNewer: false });
+});
+
 test("membership failure autonomously retries without another visibility declaration", async () => {
   const world = new TimelineWorld();
   world.sync.setConnected(true);

packages/app/src/timeline/viewed-timeline-sync.ts

@@ -39,9 +39,34 @@ type CatchUpStatus = "running" | "complete" | "error";
 interface CatchUpState {
   generation: number;
   status: CatchUpStatus;
+  request?: ProjectedTimelineForwardFetchPlan;
   cancelRetry?: () => void;
 }
 
+function isSameCatchUpRequest(
+  left: ProjectedTimelineForwardFetchPlan | undefined,
+  right: ProjectedTimelineForwardFetchPlan | undefined,
+): boolean {
+  if (!left || !right || left.direction !== right.direction) return false;
+  if (left.direction !== "after" || right.direction !== "after") return true;
+  return left.cursor.epoch === right.cursor.epoch && left.cursor.seq === right.cursor.seq;
+}
+
+function shouldKeepCurrentCatchUp(input: {
+  current: CatchUpState | undefined;
+  request: ProjectedTimelineForwardFetchPlan | undefined;
+  supersede: boolean;
+}): boolean {
+  if (!input.current) return false;
+  if (input.supersede) {
+    return (
+      input.current.status === "running" &&
+      isSameCatchUpRequest(input.current.request, input.request)
+    );
+  }
+  return input.current.status === "running" || input.current.status === "complete";
+}
+
 function normalizeAgentIds(agentIds: string[]): string[] {
   return [...new Set(agentIds)].filter(Boolean).sort();
 }
@@ -138,13 +163,13 @@ export function createViewedTimelineSync(ports: ViewedTimelineSyncPorts): Viewed
       return;
     }
     const current = catchUps.get(agentId);
-    if (!supersede && (current?.status === "running" || current?.status === "complete")) {
+    if (shouldKeepCurrentCatchUp({ current, request, supersede })) {
       return;
     }
     current?.cancelRetry?.();
     const generation = (catchUpGenerations.get(agentId) ?? 0) + 1;
     catchUpGenerations.set(agentId, generation);
-    catchUps.set(agentId, { generation, status: "running" });
+    catchUps.set(agentId, { generation, status: "running", request });
     pendingGaps.delete(agentId);
     const cursor = ports.readCursor(agentId);
     const nextRequest =

packages/app/src/utils/agent-directory-sync.test.ts

@@ -65,6 +65,23 @@ function permission(id: string): AgentPermissionRequest {
 }
 
 describe("replaceFetchedAgentDirectory", () => {
+  it("preserves timeline initialization while replacing directory state", () => {
+    const serverId = "server-initializing";
+    const store = useSessionStore.getState();
+    store.initializeSession(serverId, null as unknown as DaemonClient);
+    store.setInitializingAgents(serverId, new Map([["agent", true]]));
+
+    replaceFetchedAgentDirectory({
+      serverId,
+      entries: [createEntry(createAgentPayload({ id: "agent" }))],
+    });
+
+    expect(useSessionStore.getState().sessions[serverId]?.initializingAgents.get("agent")).toBe(
+      true,
+    );
+    store.clearSession(serverId);
+  });
+
   it("re-derives parentAgentId every time an agent snapshot is ingested", () => {
     const serverId = "server-1";
     const store = useSessionStore.getState();

packages/app/src/utils/agent-directory-sync.ts

@@ -180,7 +180,6 @@ export function replaceFetchedAgentDirectory(input: {
   store.setAgentLastActivityBatch(lastActivityByAgentId);
 
   store.setPendingPermissions(input.serverId, new Map(pendingPermissions));
-  store.setInitializingAgents(input.serverId, new Map());
   store.setHasHydratedAgents(input.serverId, true);
   return { agents: fetchedAgents };
 }

packages/server/src/server/selective-timeline-delivery.e2e.test.ts

@@ -137,6 +137,20 @@ async function connect(input: { clientId: string; selective: boolean }): Promise
   return connected;
 }
 
+test("subscription acknowledgements stay on the requesting socket of a retained session", async () => {
+  const legacy = await connect({ clientId: "shared-client", selective: false });
+  const capable = await connect({ clientId: "shared-client", selective: true });
+  legacy.clear();
+  capable.clear();
+
+  await capable.client.setAgentTimelineSubscription(["agent-a"]);
+  await capable.barrier("targeted-subscription-ack");
+
+  expect(
+    legacy.messages.some((message) => message.type === "agent.timeline.set_subscription.response"),
+  ).toBe(false);
+});
+
 test("real WebSocket sessions enforce selective delivery, retained resets, downgrade, and dedicated attention", async () => {
   const legacy = await connect({ clientId: "legacy-client", selective: false });
   let capable = await connect({ clientId: "capable-client", selective: true });

packages/server/src/server/session.test.ts

@@ -311,6 +311,7 @@ interface SessionForTestOptions {
   daemonRuntimeConfig?: SessionOptions["daemonRuntimeConfig"];
   downloadTokenStore?: SessionOptions["downloadTokenStore"];
   messages?: unknown[];
+  targetedMessages?: Array<{ source: object; message: SessionOutboundMessage }>;
   binaryMessages?: Uint8Array[];
 }
 
@@ -348,6 +349,12 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session {
   return new Session({
     clientId: "test-client",
     onMessage: (message) => messages.push(message),
+    ...(options.targetedMessages
+      ? {
+          onMessageToSource: (source: object, message: SessionOutboundMessage) =>
+            options.targetedMessages?.push({ source, message }),
+        }
+      : {}),
     onBinaryMessage: createBinaryMessageHandler(options.binaryMessages),
     logger,
     downloadTokenStore: options.downloadTokenStore ?? asDownloadTokenStore(),
@@ -4578,6 +4585,37 @@ test("replaces a capable session's complete viewed timeline set", async () => {
   ]);
 });
 
+test("acknowledges a timeline subscription only to its socket source", async () => {
+  const messages: SessionOutboundMessage[] = [];
+  const targetedMessages: Array<{ source: object; message: SessionOutboundMessage }> = [];
+  const session = createSessionForTest({ messages, targetedMessages });
+  const capableSocket = {};
+  session.updateClientCapabilities({ selective_agent_timeline: true }, capableSocket);
+
+  await session.handleMessage(
+    {
+      type: "agent.timeline.set_subscription.request",
+      agentIds: ["agent-a"],
+      requestId: "timeline-subscription-targeted",
+    },
+    capableSocket,
+  );
+
+  expect(messages).toEqual([]);
+  expect(targetedMessages).toEqual([
+    {
+      source: capableSocket,
+      message: {
+        type: "agent.timeline.set_subscription.response",
+        payload: {
+          agentIds: ["agent-a"],
+          requestId: "timeline-subscription-targeted",
+        },
+      },
+    },
+  ]);
+});
+
 test("unions viewed timelines across socket sources and removes detached sources", async () => {
   const messages: SessionOutboundMessage[] = [];
   const agentEventListeners: Array<(event: AgentManagerEvent) => void> = [];

packages/server/src/server/session.ts

@@ -429,6 +429,7 @@ export interface SessionOptions {
   appVersion?: string | null;
   clientCapabilities?: Record<string, unknown> | null;
   onMessage: (msg: SessionOutboundMessage) => void;
+  onMessageToSource?: (source: object, msg: SessionOutboundMessage) => void;
   onBinaryMessage?: (frame: Uint8Array) => void;
   getTransportBufferedAmount?: () => number | null;
   onLifecycleIntent?: (intent: SessionLifecycleIntent) => void;
@@ -558,6 +559,9 @@ export class Session {
   private clientCapabilities: ReadonlySet<ClientCapability>;
   private readonly sessionId: string;
   private readonly onMessage: (msg: SessionOutboundMessage) => void;
+  private readonly onMessageToSource:
+    | ((source: object, msg: SessionOutboundMessage) => void)
+    | null;
   private readonly onBinaryMessage: ((frame: Uint8Array) => void) | null;
   private readonly getTransportBufferedAmount: () => number | null;
   private readonly onLifecycleIntent: ((intent: SessionLifecycleIntent) => void) | null;
@@ -629,6 +633,7 @@ export class Session {
       appVersion,
       clientCapabilities,
       onMessage,
+      onMessageToSource,
       onBinaryMessage,
       getTransportBufferedAmount,
       onLifecycleIntent,
@@ -679,6 +684,7 @@ export class Session {
     this.clientCapabilities = parseClientCapabilities(clientCapabilities);
     this.sessionId = uuidv4();
     this.onMessage = onMessage;
+    this.onMessageToSource = onMessageToSource ?? null;
     this.onBinaryMessage = onBinaryMessage ?? null;
     this.getTransportBufferedAmount = getTransportBufferedAmount ?? (() => 0);
     this.onLifecycleIntent = onLifecycleIntent ?? null;
@@ -1618,10 +1624,12 @@ export class Session {
         ) {
           this.replaceAgentTimelineSubscription(source, agentIds);
         }
-        this.emit({
+        const response: SessionOutboundMessage = {
           type: "agent.timeline.set_subscription.response",
           payload: { agentIds, requestId: msg.requestId },
-        });
+        };
+        if (source && this.onMessageToSource) this.onMessageToSource(source, response);
+        else this.emit(response);
         return undefined;
       }
       case "agent.fork_context.request":

packages/server/src/server/websocket-server.ts

@@ -988,6 +988,12 @@ export class VoiceAssistantWebSocketServer {
         }
         this.sendToConnection(connection, wrapSessionMessage(msg));
       },
+      onMessageToSource: (source, msg) => {
+        if (!connection || !connection.sockets.has(source as WebSocketLike)) {
+          return;
+        }
+        this.sendToClient(source as WebSocketLike, wrapSessionMessage(msg));
+      },
       onBinaryMessage: (frame) => {
         if (!connection) {
           return;