fix(app): automatically send queued messages

packages/app/e2e/agent-message-submission.spec.ts

@@ -0,0 +1,50 @@
+import { expect, test } from "./fixtures";
+import {
+  expectQueuedMessageButton,
+  fillComposerDraft,
+  sendDraftToQueue,
+  startRunningMockAgent,
+} from "./helpers/composer";
+import { installDaemonWebSocketGate } from "./helpers/daemon-websocket-gate";
+
+test.describe("Agent message submission", () => {
+  test("automatically sends the next queued message after an authoritative turn completion", async ({
+    page,
+  }) => {
+    test.setTimeout(120_000);
+    const gate = await installDaemonWebSocketGate(page);
+    const agent = await startRunningMockAgent(page, {
+      prefix: "queued-auto-drain-",
+      model: "ten-second-stream",
+      prompt: "Run the first turn before the queued follow-up.",
+    });
+    const queuedPrompt = "Automatically send this queued follow-up.";
+
+    try {
+      gate.holdNextStoppedAgentUpdate(agent.agentId);
+      await fillComposerDraft(page, queuedPrompt);
+      await sendDraftToQueue(page);
+      await expectQueuedMessageButton(page);
+
+      await gate.waitForHeldStoppedAgentUpdate();
+      await expect(page.getByRole("button", { name: /stop|cancel/i })).toHaveCount(0, {
+        timeout: 15_000,
+      });
+      await expectQueuedMessageButton(page);
+      const queuedUserMessage = page.getByTestId("user-message").filter({ hasText: queuedPrompt });
+      await expect(queuedUserMessage).toHaveCount(0);
+
+      gate.releaseHeldStoppedAgentUpdate();
+
+      await expect(queuedUserMessage).toBeVisible({ timeout: 15_000 });
+      await expect(page.getByRole("button", { name: "Send queued message now" })).toHaveCount(0);
+      await expect(page.getByRole("button", { name: /stop|cancel/i }).first()).toBeVisible({
+        timeout: 15_000,
+      });
+      await expect(page.getByTestId("user-message")).toHaveCount(2, { timeout: 15_000 });
+    } finally {
+      gate.releaseHeldStoppedAgentUpdate();
+      await agent.cleanup();
+    }
+  });
+});

packages/app/e2e/helpers/composer.ts

