fix(server): reject attachment starts closed during conversion

Keep interruption terminal for prompts still being materialized, and reject a start if its session closes before provider submission. Cover both providers with real large-image conversions and make history reconciliation tests await the async timeline capability.

packages/server/src/server/agent/providers/omp/agent.test.ts

@@ -1332,13 +1332,14 @@ describe("OMP agent client and session", () => {
     expect(omp.runtimeLaunches()).toHaveLength(1);
   });
 
-  test("closes a replacement process if the session closes during relaunch", async () => {
+  test("rejects a pending recovery start and closes its processes when the session closes", async () => {
     const omp = new OmpHarness();
     await omp.start();
     omp.processExit("OMP RPC process exited with code 1 and signal null");
     const turn = omp.startTurnDetached("continue");
+    const rejected = expect(turn).rejects.toThrow("closed");
     await omp.close();
-    await turn;
+    await rejected;
     await waitForImmediate();
     expect(omp.runtimeSessions().every((session) => session.closed)).toBe(true);
   });
@@ -1664,3 +1665,56 @@ test.each([false, true])(
     }
   },
 );
+
+test("interrupting OMP during image conversion cancels the admitted turn and permits a follow-up", async () => {
+  const runtime = new FakeOmp();
+  const client = new OmpAgentClient({ logger: createTestLogger(), runtime });
+  const session = await client.createSession({ provider: "omp", cwd: process.cwd() });
+  onTestFinished(() => session.close());
+  const events: AgentStreamEvent[] = [];
+  session.subscribe((event) => events.push(event));
+  let admitted = false;
+  const first = session
+    .startTurn([
+      {
+        type: "image",
+        data: Buffer.alloc(16 * 1024 * 1024, 53).toString("base64"),
+        mimeType: "image/png",
+      },
+    ])
+    .then((result) => {
+      admitted = true;
+      return result;
+    });
+  await waitForImmediate();
+  expect(admitted).toBe(false);
+  expect(runtime.latestSession().prompts).toHaveLength(0);
+  await session.interrupt();
+  const followup = await session.startTurn("follow-up after image interrupt");
+  const original = await first;
+  expect(events).toContainEqual(
+    expect.objectContaining({ type: "turn_canceled", turnId: original.turnId }),
+  );
+  expect(followup.turnId).not.toBe(original.turnId);
+  expect(runtime.latestSession().prompts).toEqual([
+    { message: "follow-up after image interrupt", imageCount: 0 },
+  ]);
+});
+
+test("closing OMP during image conversion rejects its pending start", async () => {
+  const runtime = new FakeOmp();
+  const client = new OmpAgentClient({ logger: createTestLogger(), runtime });
+  const session = await client.createSession({ provider: "omp", cwd: process.cwd() });
+  const first = session.startTurn([
+    {
+      type: "image",
+      data: Buffer.alloc(16 * 1024 * 1024, 54).toString("base64"),
+      mimeType: "image/png",
+    },
+  ]);
+  const rejected = expect(first).rejects.toThrow("closed");
+  await waitForImmediate();
+  expect(runtime.latestSession().prompts).toHaveLength(0);
+  await session.close();
+  await rejected;
+});

packages/server/src/server/agent/providers/omp/agent.ts

