feat(remote): queue messages sent during a running remote turn

docs/remote-access.md

@@ -37,6 +37,8 @@ The **Explorer** sidebar and Go to File use the normal file views for host folde
 
 The composer’s **+** menu supports file and image attachments, Plan mode, and saved drafts when the host advertises these capabilities. Attachments are copied to the host’s private data directory before the turn or draft is recorded; each file is limited to 20 MiB. Image previews are restored from the host when you reopen a conversation. A draft can be sent or removed from its transcript card. Plan mode uses the host provider and produces a reviewable plan card whose Build action continues on the host with the session’s current model. Update older hosts to enable these menu actions.
 
+Messages sent while a turn is running wait in a queue on the host, which sends each one when the turn before it finishes, also while this computer is disconnected. Queued messages can be edited or removed until they are sent. Stopping a turn pauses the queue; **Resume** continues the interrupted work first, then the queue. Host providers cannot steer a running turn. Older hosts reject messages sent during a turn; update the host to queue them.
+
 Features that read or run on this computer are not available in these projects: `@` file mentions, skills and slash commands other than `/plan` and `/compact`, operator mode, and terminals. Worktree deletion and the local worktree settings page are not available remotely yet. Plans from the transcript open normal read-only plan tabs. Source files stay on the host; this feature shares host-owned sessions, not working-directory synchronization.
 
 ## Manual connection (advanced / development)
@@ -100,7 +102,7 @@ The release workflow publishes `monocode-host-{darwin,linux}-{arm64,x64}.tar.gz`
 
 Supported: persistent remote text conversations with all ten local provider adapters, follow-up turns, approvals, questions where the provider offers them, cancellation, per-device revocation, reconnect, remote file browsing and text editing, Git status, file diffs, and history, staging and commits, branch selection and creation, worktree selection and creation, and a tracked Git diff against HEAD. OpenCode's server and event stream stay on the host's loopback interface. Cursor's optional enrichment from its native session database is not available on the headless host; basic transcript and subagent events still work. The desktop polls the host and downloads only transcript blocks that changed since its last update (every 0.75 s while a session runs, 3 s otherwise). The desktop rejects any single host response over 16 MiB. A sync above 4 MiB, such as reopening a very long transcript or one very large tool output, is sent as a series of bounded pieces of one consistent revision. Transcript size is therefore not limited by the response cap. The host writes streamed output in 120 ms batches and keeps a bounded event journal.
 
-Remote history appears in the Sessions sidebar of each project on a machine. Host snapshots also populate the app's normal session state while the tab is open; they are not written to the local session store. Remote `/compact` uses the provider's context compaction, and `/plan` selects the host provider's plan mode. Queued follow-ups and editing the last message still need host commands. Other local slash commands and skill expansion are not yet available remotely. Worktree deletion, terminals, generated image output, named provider accounts, `/operator`, automations, orchestration, LAN discovery, and account-based tunnels are not implemented yet. Other remote prompts are sent directly to the provider.
+Remote history appears in the Sessions sidebar of each project on a machine. Host snapshots also populate the app's normal session state while the tab is open; they are not written to the local session store. Remote `/compact` uses the provider's context compaction, and `/plan` selects the host provider's plan mode. Editing the last message still needs a host command. Other local slash commands and skill expansion are not yet available remotely. Worktree deletion, terminals, generated image output, named provider accounts, `/operator`, automations, orchestration, LAN discovery, and account-based tunnels are not implemented yet. Other remote prompts are sent directly to the provider.
 
 The headless host runs the reused TypeScript adapters with a Node process backend. It proves the execution boundary without introducing the planned Rust daemon/worker IPC yet. Node is included in host release archives, separately from the desktop application.
 

host/engine.test.ts

@@ -698,6 +698,108 @@ describe("headless session ownership", () => {
     expect(provider.answer).toHaveBeenCalledTimes(1);
   });
 
