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

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

@@ -566,6 +566,321 @@ 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),
+      },
+    });
+
+    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("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),
+      },
+    });
+
+    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),
+      },
+    });
+
+    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 +895,7 @@ describe("OpenCode adapter startTurn error handling", () => {
       fakeClient,
       "ses_unit_test",
       createTestLogger(),
+      "/tmp/opencode-storage",
       new Map(),
       undefined,
       false,
@@ -607,6 +923,7 @@ describe("OpenCode adapter startTurn error handling", () => {
       fakeClient,
       "ses_unit_test",
       createTestLogger(),
+      "/tmp/opencode-storage",
     );
 
     await session.close();
@@ -641,6 +958,7 @@ describe("OpenCode adapter startTurn error handling", () => {
       fakeClient,
       "ses_unit_test",
       createTestLogger(),
+      "/tmp/opencode-storage",
     );
 
     const events: AgentStreamEvent[] = [];

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

@@ -138,6 +138,12 @@ 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;
+}
+
 type OpenCodeAgentConfig = AgentSessionConfig & { provider: "opencode" };
 type OpenCodeMessageRole = "user" | "assistant";
 
@@ -818,23 +824,133 @@ 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 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 +996,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,
@@ -984,6 +1127,7 @@ export class OpenCodeAgentClient implements AgentClient {
         client,
         session.id,
         this.logger,
+        this.storageRoot,
         new Map(this.modelContextWindows),
         acquisition.release,
         options?.persistSession,
@@ -1025,6 +1169,7 @@ export class OpenCodeAgentClient implements AgentClient {
         client,
         handle.sessionId,
         this.logger,
+        this.storageRoot,
         new Map(this.modelContextWindows),
         acquisition.release,
       );
@@ -2130,6 +2275,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>();
@@ -2157,11 +2303,17 @@ class OpenCodeAgentSession implements AgentSession {
   private releaseServer: (() => void) | null;
   private readonly persistSession: boolean;
   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,
@@ -2170,6 +2322,7 @@ class OpenCodeAgentSession implements AgentSession {
     this.client = client;
     this.sessionId = sessionId;
     this.logger = logger;
+    this.storageRoot = storageRoot;
     this.modelContextWindowsByModelKey = modelContextWindowsByModelKey;
     this.currentMode = normalizeOpenCodeModeId(config.modeId);
     this.releaseServer = releaseServer ?? null;
@@ -2243,10 +2396,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 +2624,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 +2665,102 @@ class OpenCodeAgentSession implements AgentSession {
     }
   }
 
+  private async recoverTurnFromPersistedCompletion(turnId: string): Promise<boolean> {
+    if (this.foregroundTurnStartedAt === null) {
+      return false;
+    }
+
+    const completion = await readOpenCodePersistedAssistantCompletion(
+      this.storageRoot,
+      this.sessionId,
+      this.foregroundKnownMessageIds,
+      this.foregroundTurnStartedAt,
+    );
+    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 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 +2773,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 +2819,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 {