@@ -879,7 +879,8 @@ export class OmpAgentSession implements AgentSession {
     this.activePromptRequestId = null;
     this.clearNoTurnBuffers();
     const payload = await convertPromptInput(prompt, { model: this.state.model });
-    if (this.closed || this.activeTurnId !== turnId) return { turnId };
+    if (this.closed) throw new Error("OMP session is closed");
+    if (this.activeTurnId !== turnId) return { turnId };
     this.rememberClientMessage(options?.clientMessageId ?? null, payload.text);
     this.activeNoTurnPromptText = payload.text;
     this.usagePoller.startTurn();

packages/server/src/server/agent/providers/opencode/event-translator.test.ts

@@ -1560,7 +1560,7 @@ describe("translateOpenCodeEvent", () => {
 });
 
 describe("OpenCode v2 history reconciliation", () => {
-  it("keeps client message identity on live echoes but omits it from history", () => {
+  it("keeps client message identity on live echoes but omits it from history", async () => {
     const user: SessionMessageUser = {
       id: "native-message",
       type: "user",
@@ -1568,13 +1568,13 @@ describe("OpenCode v2 history reconciliation", () => {
       time: { created: 1 },
       metadata: { paseoClientMessageId: "client-message" },
     };
-    expect(new V2Timeline().messages([user])).toMatchObject([
+    expect(await new V2Timeline().messages([user])).toMatchObject([
       { item: { clientMessageId: "client-message", messageId: "native-message" } },
     ]);
-    expect(new V2Timeline(false).messages([user])).toMatchObject([
+    expect(await new V2Timeline(false).messages([user])).toMatchObject([
       { item: { type: "user_message", messageId: "native-message" } },
     ]);
-    const [event] = new V2Timeline(false).messages([user]);
+    const [event] = await new V2Timeline(false).messages([user]);
     expect(event && "item" in event && event.item).not.toHaveProperty("clientMessageId");
   });
   function assistant(text: string): SessionMessageAssistant {
@@ -1587,31 +1587,33 @@ describe("OpenCode v2 history reconciliation", () => {
       content: [{ type: "text", text }],
     };
   }
-  it("emits only the missing suffix after reconnect, even if an intermediate snapshot lags", () => {
+  it("emits only the missing suffix after reconnect, even if an intermediate snapshot lags", async () => {
     const timeline = new V2Timeline();
-    const text = (messages: ReturnType<V2Timeline["messages"]>) =>
+    const text = (messages: Awaited<ReturnType<V2Timeline["messages"]>>) =>
       messages.flatMap((event) =>
         event.type === "timeline" && event.item.type === "assistant_message"
           ? [event.item.text]
           : [],
       );
-    expect(text(timeline.messages([assistant("hello")]))).toEqual(["hello"]);
-    expect(timeline.messages([assistant("hel")])).toEqual([]);
-    expect(text(timeline.messages([assistant("hello world")]))).toEqual([" world"]);
-    expect(timeline.messages([assistant("hello world")])).toEqual([]);
+    expect(text(await timeline.messages([assistant("hello")]))).toEqual(["hello"]);
+    expect(await timeline.messages([assistant("hel")])).toEqual([]);
+    expect(text(await timeline.messages([assistant("hello world")]))).toEqual([" world"]);
+    expect(await timeline.messages([assistant("hello world")])).toEqual([]);
   });
-  it("keeps reasoning separate from assistant output", () => {
+  it("keeps reasoning separate from assistant output", async () => {
     const timeline = new V2Timeline();
     const message = assistant("answer");
     message.content.unshift({ type: "reasoning", text: "thinking" });
     expect(
-      timeline.messages([message]).map((event) => (event.type === "timeline" ? event.item : null)),
+      (await timeline.messages([message])).map((event) =>
+        event.type === "timeline" ? event.item : null,
+      ),
     ).toEqual([
       { type: "reasoning", text: "thinking" },
       { type: "assistant_message", text: "answer", messageId: "message" },
     ]);
   });
-  it("emits tool state transitions once and keeps their call identity", () => {
+  it("emits tool state transitions once and keeps their call identity", async () => {
     const timeline = new V2Timeline();
     const message = assistant("");
     message.content = [
@@ -1623,7 +1625,7 @@ describe("OpenCode v2 history reconciliation", () => {
         state: { status: "running", input: { command: "pwd" }, metadata: {} },
       },
     ];
-    expect(timeline.messages([message])).toMatchObject([
+    expect(await timeline.messages([message])).toMatchObject([
       { item: { type: "tool_call", callId: "call", status: "running" } },
     ]);
     message.content = [
@@ -1639,9 +1641,9 @@ describe("OpenCode v2 history reconciliation", () => {
         },
       },
     ];
-    expect(timeline.messages([message])).toMatchObject([
+    expect(await timeline.messages([message])).toMatchObject([
       { item: { type: "tool_call", callId: "call", status: "completed" } },
     ]);
-    expect(timeline.messages([message])).toEqual([]);
+    expect(await timeline.messages([message])).toEqual([]);
   });
 });

packages/server/src/server/agent/providers/pi/agent.test.ts

@@ -3594,3 +3594,52 @@ test("reserves the Pi turn while an image prompt is materialized", async () => {
   await first;
   expect(pi.latestSession().prompts).toHaveLength(1);
 });
+
+test("interrupting Pi during image conversion cancels the admitted turn and permits a follow-up", async () => {
+  const { pi, session } = await createSession();
+  onTestFinished(() => session.close());
+  const events: AgentStreamEvent[] = [];
+  session.subscribe((event) => events.push(event));
+  let admitted = false;
+  const first = session
+    .startTurn([
+      {
+        type: "image",
+        data: Buffer.alloc(16 * 1024 * 1024, 51).toString("base64"),
+        mimeType: "image/png",
+      },
+    ])
+    .then((result) => {
+      admitted = true;
+      return result;
+    });
+  await waitForImmediate();
+  expect(admitted).toBe(false);
+  expect(pi.latestSession().prompts).toHaveLength(0);
+  await session.interrupt();
+  const followup = await session.startTurn("follow-up after image interrupt");
+  const original = await first;
+  expect(events).toContainEqual(
+    expect.objectContaining({ type: "turn_canceled", turnId: original.turnId }),
+  );
+  expect(followup.turnId).not.toBe(original.turnId);
+  expect(pi.latestSession().prompts).toEqual([
+    { message: "follow-up after image interrupt", imageCount: 0 },
+  ]);
+});
+
+test("closing Pi during image conversion rejects its pending start", async () => {
+  const { pi, session } = await createSession();
+  const first = session.startTurn([
+    {
+      type: "image",
+      data: Buffer.alloc(16 * 1024 * 1024, 52).toString("base64"),
+      mimeType: "image/png",
+    },
+  ]);
+  const rejected = expect(first).rejects.toThrow("closed");
+  await waitForImmediate();
+  expect(pi.latestSession().prompts).toHaveLength(0);
+  await session.close();
+  await rejected;
+});

packages/server/src/server/agent/providers/pi/agent.ts

@@ -1292,7 +1292,8 @@ export class PiRpcAgentSession implements AgentSession {
     this.pendingSteerSubmissions.length = 0;
     this.clearNoTurnBuffers();
     const payload = await convertPromptInput(prompt, { model: this.state.model });
-    if (this.closed || this.activeTurnId !== turnId) return { turnId };
+    if (this.closed) throw new Error("Pi session is closed");
+    if (this.activeTurnId !== turnId) return { turnId };
     this.activeNoTurnPromptText = payload.text;
     const shouldProbeForNoTurnPrompt = this.parseSlashCommandInput(payload.text) !== null;