+  it("queues follow-ups during a turn and sends them in order as each turn ends", async () => {
+    const { engine, store, turns, id } = setup();
+    engine.command({ type: "send", commandId: "send", sessionId: id, text: "Work" });
+    await vi.waitFor(() => expect(turns).toHaveLength(1));
+    engine.command({ type: "enqueue", commandId: "q1", sessionId: id, text: "Next", intent: "plan" });
+    engine.command({ type: "enqueue", commandId: "q2", sessionId: id, text: "Then" });
+    expect(store.session(id).session).toMatchObject({
+      queuedMessages: [{ id: "q1", text: "Next" }, { id: "q2", text: "Then" }],
+      queueStatus: "active",
+    });
+    expect(turns).toHaveLength(1);
+
+    turns[0].finish();
+    await vi.waitFor(() => expect(turns).toHaveLength(2));
+    expect(turns[1].input).toMatchObject({ text: "Next", intent: "plan" });
+    expect(store.session(id).session.blocks.at(-1)).toMatchObject({ id: "q1", role: "user" });
+    expect(store.session(id).session.queuedMessages).toEqual([
+      expect.objectContaining({ id: "q2" }),
+    ]);
+
+    turns[1].finish();
+    await vi.waitFor(() => expect(turns).toHaveLength(3));
+    expect(turns[2].input.text).toBe("Then");
+    expect(store.session(id).session.queuedMessages).toBeUndefined();
+    expect(store.session(id).session.queueStatus).toBeUndefined();
+    turns[2].finish();
+  });
+
+  it("sends a follow-up at once when it reaches the host after the turn ended", async () => {
+    const { engine, turns, id } = setup();
+    engine.command({ type: "enqueue", commandId: "late", sessionId: id, text: "Next" });
+    await vi.waitFor(() => expect(turns).toHaveLength(1));
+    expect(turns[0].input.text).toBe("Next");
+    turns[0].finish();
+  });
+
+  it("edits and removes queued follow-ups until they are sent", async () => {
+    const { engine, turns, id } = setup();
+    engine.command({ type: "send", commandId: "send", sessionId: id, text: "Work" });
+    await vi.waitFor(() => expect(turns).toHaveLength(1));
+    engine.command({ type: "enqueue", commandId: "q1", sessionId: id, text: "Next" });
+    engine.command({ type: "enqueue", commandId: "q2", sessionId: id, text: "Then" });
+    engine.command({ type: "editQueued", commandId: "edit", sessionId: id, messageId: "q1", text: "Edited" });
+    expect(() =>
+      engine.command({ type: "editQueued", commandId: "blank", sessionId: id, messageId: "q1", text: " " }),
+    ).toThrow("Invalid prompt");
+    engine.command({ type: "dequeue", commandId: "drop", sessionId: id, messageId: "q2" });
+
+    turns[0].finish();
+    await vi.waitFor(() => expect(turns).toHaveLength(2));
+    expect(turns[1].input.text).toBe("Edited");
+    expect(() =>
+      engine.command({ type: "dequeue", commandId: "late-drop", sessionId: id, messageId: "q1" }),
+    ).toThrow("no longer waiting");
+    turns[1].finish();
+  });
+
+  it("pauses the queue when the turn is stopped and resumes the work before it", async () => {
+    const { engine, store, turns, id } = setup();
+    engine.command({ type: "send", commandId: "send", sessionId: id, text: "Work" });
+    await vi.waitFor(() => expect(turns).toHaveLength(1));
+    engine.command({ type: "enqueue", commandId: "q1", sessionId: id, text: "Next" });
+    expect(() =>
+      engine.command({ type: "resumeQueue", commandId: "early", sessionId: id }),
+    ).toThrow("not paused");
+    engine.command({ type: "cancel", commandId: "stop", sessionId: id, runId: store.session(id).runId });
+    await vi.waitFor(() => expect(store.session(id).status).toBe("idle"));
+    expect(store.session(id).session.queueStatus).toBe("paused");
+    await new Promise((resolve) => setTimeout(resolve, 10));
+    expect(turns).toHaveLength(1);
+
+    engine.command({ type: "resumeQueue", commandId: "resume", sessionId: id });
+    await vi.waitFor(() => expect(turns).toHaveLength(2));
+    expect(turns[1].input.text).toBe("Continue from where you left off.");
+    expect(store.session(id).session.queueStatus).toBe("resuming");
+
+    turns[1].finish();
+    await vi.waitFor(() => expect(turns).toHaveLength(3));
+    expect(turns[2].input.text).toBe("Next");
+    turns[2].finish();
+  });
+
+  it("keeps a draft saved while the queue is paused through the queued turns", async () => {
+    const { engine, store, turns, id } = setup();
+    engine.command({ type: "send", commandId: "send", sessionId: id, text: "Work" });
+    await vi.waitFor(() => expect(turns).toHaveLength(1));
+    engine.command({ type: "enqueue", commandId: "q1", sessionId: id, text: "Next" });
+    engine.command({ type: "cancel", commandId: "stop", sessionId: id, runId: store.session(id).runId });
+    await vi.waitFor(() => expect(store.session(id).status).toBe("idle"));
+    engine.command({ type: "draft", commandId: "draft", sessionId: id, text: "Later" });
+
+    engine.command({ type: "resumeQueue", commandId: "resume", sessionId: id });
+    await vi.waitFor(() => expect(turns).toHaveLength(2));
+    turns[1].finish();
+    await vi.waitFor(() => expect(turns).toHaveLength(3));
+    expect(turns[2].input.text).toBe("Next");
+    expect(store.session(id).session.blocks).toContainEqual(
+      expect.objectContaining({ id: "draft", text: "Later", draft: true }),
+    );
+    turns[2].finish();
+  });
+
   it("recovers interrupted durable state without replaying an uncertain provider send", async () => {
     const { store, provider, id } = setup();
     const value = store.session(id);
@@ -712,6 +814,8 @@ describe("headless session ownership", () => {
             ...value.session,
             busy: true,
             providerSessionId: "retained",
+            queuedMessages: [{ id: "queued", text: "Next", attachments: [] }],
+            queueStatus: "active",
             blocks: [
               {
                 id: "interrupted-turn",
@@ -729,6 +833,8 @@ describe("headless session ownership", () => {
     expect(store.session(id).status).toBe("interrupted");
     expect(store.session(id).session.busy).toBe(false);
     expect(store.session(id).session.blocks[0].durationMs).toBe(2_000);
+    // Queued follow-ups wait for the user to inspect the interrupted turn.
+    expect(store.session(id).session.queueStatus).toBe("paused");
     expect(provider.send).not.toHaveBeenCalled();
     expect(provider.bind).toHaveBeenCalledWith(
       id,
@@ -893,5 +999,11 @@ describe("headless session ownership", () => {
         reply: { kind: "answered", answers: { a: [42] } },
       }),
     ).toThrow();
+    expect(() =>
+      parseCommand({ type: "enqueue", commandId: "x", sessionId: "y", text: " " }),
+    ).toThrow("Invalid prompt");
+    expect(() =>
+      parseCommand({ type: "dequeue", commandId: "x", sessionId: "y" }),
+    ).toThrow("queued message ID");
   });
 });