@@ -174,6 +174,7 @@ export async function selectGithubOption(
 }
 
 export interface MockAgentSetup {
+  agentId: string;
   client: SeedDaemonClient;
   repo: Awaited<ReturnType<typeof createTempGitRepo>>;
   cleanup: () => Promise<void>;
@@ -209,6 +210,7 @@ export async function startRunningMockAgent(
     timeout: 30_000,
   });
   return {
+    agentId: agent.id,
     client,
     repo,
     cleanup: async () => {

packages/app/e2e/helpers/daemon-websocket-gate.ts

@@ -23,6 +23,11 @@ interface ClientRequest {
 interface ServerMessage {
   type?: unknown;
   payload?: {
+    kind?: unknown;
+    agent?: {
+      id?: unknown;
+      status?: unknown;
+    };
     initial?: {
       path?: unknown;
     };
@@ -65,6 +70,15 @@ function directoryForRequest(request: ClientRequest): keyof DirectoryBootstrapCo
   return null;
 }
 
+function isStoppedAgentUpdate(message: ServerMessage | null, agentId: string | null): boolean {
+  return (
+    message?.type === "agent_update" &&
+    message.payload?.kind === "upsert" &&
+    message.payload.agent?.id === agentId &&
+    message.payload.agent.status !== "running"
+  );
+}
+
 export async function installDaemonWebSocketGate(page: Page) {
   let acceptingConnections = true;
   const activeSockets = new Set<WebSocketRoute>();
@@ -84,6 +98,23 @@ export async function installDaemonWebSocketGate(page: Page) {
   let heldReadyFileUpdate: (() => void) | null = null;
   let resolveHeldReadyFileUpdate: (() => void) | null = null;
   let heldReadyFileUpdatePromise = Promise.resolve();
+  let stoppedAgentUpdateIdToHold: string | null = null;
+  let heldStoppedAgentUpdate: (() => void) | null = null;
+  let resolveHeldStoppedAgentUpdate: (() => void) | null = null;
+  let heldStoppedAgentUpdatePromise = Promise.resolve();
+
+  const recordFileSubscription = (message: ServerMessage | null): void => {
+    if (
+      message?.type !== "fs.file.subscribe.response" ||
+      typeof message.payload?.initial?.path !== "string"
+    ) {
+      return;
+    }
+    const path = message.payload.initial.path;
+    subscribedFilePaths.add(path);
+    fileSubscriptionWaiters.get(path)?.();
+    fileSubscriptionWaiters.delete(path);
+  };
 
   await page.routeWebSocket(daemonWsRoutePattern(), (ws) => {
     if (!acceptingConnections) {
@@ -126,14 +157,14 @@ export async function installDaemonWebSocketGate(page: Page) {
       if (!acceptingConnections) return;
       const serverMessage = readServerMessage(message);
       if (
-        serverMessage?.type === "fs.file.subscribe.response" &&
-        typeof serverMessage.payload?.initial?.path === "string"
+        isStoppedAgentUpdate(serverMessage, stoppedAgentUpdateIdToHold) &&
+        !heldStoppedAgentUpdate
       ) {
-        const path = serverMessage.payload.initial.path;
-        subscribedFilePaths.add(path);
-        fileSubscriptionWaiters.get(path)?.();
-        fileSubscriptionWaiters.delete(path);
+        heldStoppedAgentUpdate = () => ws.send(message);
+        resolveHeldStoppedAgentUpdate?.();
+        return;
       }
+      recordFileSubscription(serverMessage);
       if (
         serverMessage?.type === "fs.file.update" &&
         serverMessage.payload?.version?.status === "ready" &&
@@ -191,6 +222,23 @@ export async function installDaemonWebSocketGate(page: Page) {
       resolveHeldReadyFileUpdate = null;
       forward?.();
     },
+    holdNextStoppedAgentUpdate(agentId: string): void {
+      stoppedAgentUpdateIdToHold = agentId;
+      heldStoppedAgentUpdate = null;
+      heldStoppedAgentUpdatePromise = new Promise((resolve) => {
+        resolveHeldStoppedAgentUpdate = resolve;
+      });
+    },
+    waitForHeldStoppedAgentUpdate(): Promise<void> {
+      return heldStoppedAgentUpdatePromise;
+    },
+    releaseHeldStoppedAgentUpdate(): void {
+      const forward = heldStoppedAgentUpdate;
+      heldStoppedAgentUpdate = null;
+      stoppedAgentUpdateIdToHold = null;
+      resolveHeldStoppedAgentUpdate = null;
+      forward?.();
+    },
     async drop(): Promise<void> {
       acceptingConnections = false;
       const sockets = Array.from(activeSockets);

packages/app/src/runtime/directory-sync/agent-replica.test.ts

@@ -2,18 +2,24 @@ import { describe, expect, it } from "vitest";
 import type { DaemonClient, FetchAgentsEntry } from "@getpaseo/client/internal/daemon-client";
 import type { AgentSnapshotPayload } from "@getpaseo/protocol/messages";
 import { useSessionStore } from "@/stores/session-store";
+import { processAgentStreamEvent } from "@/timeline/session-stream-reducers";
 import { AgentDirectoryReplica } from "./agent-replica";
 
-function payload(title: string): AgentSnapshotPayload {
+function payload(input: {
+  title: string;
+  status?: AgentSnapshotPayload["status"];
+  updatedAt?: string;
+  archivedAt?: string | null;
+}): AgentSnapshotPayload {
   return {
     id: "agent",
     provider: "codex",
     cwd: "/repo",
     model: null,
     createdAt: "2026-07-17T00:00:00.000Z",
-    updatedAt: "2026-07-17T00:01:00.000Z",
+    updatedAt: input.updatedAt ?? "2026-07-17T00:01:00.000Z",
     lastUserMessageAt: null,
-    status: "idle",
+    status: input.status ?? "idle",
     capabilities: {
       supportsStreaming: true,
       supportsSessionPersistence: true,
@@ -26,8 +32,9 @@ function payload(title: string): AgentSnapshotPayload {
     availableModes: [],
     pendingPermissions: [],
     persistence: null,
-    title,
+    title: input.title,
     labels: {},
+    ...(input.archivedAt !== undefined ? { archivedAt: input.archivedAt } : {}),
   };
 }
 
@@ -51,30 +58,101 @@ function entry(agent: AgentSnapshotPayload): FetchAgentsEntry {
 }
 
 describe("AgentDirectoryReplica", () => {
+  it("detects an authoritative stop after the timeline optimistically becomes idle", () => {
+    const serverId = "agent-replica-authoritative-transition";
+    const store = useSessionStore.getState();
+    store.initializeSession(serverId, null as unknown as DaemonClient);
+    const stoppedAgentIds: string[] = [];
+    const replica = new AgentDirectoryReplica(serverId, (agentId) => stoppedAgentIds.push(agentId));
+    replica.commitSnapshot([entry(payload({ title: "running", status: "running" }))], []);
+
+    const runningAgent = useSessionStore.getState().sessions[serverId]?.agents.get("agent");
+    if (!runningAgent) throw new Error("Expected running agent after authoritative snapshot");
+    const timelineResult = processAgentStreamEvent({
+      event: { type: "turn_completed", provider: "codex" },
+      seq: undefined,
+      epoch: undefined,
+      currentTail: [],
+      currentHead: [],
+      currentCursor: undefined,
+      currentAgent: {
+        status: runningAgent.status,
+        updatedAt: runningAgent.updatedAt,
+        lastActivityAt: runningAgent.lastActivityAt,
+      },
+      timestamp: new Date("2026-07-17T00:02:00.000Z"),
+    });
+    const timelineAgent = timelineResult.agent;
+    if (!timelineAgent) throw new Error("Expected turn completion to update the agent");
+    expect(timelineAgent.status).toBe("idle");
+    store.setAgents(serverId, (current) => {
+      const next = new Map(current);
+      next.set("agent", { ...runningAgent, ...timelineAgent });
+      return next;
+    });
+
+    replica.applyDelta({
+      kind: "upsert",
+      agent: payload({
+        title: "idle",
+        status: "idle",
+        updatedAt: "2026-07-17T00:03:00.000Z",
+      }),
+      project: entry(payload({ title: "idle" })).project,
+    });
+
+    expect(stoppedAgentIds).toEqual(["agent"]);
+    store.clearSession(serverId);
+  });
+
+  it("does not report an authoritative stop when the agent is archived", () => {
+    const serverId = "agent-replica-archived-transition";
+    const store = useSessionStore.getState();
+    store.initializeSession(serverId, null as unknown as DaemonClient);
+    const stoppedAgentIds: string[] = [];
+    const replica = new AgentDirectoryReplica(serverId, (agentId) => stoppedAgentIds.push(agentId));
+    const running = payload({ title: "running", status: "running" });
+    replica.commitSnapshot([entry(running)], []);
+
+    replica.applyDelta({
+      kind: "upsert",
+      agent: payload({
+        title: "archived",
+        status: "idle",
+        updatedAt: "2026-07-17T00:03:00.000Z",
+        archivedAt: "2026-07-17T00:02:00.000Z",
+      }),
+      project: entry(running).project,
+    });
+
+    expect(stoppedAgentIds).toEqual([]);
+    store.clearSession(serverId);
+  });
+
   it("keeps membership authoritative across remove, stale timeline, and re-add", () => {
     const serverId = "agent-replica";
     const store = useSessionStore.getState();
     store.initializeSession(serverId, null as unknown as DaemonClient);
     const replica = new AgentDirectoryReplica(serverId, () => undefined);
-    replica.commitSnapshot([entry(payload("directory"))], []);
+    replica.commitSnapshot([entry(payload({ title: "directory" }))], []);
     const directoryPlacement = useSessionStore
       .getState()
       .sessions[serverId]?.agents.get("agent")?.projectPlacement;
     expect(directoryPlacement).toBeDefined();
     const staleToken = replica.captureTimeline("agent");
 
     replica.remove("agent");
-    expect(replica.submitTimelineAgent(staleToken, payload("stale"))).toBe(false);
+    expect(replica.submitTimelineAgent(staleToken, payload({ title: "stale" }))).toBe(false);
     expect(useSessionStore.getState().sessions[serverId]?.agents.has("agent")).toBe(false);
 
     replica.applyDelta({
       kind: "upsert",
-      agent: payload("re-added"),
-      project: entry(payload("x")).project,
+      agent: payload({ title: "re-added" }),
+      project: entry(payload({ title: "x" })).project,
     });
-    expect(replica.submitTimelineAgent(staleToken, payload("still stale"))).toBe(false);
+    expect(replica.submitTimelineAgent(staleToken, payload({ title: "still stale" }))).toBe(false);
     const currentToken = replica.captureTimeline("agent");
-    expect(replica.submitTimelineAgent(currentToken, payload("current"))).toBe(true);
+    expect(replica.submitTimelineAgent(currentToken, payload({ title: "current" }))).toBe(true);
     expect(useSessionStore.getState().sessions[serverId]?.agents.get("agent")?.title).toBe(
       "current",
     );