fix(server): recover completed opencode turns after SSE EOF (#895)

* fix(server): recover completed opencode turns after SSE EOF (#861)

* fix(server): wait for opencode completion after EOF

The original EOF recovery only checked OpenCode storage once after the SSE stream ended. That missed the real failure mode from #861: OpenCode can drop /event while the turn continues behind the scenes, then persist the assistant completion a little later.

Poll the persisted session for the active turn before failing EOF, while still requiring strong completion evidence and ignoring messages that predate the turn. This preserves failure behavior when there is no persisted completion.

Add behavioral coverage for delayed completion after EOF, partial streamed text plus persisted completion without duplication, stale old completions, and the no-evidence failure path.

---------

Co-authored-by: pluto <plutofog@proton.me>
Co-authored-by: Mohamed Boudra <boudra.moha@gmail.com>

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

@@ -13,6 +13,7 @@ import {
 } from "./opencode-agent.js";
 import { streamSession } from "./test-utils/session-stream-adapter.js";
 import type {
+  AgentSession,
   AgentSessionConfig,
   AgentStreamEvent,
   ToolCallTimelineItem,
@@ -42,6 +43,8 @@ interface TurnResult {
   error?: string;
 }
 
+const FAST_OPENCODE_EOF_RECOVERY_POLICY = { maxAttempts: 5, delayMs: 1 };
+
 async function collectTurnEvents(iterator: AsyncGenerator<AgentStreamEvent>): Promise<TurnResult> {
   const result: TurnResult = {
     events: [],
@@ -78,6 +81,140 @@ async function collectTurnEvents(iterator: AsyncGenerator<AgentStreamEvent>): Pr
   return result;
 }
 
+interface FakeOpenCodeStream {
+  stream: AsyncIterable<OpenCodeEvent>;
+  close: () => void;
+  closed: Promise<void>;
+}
+
+function createFakeOpenCodeStream(events: OpenCodeEvent[] = []): FakeOpenCodeStream {
+  let releaseStream!: () => void;
+  const canEnd = new Promise<void>((resolve) => {
+    releaseStream = resolve;
+  });
+  let notifyEnded!: () => void;
+  const closed = new Promise<void>((resolve) => {
+    notifyEnded = resolve;
+  });
+
+  return {
+    close: releaseStream,
+    closed,
+    stream: {
+      [Symbol.asyncIterator]: () => {
+        let index = 0;
+        return {
+          next: async () => {
+            if (index < events.length) {
+              return { done: false, value: events[index++] };
+            }
+            await canEnd;
+            notifyEnded();
+            return { done: true, value: undefined };
+          },
+        };
+      },
+    },
+  };
+}
+
+function createFakeOpenCode(stream: FakeOpenCodeStream, onPrompt: () => void | Promise<void>) {
+  return {
+    event: {
+      subscribe: async () => ({ stream: stream.stream }),
+    },
+    provider: {
+      list: async () => ({ data: { connected: [], all: [] }, error: undefined }),
+    },
+    session: {
+      create: async () => ({ data: { id: "ses_unit_test" }, error: undefined }),
+      promptAsync: async () => {
+        await onPrompt();
+        return { data: {}, error: undefined };
+      },
+      abort: async () => ({ data: true, error: undefined }),
+      update: async () => ({ data: true, error: undefined }),
+      delete: async () => ({ data: true, error: undefined }),
+    },
+  } as never;
+}
+
+async function createSessionWithFakeOpenCode(params: {
+  storageRoot: string;
+  cwd: string;
+  stream: FakeOpenCodeStream;
+  onPrompt: () => void | Promise<void>;
+}): Promise<AgentSession> {
+  const fakeClient = createFakeOpenCode(params.stream, params.onPrompt);
+  const client = new OpenCodeAgentClient(createTestLogger(), undefined, params.storageRoot, {
+    runtime: {
+      acquireServer: async () => ({
+        server: { port: 0, url: "http://localhost" },
+        release: () => {},
+      }),
+      ensureServerRunning: async () => ({ port: 0, url: "http://localhost" }),
+      createClient: () => fakeClient,
+      shutdown: async () => undefined,
+    },
+    eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY,
+  });
+
+  return client.createSession({ provider: "opencode", cwd: params.cwd });
+}
+
+function observeTurn(session: AgentSession) {
+  const events: AgentStreamEvent[] = [];
+  let resolveTerminal!: (event: AgentStreamEvent) => void;
+  const terminal = new Promise<AgentStreamEvent>((resolve) => {
+    resolveTerminal = resolve;
+  });
+
+  session.subscribe((event) => {
+    events.push(event);
+    if (
+      event.type === "turn_completed" ||
+      event.type === "turn_failed" ||
+      event.type === "turn_canceled"
+    ) {
+      resolveTerminal(event);
+    }
+  });
+
+  return {
+    terminal,
+    assistantMessages: () =>
+      events
+        .flatMap((event) => (event.type === "timeline" ? [event.item] : []))
+        .filter((item): item is AssistantMessageTimelineItem => item.type === "assistant_message"),
+  };
+}
+
+function openCodeAssistantStarted(messageId: string): OpenCodeEvent {
+  return {
+    type: "message.updated",
+    properties: {
+      info: {
+        id: messageId,
+        sessionID: "ses_unit_test",
+        role: "assistant",
+      },
+    },
+  } as OpenCodeEvent;
+}
+
+function openCodeTextDelta(messageId: string, partId: string, delta: string): OpenCodeEvent {
+  return {
+    type: "message.part.delta",
+    properties: {
+      sessionID: "ses_unit_test",
+      messageID: messageId,
+      partID: partId,
+      field: "text",
+      delta,
+    },
+  } as OpenCodeEvent;
+}
+
 function isBinaryInstalled(binary: string): boolean {
   try {
     const out = execFileSync("which", [binary], { encoding: "utf8" }).trim();
@@ -566,6 +703,444 @@ describe("OpenCode adapter context-window normalization", () => {
 });
 
 describe("OpenCode adapter startTurn error handling", () => {
+  test("recovers SSE EOF into turn_completed when persisted assistant completion exists", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    writeOpenCodeJson(storageRoot, "session/project-1/ses_unit_test.json", {
+      id: "ses_unit_test",
+      directory: cwd,
+      time: { created: 1000, updated: 3000 },
+    });
+    let releaseStream!: () => void;
+    const streamMayEnd = new Promise<void>((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockResolvedValue({
+          stream: {
+            [Symbol.asyncIterator]: () => {
+              let index = 0;
+              const events: OpenCodeEvent[] = [
+                {
+                  type: "message.updated",
+                  properties: {
+                    info: {
+                      id: "msg_assistant",
+                      sessionID: "ses_unit_test",
+                      role: "assistant",
+                    },
+                  },
+                } as OpenCodeEvent,
+                {
+                  type: "message.part.delta",
+                  properties: {
+                    sessionID: "ses_unit_test",
+                    messageID: "msg_assistant",
+                    partID: "prt_text",
+                    field: "text",
+                    delta: "Recovered ",
+                  },
+                } as OpenCodeEvent,
+              ];
+
+              return {
+                next: async () => {
+                  if (index < events.length) {
+                    return { done: false, value: events[index++] };
+                  }
+                  await streamMayEnd;
+                  return { done: true, value: undefined };
+                },
+              };
+            },
+          },
+        }),
+      },
+      provider: {
+        list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }),
+      },
+      session: {
+        create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }),
+        promptAsync: vi.fn().mockImplementation(async () => {
+          writeOpenCodeJson(storageRoot, "message/ses_unit_test/msg_assistant.json", {
+            id: "msg_assistant",
+            sessionID: "ses_unit_test",
+            role: "assistant",
+            finish: "stop",
+            time: { created: 2000, completed: 2500 },
+          });
+          writeOpenCodeJson(storageRoot, "part/msg_assistant/prt_text.json", {
+            id: "prt_text",
+            sessionID: "ses_unit_test",
+            messageID: "msg_assistant",
+            type: "text",
+            text: "Recovered assistant reply",
+            time: { start: 2100, end: 2400 },
+          });
+          writeOpenCodeJson(storageRoot, "part/msg_assistant/prt_finish.json", {
+            id: "prt_finish",
+            sessionID: "ses_unit_test",
+            messageID: "msg_assistant",
+            type: "step-finish",
+            time: { start: 2400, end: 2500 },
+            tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 } },
+          });
+          releaseStream();
+          return { data: {}, error: undefined };
+        }),
+        abort: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        update: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        delete: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+      },
+    } as never;
+
+    const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, {
+      runtime: {
+        acquireServer: vi.fn().mockResolvedValue({
+          server: { port: 0, url: "http://localhost" },
+          release: () => {},
+        }),
+        ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }),
+        createClient: vi.fn().mockReturnValue(fakeClient),
+        shutdown: vi.fn().mockResolvedValue(undefined),
+      },
+      eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY,
+    });
+
+    const session = await client.createSession({ provider: "opencode", cwd });
+    const turn = await collectTurnEvents(streamSession(session, "hello"));
+
+    expect(turn.turnCompleted).toBe(true);
+    expect(turn.turnFailed).toBe(false);
+    expect(turn.assistantMessages.map((message) => message.text).join("")).toBe(
+      "Recovered assistant reply",
+    );
+    expect(turn.events).toContainEqual(
+      expect.objectContaining({
+        type: "turn_completed",
+        usage: expect.objectContaining({
+          contextWindowUsedTokens: 15,
+          inputTokens: 10,
+          outputTokens: 5,
+        }),
+      }),
+    );
+
+    dateNowSpy.mockRestore();
+    rmSync(storageRoot, { recursive: true, force: true });
+  });
+
+  test("recovers SSE EOF when persisted assistant completion appears after the stream closes", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    try {
+      writeOpenCodeStoredSession(storageRoot, cwd);
+      const stream = createFakeOpenCodeStream();
+      const session = await createSessionWithFakeOpenCode({
+        storageRoot,
+        cwd,
+        stream,
+        onPrompt: stream.close,
+      });
+      const turn = observeTurn(session);
+
+      await session.startTurn("hello");
+      await stream.closed;
+      writeRecoveredAssistantCompletion(storageRoot, "msg_assistant", "Recovered assistant reply");
+
+      expect((await turn.terminal).type).toBe("turn_completed");
+      expect(
+        turn
+          .assistantMessages()
+          .map((message) => message.text)
+          .join(""),
+      ).toBe("Recovered assistant reply");
+    } finally {
+      dateNowSpy.mockRestore();
+      rmSync(storageRoot, { recursive: true, force: true });
+    }
+  });
+
+  test("recovers delayed persisted completion without duplicating text already streamed before EOF", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+    let completionWriteSettled: Promise<void> = Promise.resolve();
+    const writeDelayedCompletion = async () => {
+      await nextMacrotask();
+      writeRecoveredAssistantCompletion(storageRoot, "msg_assistant", "Recovered assistant reply");
+    };
+
+    try {
+      writeOpenCodeStoredSession(storageRoot, cwd);
+      const stream = createFakeOpenCodeStream([
+        openCodeAssistantStarted("msg_assistant"),
+        openCodeTextDelta("msg_assistant", "prt_text", "Recovered "),
+      ]);
+      const session = await createSessionWithFakeOpenCode({
+        storageRoot,
+        cwd,
+        stream,
+        onPrompt: async () => {
+          stream.close();
+          completionWriteSettled = writeDelayedCompletion();
+          await completionWriteSettled;
+        },
+      });
+      const turn = observeTurn(session);
+
+      await session.startTurn("hello");
+      await stream.closed;
+
+      expect((await turn.terminal).type).toBe("turn_completed");
+      const assistantMessages = turn.assistantMessages();
+      expect(assistantMessages.map((message) => message.text).join("")).toBe(
+        "Recovered assistant reply",
+      );
+      expect(assistantMessages.map((message) => message.text)).toEqual([
+        "Recovered ",
+        "assistant reply",
+      ]);
+    } finally {
+      await completionWriteSettled;
+      dateNowSpy.mockRestore();
+      rmSync(storageRoot, { recursive: true, force: true });
+    }
+  });
+
+  test("ignores old completed assistant messages while waiting for delayed current-turn completion", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    try {
+      writeOpenCodeStoredSession(storageRoot, cwd);
+      writeOpenCodeAssistantCompletion(storageRoot, "msg_assistant_old", "Old assistant reply", {
+        created: 1500,
+        completed: 1600,
+        partStart: 1500,
+        partEnd: 1550,
+      });
+      const stream = createFakeOpenCodeStream();
+      const session = await createSessionWithFakeOpenCode({
+        storageRoot,
+        cwd,
+        stream,
+        onPrompt: stream.close,
+      });
+      const turn = observeTurn(session);
+
+      await session.startTurn("hello");
+      await stream.closed;
+      writeRecoveredAssistantCompletion(
+        storageRoot,
+        "msg_assistant_current",
+        "Current assistant reply",
+      );
+
+      expect((await turn.terminal).type).toBe("turn_completed");
+      expect(turn.assistantMessages().map((message) => message.text)).toEqual([
+        "Current assistant reply",
+      ]);
+    } finally {
+      dateNowSpy.mockRestore();
+      rmSync(storageRoot, { recursive: true, force: true });
+    }
+  });
+
+  test("keeps SSE EOF as turn_failed without persisted completion evidence", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    writeOpenCodeJson(storageRoot, "session/project-1/ses_unit_test.json", {
+      id: "ses_unit_test",
+      directory: cwd,
+      time: { created: 1000, updated: 3000 },
+    });
+    let releaseStream!: () => void;
+    const streamMayEnd = new Promise<void>((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockResolvedValue({
+          stream: {
+            [Symbol.asyncIterator]: () => ({
+              next: async () => {
+                await streamMayEnd;
+                return { done: true, value: undefined };
+              },
+            }),
+          },
+        }),
+      },
+      provider: {
+        list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }),
+      },
+      session: {
+        create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }),
+        promptAsync: vi.fn().mockImplementation(async () => {
+          writeOpenCodeJson(storageRoot, "message/ses_unit_test/msg_assistant.json", {
+            id: "msg_assistant",
+            sessionID: "ses_unit_test",
+            role: "assistant",
+            time: { created: 2000 },
+          });
+          writeOpenCodeJson(storageRoot, "part/msg_assistant/prt_text.json", {
+            id: "prt_text",
+            sessionID: "ses_unit_test",
+            messageID: "msg_assistant",
+            type: "text",
+            text: "Incomplete assistant reply",
+            time: { start: 2100 },
+          });
+          releaseStream();
+          return { data: {}, error: undefined };
+        }),
+        abort: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        update: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        delete: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+      },
+    } as never;
+
+    const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, {
+      runtime: {
+        acquireServer: vi.fn().mockResolvedValue({
+          server: { port: 0, url: "http://localhost" },
+          release: () => {},
+        }),
+        ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }),
+        createClient: vi.fn().mockReturnValue(fakeClient),
+        shutdown: vi.fn().mockResolvedValue(undefined),
+      },
+      eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY,
+    });
+
+    const session = await client.createSession({ provider: "opencode", cwd });
+    const turn = await collectTurnEvents(streamSession(session, "hello"));
+
+    expect(turn.turnCompleted).toBe(false);
+    expect(turn.turnFailed).toBe(true);
+    expect(turn.error).toBe("OpenCode event stream ended before the turn reached a terminal state");
+
+    dateNowSpy.mockRestore();
+    rmSync(storageRoot, { recursive: true, force: true });
+  });
+
+  test("does not recover a previous turn's completed assistant reply when the current turn only persists incomplete output", async () => {
+    const storageRoot = mkdtempSync(path.join(os.tmpdir(), "opencode-storage-"));
+    const cwd = "/tmp/test";
+    const dateNowSpy = vi.spyOn(Date, "now").mockReturnValue(2000);
+
+    writeOpenCodeJson(storageRoot, "session/project-1/ses_unit_test.json", {
+      id: "ses_unit_test",
+      directory: cwd,
+      time: { created: 1000, updated: 3000 },
+    });
+
+    let releaseStream!: () => void;
+    const streamMayEnd = new Promise<void>((resolve) => {
+      releaseStream = resolve;
+    });
+
+    const fakeClient = {
+      event: {
+        subscribe: vi.fn().mockResolvedValue({
+          stream: {
+            [Symbol.asyncIterator]: () => ({
+              next: async () => {
+                await streamMayEnd;
+                return { done: true, value: undefined };
+              },
+            }),
+          },
+        }),
+      },
+      provider: {
+        list: vi.fn().mockResolvedValue({ data: { connected: [], all: [] }, error: undefined }),
+      },
+      session: {
+        create: vi.fn().mockResolvedValue({ data: { id: "ses_unit_test" }, error: undefined }),
+        promptAsync: vi.fn().mockImplementation(async () => {
+          writeOpenCodeJson(storageRoot, "message/ses_unit_test/msg_assistant_old.json", {
+            id: "msg_assistant_old",
+            sessionID: "ses_unit_test",
+            role: "assistant",
+            finish: "stop",
+            time: { created: 1500, completed: 1600 },
+          });
+          writeOpenCodeJson(storageRoot, "part/msg_assistant_old/prt_text.json", {
+            id: "prt_text_old",
+            sessionID: "ses_unit_test",
+            messageID: "msg_assistant_old",
+            type: "text",
+            text: "Old assistant reply",
+            time: { start: 1500, end: 1550 },
+          });
+          writeOpenCodeJson(storageRoot, "part/msg_assistant_old/prt_finish.json", {
+            id: "prt_finish_old",
+            sessionID: "ses_unit_test",
+            messageID: "msg_assistant_old",
+            type: "step-finish",
+            time: { start: 1550, end: 1600 },
+            tokens: { input: 10, output: 5, reasoning: 0, cache: { read: 0, write: 0 } },
+          });
+          writeOpenCodeJson(storageRoot, "message/ses_unit_test/msg_assistant_current.json", {
+            id: "msg_assistant_current",
+            sessionID: "ses_unit_test",
+            role: "assistant",
+            time: { created: 2100 },
+          });
+          writeOpenCodeJson(storageRoot, "part/msg_assistant_current/prt_text.json", {
+            id: "prt_text_current",
+            sessionID: "ses_unit_test",
+            messageID: "msg_assistant_current",
+            type: "text",
+            text: "Incomplete assistant reply",
+            time: { start: 2100 },
+          });
+          releaseStream();
+          return { data: {}, error: undefined };
+        }),
+        abort: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        update: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+        delete: vi.fn().mockResolvedValue({ data: true, error: undefined }),
+      },
+    } as never;
+
+    const client = new OpenCodeAgentClient(createTestLogger(), undefined, storageRoot, {
+      runtime: {
+        acquireServer: vi.fn().mockResolvedValue({
+          server: { port: 0, url: "http://localhost" },
+          release: () => {},
+        }),
+        ensureServerRunning: vi.fn().mockResolvedValue({ port: 0, url: "http://localhost" }),
+        createClient: vi.fn().mockReturnValue(fakeClient),
+        shutdown: vi.fn().mockResolvedValue(undefined),
+      },
+      eofRecoveryPolicy: FAST_OPENCODE_EOF_RECOVERY_POLICY,
+    });
+
+    const session = await client.createSession({ provider: "opencode", cwd });
+    const turn = await collectTurnEvents(streamSession(session, "hello"));
+
+    expect(turn.turnCompleted).toBe(false);
+    expect(turn.turnFailed).toBe(true);
+    expect(turn.assistantMessages).toHaveLength(0);
+    expect(turn.error).toBe("OpenCode event stream ended before the turn reached a terminal state");
+
+    dateNowSpy.mockRestore();
+    rmSync(storageRoot, { recursive: true, force: true });
+  });
+
   test("deletes provider session on close when persistence is disabled", async () => {
     const fakeClient = {
       session: {
@@ -580,6 +1155,7 @@ describe("OpenCode adapter startTurn error handling", () => {
       fakeClient,
       "ses_unit_test",
       createTestLogger(),
+      "/tmp/opencode-storage",
       new Map(),
       undefined,
       false,
@@ -607,6 +1183,7 @@ describe("OpenCode adapter startTurn error handling", () => {
       fakeClient,
       "ses_unit_test",
       createTestLogger(),
+      "/tmp/opencode-storage",
     );
 
     await session.close();
@@ -641,6 +1218,7 @@ describe("OpenCode adapter startTurn error handling", () => {
       fakeClient,
       "ses_unit_test",
       createTestLogger(),
+      "/tmp/opencode-storage",
     );
 
     const events: AgentStreamEvent[] = [];
@@ -740,3 +1318,51 @@ function writeOpenCodeJson(storageRoot: string, relativePath: string, value: unk
   mkdirSync(path.dirname(filePath), { recursive: true });
   writeFileSync(filePath, JSON.stringify(value), "utf8");
 }
+
+function writeOpenCodeStoredSession(storageRoot: string, cwd: string): void {
+  writeOpenCodeJson(storageRoot, "session/project-1/ses_unit_test.json", {
+    id: "ses_unit_test",
+    directory: cwd,
+    time: { created: 1000, updated: 3000 },
+  });
+}
+
+function nextMacrotask(): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, 0));
+}
+
+function writeRecoveredAssistantCompletion(
+  storageRoot: string,
+  messageId: string,
+  text: string,
+): void {
+  writeOpenCodeAssistantCompletion(storageRoot, messageId, text, {
+    created: 2000,
+    completed: 2500,
+    partStart: 2100,
+    partEnd: 2400,
+  });
+}
+
+function writeOpenCodeAssistantCompletion(
+  storageRoot: string,
+  messageId: string,
+  text: string,
+  time: { created: number; completed: number; partStart: number; partEnd: number },
+): void {
+  writeOpenCodeJson(storageRoot, `message/ses_unit_test/${messageId}.json`, {
+    id: messageId,
+    sessionID: "ses_unit_test",
+    role: "assistant",
+    finish: "stop",
+    time: { created: time.created, completed: time.completed },
+  });
+  writeOpenCodeJson(storageRoot, `part/${messageId}/prt_text.json`, {
+    id: "prt_text",
+    sessionID: "ses_unit_test",
+    messageID: messageId,
+    type: "text",
+    text,
+    time: { start: time.partStart, end: time.partEnd },
+  });
+}

packages/server/src/server/agent/providers/opencode-agent.ts

@@ -74,6 +74,10 @@ const OPENCODE_CAPABILITIES: AgentCapabilityFlags = {
 const OPENCODE_BUILD_MODE_ID = "build";
 const OPENCODE_FULL_ACCESS_MODE_ID = "full-access";
 const OPENCODE_STORAGE_SESSION_LIMIT = 200;
+const DEFAULT_OPENCODE_EOF_RECOVERY_POLICY = {
+  maxAttempts: 600,
+  delayMs: 1_000,
+};
 
 const DEFAULT_MODES: AgentMode[] = [
   {
@@ -138,6 +142,17 @@ type OpenCodeStoredSession = z.infer<typeof OpenCodeStoredSessionSchema>;
 type OpenCodeStoredMessage = z.infer<typeof OpenCodeStoredMessageSchema>;
 type OpenCodeStoredPart = z.infer<typeof OpenCodeStoredPartSchema>;
 
+interface OpenCodePersistedAssistantCompletion {
+  messageId: string;
+  text: string;
+  usage?: AgentUsage;
+}
+
+interface OpenCodeEofRecoveryPolicy {
+  maxAttempts: number;
+  delayMs: number;
+}
+
 type OpenCodeAgentConfig = AgentSessionConfig & { provider: "opencode" };
 type OpenCodeMessageRole = "user" | "assistant";
 
@@ -818,23 +833,137 @@ async function readOpenCodeSessionTimeline(
 }
 
 async function readOpenCodeMessageText(storageRoot: string, messageId: string): Promise<string> {
+  const parts = await readOpenCodeStoredParts(storageRoot, messageId);
+  return readOpenCodeTextFromParts(parts);
+}
+
+async function readOpenCodeStoredParts(
+  storageRoot: string,
+  messageId: string,
+): Promise<OpenCodeStoredPart[]> {
   const partRoot = path.join(storageRoot, "part", messageId);
   const partFiles = await findJsonFiles(partRoot);
   const parts: OpenCodeStoredPart[] = [];
   for (const file of partFiles) {
     const parsed = await readJsonFile(file, OpenCodeStoredPartSchema);
-    if (parsed?.type === "text" && typeof parsed.text === "string") {
+    if (parsed) {
       parts.push(parsed);
     }
   }
 
+  return parts.sort(
+    (left, right) => getOpenCodePartTimestamp(left) - getOpenCodePartTimestamp(right),
+  );
+}
+
+function readOpenCodeTextFromParts(parts: OpenCodeStoredPart[]): string {
   return parts
-    .sort((left, right) => getOpenCodePartTimestamp(left) - getOpenCodePartTimestamp(right))
+    .filter((part) => part.type === "text" && typeof part.text === "string")
     .map((part) => part.text?.trim() ?? "")
     .filter(Boolean)
     .join("\n\n");
 }
 
+async function readOpenCodePersistedAssistantCompletion(
+  storageRoot: string,
+  sessionId: string,
+  knownMessageIds: ReadonlySet<string>,
+  turnStartedAt: number,
+): Promise<OpenCodePersistedAssistantCompletion | null> {
+  const messageRoot = path.join(storageRoot, "message", sessionId);
+  const messageFiles = await findJsonFiles(messageRoot);
+  const messages: OpenCodeStoredMessage[] = [];
+
+  for (const file of messageFiles) {
+    const parsed = await readJsonFile(file, OpenCodeStoredMessageSchema);
+    if (
+      parsed?.sessionID === sessionId &&
+      parsed.role === "assistant" &&
+      !knownMessageIds.has(parsed.id)
+    ) {
+      messages.push(parsed);
+    }
+  }
+
+  const candidates = messages.sort(
+    (left, right) => getOpenCodeMessageTimestamp(right) - getOpenCodeMessageTimestamp(left),
+  );
+
+  for (const message of candidates) {
+    const parts = (await readOpenCodeStoredParts(storageRoot, message.id)).filter((part) =>
+      isOpenCodePartAtOrAfterTurnStart(part, turnStartedAt),
+    );
+    if (!isOpenCodeMessageAtOrAfterTurnStart(message, turnStartedAt) && parts.length === 0) {
+      continue;
+    }
+
+    const text = readOpenCodeTextFromParts(parts);
+    if (!text || !hasStrongPersistedCompletionEvidence(message, parts, turnStartedAt)) {
+      continue;
+    }
+    const usage = readPersistedStepFinishUsage(parts);
+
+    return {
+      messageId: message.id,
+      text,
+      ...(hasNormalizedOpenCodeUsage(usage) ? { usage } : {}),
+    };
+  }
+
+  return null;
+}
+
+function sleep(ms: number): Promise<void> {
+  return new Promise((resolve) => setTimeout(resolve, ms));
+}
+
+function hasStrongPersistedCompletionEvidence(
+  message: OpenCodeStoredMessage,
+  parts: OpenCodeStoredPart[],
+  turnStartedAt: number,
+): boolean {
+  const messageRecord = readOpenCodeRecord(message);
+  const infoRecord = readOpenCodeRecord(messageRecord?.["info"]);
+  const finish =
+    readNonEmptyString(messageRecord?.["finish"]) ?? readNonEmptyString(infoRecord?.["finish"]);
+  const hasCompletedMessage = isOpenCodeTimestampAtOrAfter(message.time?.completed, turnStartedAt);
+  const hasCompletedTextPart = parts.some(
+    (part) => part.type === "text" && typeof part.time?.end === "number",
+  );
+  const hasStepFinish = parts.some((part) => part.type === "step-finish");
+
+  return finish === "stop" && (hasCompletedMessage || hasCompletedTextPart || hasStepFinish);
+}
+
+function readPersistedStepFinishUsage(parts: OpenCodeStoredPart[]): AgentUsage {
+  const usage: AgentUsage = {};
+
+  for (const part of parts.filter((candidate) => candidate.type === "step-finish")) {
+    const partRecord = readOpenCodeRecord(part);
+    const tokensRecord = readOpenCodeRecord(partRecord?.["tokens"]);
+    const cacheRecord = readOpenCodeRecord(tokensRecord?.["cache"]);
+    mergeOpenCodeStepFinishUsage(usage, {
+      cost: partRecord?.["cost"],
+      tokens: tokensRecord
+        ? {
+            input: tokensRecord["input"],
+            output: tokensRecord["output"],
+            reasoning: tokensRecord["reasoning"],
+            total: tokensRecord["total"],
+            cache: cacheRecord
+              ? {
+                  read: cacheRecord["read"],
+                  write: cacheRecord["write"],
+                }
+              : undefined,
+          }
+        : undefined,
+    });
+  }
+
+  return usage;
+}
+
 async function findJsonFiles(root: string): Promise<string[]> {
   let entries;
   try {
@@ -880,6 +1009,33 @@ function getOpenCodePartTimestamp(part: OpenCodeStoredPart): number {
   return part.time?.start ?? part.time?.end ?? 0;
 }
 
+function isOpenCodeTimestampAtOrAfter(
+  timestamp: number | undefined,
+  turnStartedAt: number,
+): boolean {
+  return typeof timestamp === "number" && timestamp >= turnStartedAt;
+}
+
+function isOpenCodeMessageAtOrAfterTurnStart(
+  message: OpenCodeStoredMessage,
+  turnStartedAt: number,
+): boolean {
+  return (
+    isOpenCodeTimestampAtOrAfter(message.time?.created, turnStartedAt) ||
+    isOpenCodeTimestampAtOrAfter(message.time?.completed, turnStartedAt)
+  );
+}
+
+function isOpenCodePartAtOrAfterTurnStart(
+  part: OpenCodeStoredPart,
+  turnStartedAt: number,
+): boolean {
+  return (
+    isOpenCodeTimestampAtOrAfter(part.time?.start, turnStartedAt) ||
+    isOpenCodeTimestampAtOrAfter(part.time?.end, turnStartedAt)
+  );
+}
+
 export const __openCodeInternals = {
   buildOpenCodePromptParts,
   buildOpenCodeModelContextWindowLookup,
@@ -900,6 +1056,7 @@ export const __openCodeInternals = {
 
 interface OpenCodeAgentClientDeps {
   runtime?: OpenCodeRuntime;
+  eofRecoveryPolicy?: OpenCodeEofRecoveryPolicy;
 }
 
 class ProductionOpenCodeRuntime implements OpenCodeRuntime {
@@ -931,6 +1088,7 @@ export class OpenCodeAgentClient implements AgentClient {
   private readonly runtimeSettings?: ProviderRuntimeSettings;
   private readonly modelContextWindows = new Map<string, number>();
   private readonly storageRoot: string;
+  private readonly eofRecoveryPolicy: OpenCodeEofRecoveryPolicy;
 
   constructor(
     logger: Logger,
@@ -941,6 +1099,7 @@ export class OpenCodeAgentClient implements AgentClient {
     this.logger = logger.child({ module: "agent", provider: "opencode" });
     this.runtimeSettings = runtimeSettings;
     this.storageRoot = storageRoot ?? resolveOpenCodeStorageRoot();
+    this.eofRecoveryPolicy = deps.eofRecoveryPolicy ?? DEFAULT_OPENCODE_EOF_RECOVERY_POLICY;
     this.runtime =
       deps.runtime ??
       new ProductionOpenCodeRuntime(
@@ -984,9 +1143,11 @@ export class OpenCodeAgentClient implements AgentClient {
         client,
         session.id,
         this.logger,
+        this.storageRoot,
         new Map(this.modelContextWindows),
         acquisition.release,
         options?.persistSession,
+        this.eofRecoveryPolicy,
       );
     } catch (error) {
       acquisition.release();
@@ -1025,8 +1186,11 @@ export class OpenCodeAgentClient implements AgentClient {
         client,
         handle.sessionId,
         this.logger,
+        this.storageRoot,
         new Map(this.modelContextWindows),
         acquisition.release,
+        undefined,
+        this.eofRecoveryPolicy,
       );
     } catch (error) {
       acquisition.release();
@@ -2130,6 +2294,7 @@ class OpenCodeAgentSession implements AgentSession {
   private readonly client: OpencodeClient;
   private readonly sessionId: string;
   private readonly logger: Logger;
+  private readonly storageRoot: string;
   private readonly modelContextWindowsByModelKey: ReadonlyMap<string, number>;
   private currentMode: string = "default";
   private pendingPermissions = new Map<string, AgentPermissionRequest>();
@@ -2156,24 +2321,34 @@ class OpenCodeAgentSession implements AgentSession {
   private selectedModelContextWindowMaxTokens: number | undefined;
   private releaseServer: (() => void) | null;
   private readonly persistSession: boolean;
+  private readonly eofRecoveryPolicy: OpenCodeEofRecoveryPolicy;
   private deletedFromProvider = false;
+  private foregroundAssistantMessageEmitted = false;
+  private foregroundAssistantText = "";
+  private foregroundUsageUpdated = false;
+  private foregroundKnownMessageIds = new Set<string>();
+  private foregroundTurnStartedAt: number | null = null;
   constructor(
     config: OpenCodeAgentConfig,
     client: OpencodeClient,
     sessionId: string,
     logger: Logger,
+    storageRoot: string,
     modelContextWindowsByModelKey: ReadonlyMap<string, number> = new Map(),
     releaseServer?: () => void,
     persistSession = true,
+    eofRecoveryPolicy: OpenCodeEofRecoveryPolicy = DEFAULT_OPENCODE_EOF_RECOVERY_POLICY,
   ) {
     this.config = config;
     this.client = client;
     this.sessionId = sessionId;
     this.logger = logger;
+    this.storageRoot = storageRoot;
     this.modelContextWindowsByModelKey = modelContextWindowsByModelKey;
     this.currentMode = normalizeOpenCodeModeId(config.modeId);
     this.releaseServer = releaseServer ?? null;
     this.persistSession = persistSession;
+    this.eofRecoveryPolicy = eofRecoveryPolicy;
     this.selectedModelContextWindowMaxTokens = this.resolveConfiguredModelContextWindowMaxTokens(
       config.model,
     );
@@ -2243,10 +2418,15 @@ class OpenCodeAgentSession implements AgentSession {
       throw new Error("A foreground turn is already active");
     }
 
+    this.foregroundTurnStartedAt = Date.now();
     this.runningToolCalls.clear();
     this.subAgentsByCallId.clear();
     this.subAgentCallIdByChildSessionId.clear();
     this.pendingChildToolPartsBySessionId.clear();
+    this.foregroundAssistantMessageEmitted = false;
+    this.foregroundAssistantText = "";
+    this.foregroundUsageUpdated = false;
+    this.foregroundKnownMessageIds = await this.readPersistedSessionMessageIds();
     const turnAbortController = new AbortController();
     this.abortController = turnAbortController;
     await this.ensureMcpServersConfigured();
@@ -2466,6 +2646,9 @@ class OpenCodeAgentSession implements AgentSession {
       }
 
       if (!turnAbortController.signal.aborted && this.activeForegroundTurnId === turnId) {
+        if (await this.recoverTurnFromPersistedCompletion(turnId)) {
+          return;
+        }
         this.finishForegroundTurn(
           {
             type: "turn_failed",
@@ -2504,6 +2687,123 @@ class OpenCodeAgentSession implements AgentSession {
     }
   }
 
+  private async recoverTurnFromPersistedCompletion(turnId: string): Promise<boolean> {
+    if (this.foregroundTurnStartedAt === null) {
+      return false;
+    }
+
+    const completion = await this.waitForPersistedAssistantCompletion(turnId);
+    if (!completion || this.activeForegroundTurnId !== turnId) {
+      return false;
+    }
+    this.foregroundKnownMessageIds.add(completion.messageId);
+
+    this.logger.warn(
+      { sessionId: this.sessionId, turnId },
+      "Recovered OpenCode turn completion from persisted session state after SSE EOF",
+    );
+
+    const recoveryText = this.resolvePersistedAssistantRecoveryText(completion.text);
+    if (recoveryText === null) {
+      return false;
+    }
+
+    if (recoveryText.length > 0) {
+      this.notifySubscribers(
+        {
+          type: "timeline",
+          provider: "opencode",
+          item: { type: "assistant_message", text: recoveryText },
+        },
+        turnId,
+      );
+      this.foregroundAssistantMessageEmitted = true;
+    }
+
+    if (completion.usage && !this.foregroundUsageUpdated) {
+      this.accumulatedUsage = {
+        ...this.accumulatedUsage,
+        ...completion.usage,
+      };
+      this.notifySubscribers(
+        {
+          type: "usage_updated",
+          provider: "opencode",
+          usage: { ...this.accumulatedUsage },
+        },
+        turnId,
+      );
+      this.foregroundUsageUpdated = true;
+    }
+
+    this.finishForegroundTurn(
+      {
+        type: "turn_completed",
+        provider: "opencode",
+        usage: hasNormalizedOpenCodeUsage(this.accumulatedUsage)
+          ? { ...this.accumulatedUsage }
+          : undefined,
+      },
+      turnId,
+    );
+    return true;
+  }
+
+  private async waitForPersistedAssistantCompletion(
+    turnId: string,
+  ): Promise<OpenCodePersistedAssistantCompletion | null> {
+    for (let attempt = 0; attempt < this.eofRecoveryPolicy.maxAttempts; attempt += 1) {
+      if (this.foregroundTurnStartedAt === null || this.activeForegroundTurnId !== turnId) {
+        return null;
+      }
+
+      const completion = await readOpenCodePersistedAssistantCompletion(
+        this.storageRoot,
+        this.sessionId,
+        this.foregroundKnownMessageIds,
+        this.foregroundTurnStartedAt,
+      );
+      if (completion) {
+        return completion;
+      }
+
+      if (attempt < this.eofRecoveryPolicy.maxAttempts - 1) {
+        await sleep(this.eofRecoveryPolicy.delayMs);
+      }
+    }
+
+    return null;
+  }
+
+  private resolvePersistedAssistantRecoveryText(completedText: string): string | null {
+    if (!this.foregroundAssistantMessageEmitted) {
+      return completedText;
+    }
+
+    if (completedText === this.foregroundAssistantText) {
+      return "";
+    }
+
+    return completedText.startsWith(this.foregroundAssistantText)
+      ? completedText.slice(this.foregroundAssistantText.length)
+      : null;
+  }
+
+  private async readPersistedSessionMessageIds(): Promise<Set<string>> {
+    const messageRoot = path.join(this.storageRoot, "message", this.sessionId);
+    const messageFiles = await findJsonFiles(messageRoot);
+    const messageIds = new Set<string>();
+
+    for (const file of messageFiles) {
+      const parsed = await readJsonFile(file, OpenCodeStoredMessageSchema);
+      if (parsed?.sessionID === this.sessionId) {
+        messageIds.add(parsed.id);
+      }
+    }
+
+    return messageIds;
+  }
+
   private finishForegroundTurn(
     event: Extract<AgentStreamEvent, { type: "turn_completed" | "turn_failed" | "turn_canceled" }>,
     turnId: string,
@@ -2516,6 +2816,7 @@ class OpenCodeAgentSession implements AgentSession {
     } else {
       this.runningToolCalls.clear();
     }
+    this.foregroundTurnStartedAt = null;
     this.activeForegroundTurnId = null;
     // Abort the SSE connection so the SDK tears down the underlying fetch.
     this.abortController?.abort();
@@ -2561,6 +2862,13 @@ class OpenCodeAgentSession implements AgentSession {
 
   private notifySubscribers(event: AgentStreamEvent, turnIdOverride?: string): void {
     const turnId = turnIdOverride ?? this.activeForegroundTurnId;
+    if (event.type === "timeline" && event.item.type === "assistant_message") {
+      this.foregroundAssistantMessageEmitted = true;
+      this.foregroundAssistantText += event.item.text;
+    }
+    if (event.type === "usage_updated") {
+      this.foregroundUsageUpdated = true;
+    }
     const tagged = turnId ? { ...event, turnId } : event;
     for (const callback of this.subscribers) {
       try {