fix(sessions): stop agent-crash recovery from wiping task history

packages/core/src/sessions/sessionService.ts

@@ -22,6 +22,7 @@ import {
   isJsonRpcRequest,
   isJsonRpcResponse,
   isRateLimitError,
+  isTransientUpstreamError,
   mergeConfigOptions,
   type OptimisticItem,
   type PermissionRequest,
@@ -903,7 +904,11 @@ export class SessionService {
     const previous = this.d.store.getSessions()[taskRunId];
 
     const session = createBaseSession(taskRunId, taskId, taskTitle);
-    session.events = events;
+    // Repainting from the log must not blank a transcript we already hold:
+    // fetchSessionLogs swallows read errors and returns empty, so keep the
+    // previous in-memory events when the log read produced nothing.
+    session.events =
+      events.length === 0 && previous?.events.length ? previous.events : events;
     if (logUrl) {
       session.logUrl = logUrl;
     }
@@ -2460,6 +2465,21 @@ export class SessionService {
         });
       }
 
+      // A provider request that timed out or dropped leaves the session
+      // healthy — no recovery ran above — so tell the user to just re-send
+      // instead of surfacing the raw "Internal error: API Error: …" text.
+      if (isTransientUpstreamError(errorMessage, errorDetails)) {
+        this.d.log.warn("Transient upstream provider failure during prompt", {
+          taskRunId: session.taskRunId,
+          errorMessage,
+          errorDetails,
+        });
+        throw new Error(
+          "The AI provider timed out or dropped the connection. Your session is unaffected — please send the message again.",
+          { cause: error },
+        );
+      }
+
       throw error;
     }
   }
@@ -3392,14 +3412,17 @@ export class SessionService {
    * effect loop.
    *
    * If the session failed before any conversation started (has an
-   * initialPrompt saved from the original creation attempt), creates
-   * a fresh session and re-sends the prompt instead of reconnecting
-   * to an empty session.
+   * initialPrompt saved from the original creation attempt, and no
+   * conversation in memory or in the run log), creates a fresh session
+   * and re-sends the prompt instead of reconnecting to an empty session.
    */
   async clearSessionError(taskId: string, repoPath: string): Promise<void> {
     this.localRepoPaths.set(taskId, repoPath);
     const session = this.d.store.getSessionByTaskId(taskId);
-    if (session?.initialPrompt?.length) {
+    if (
+      session?.initialPrompt?.length &&
+      !(await this.runHasConversationHistory(session))
+    ) {
       const {
         taskTitle,
         initialPrompt,
@@ -3434,6 +3457,34 @@ export class SessionService {
     await this.reconnectInPlace(taskId, repoPath);
   }
 
+  /**
+   * Whether the run already holds conversation beyond the user's prompt
+   * echoes. A set `initialPrompt` alone doesn't prove the conversation never
+   * started: it is only cleared when a live agent event arrives, so it
+   * survives when the event subscription drops before the first agent event
+   * while the agent keeps working and logging, and error sessions carry it
+   * forward. Recreating the run in that state orphans the populated run log
+   * behind a fresh latest_run — the task's entire history disappears — so
+   * check the in-memory transcript first and fall back to the persisted log.
+   */
+  private async runHasConversationHistory(
+    session: AgentSession,
+  ): Promise<boolean> {
+    const isPromptEcho = (event: AcpMessage): boolean =>
+      isJsonRpcRequest(event.message) &&
+      event.message.method === "session/prompt";
+    if (session.events.some((event) => !isPromptEcho(event))) {
+      return true;
+    }
+    const { rawEntries } = await this.fetchSessionLogs(
+      session.logUrl,
+      session.taskRunId,
+    );
+    return convertStoredEntriesToEvents(rawEntries).some(
+      (event) => !isPromptEcho(event),
+    );
+  }
+
   /**
    * Start a fresh session for a task, abandoning the old conversation.
    * Clears the backend sessionId so the next reconnect creates a new

packages/core/src/sessions/sessionServiceRetryConfig.test.ts

@@ -1,4 +1,4 @@
-import type { AgentSession } from "@posthog/shared";
+import type { AcpMessage, AgentSession } from "@posthog/shared";
 import type { Task } from "@posthog/shared/domain-types";
 import { describe, expect, it, vi } from "vitest";
 import {
@@ -7,6 +7,32 @@ import {
   type SessionServiceDeps,
 } from "./sessionService";
 
+const PROMPT_ECHO_EVENT: AcpMessage = {
+  type: "acp_message",
+  ts: 1,
+  message: {
+    jsonrpc: "2.0",
+    id: 1,
+    method: "session/prompt",
+    params: { prompt: [{ type: "text", text: "Ship the fix" }] },
+  } as AcpMessage["message"],
+};
+
+const AGENT_MESSAGE_EVENT: AcpMessage = {
+  type: "acp_message",
+  ts: 2,
+  message: {
+    jsonrpc: "2.0",
+    method: "session/update",
+    params: {
+      update: {
+        sessionUpdate: "agent_message_chunk",
+        content: { type: "text", text: "Working on it" },
+      },
+    },
+  } as AcpMessage["message"],
+};
+
 function makeSession(overrides: Partial<AgentSession> = {}): AgentSession {
   return {
     taskRunId: "run-1",
@@ -66,8 +92,28 @@ function createHarness(session: AgentSession) {
       "createNewLocalSession",
     )
     .mockResolvedValue(undefined);
+  const reconnectInPlace = vi
+    .spyOn(
+      service as unknown as {
+        reconnectInPlace: (...args: unknown[]) => Promise<boolean>;
+      },
+      "reconnectInPlace",
+    )
+    .mockResolvedValue(true);
+  const fetchSessionLogs = vi
+    .spyOn(
+      service as unknown as {
+        fetchSessionLogs: (...args: unknown[]) => Promise<unknown>;
+      },
+      "fetchSessionLogs",
+    )
+    .mockResolvedValue({
+      rawEntries: [],
+      totalLineCount: 0,
+      parseFailureCount: 0,
+    });
 
-  return { service, createNewLocalSession };
+  return { service, createNewLocalSession, reconnectInPlace, fetchSessionLogs };
 }
 
 describe("SessionService.clearSessionError retry config", () => {
@@ -94,6 +140,56 @@ describe("SessionService.clearSessionError retry config", () => {
       "high", // reasoningLevel
     );
   });
+
+  it("reconnects in place instead of recreating when the transcript has agent events", async () => {
+    const session = makeSession({
+      events: [PROMPT_ECHO_EVENT, AGENT_MESSAGE_EVENT],
+    });
+    const { service, createNewLocalSession, reconnectInPlace } =
+      createHarness(session);
+
+    await service.clearSessionError("task-1", "/repo");
+
+    expect(createNewLocalSession).not.toHaveBeenCalled();
+    expect(reconnectInPlace).toHaveBeenCalledWith("task-1", "/repo");
+  });
+
+  it("reconnects in place when the run log has history even if in-memory events are empty", async () => {
+    const session = makeSession({ events: [] });
+    const {
+      service,
+      createNewLocalSession,
+      reconnectInPlace,
+      fetchSessionLogs,
+    } = createHarness(session);
+    fetchSessionLogs.mockResolvedValue({
+      rawEntries: [
+        {
+          type: "notification",
+          timestamp: "2026-07-06T00:00:00.000Z",
+          notification: AGENT_MESSAGE_EVENT.message,
+        },
+      ],
+      totalLineCount: 1,
+      parseFailureCount: 0,
+    });
+
+    await service.clearSessionError("task-1", "/repo");
+
+    expect(createNewLocalSession).not.toHaveBeenCalled();
+    expect(reconnectInPlace).toHaveBeenCalledWith("task-1", "/repo");
+  });
+
+  it("recreates when the transcript holds only the user's prompt echo", async () => {
+    const session = makeSession({ events: [PROMPT_ECHO_EVENT] });
+    const { service, createNewLocalSession, reconnectInPlace } =
+      createHarness(session);
+
+    await service.clearSessionError("task-1", "/repo");
+
+    expect(createNewLocalSession).toHaveBeenCalled();
+    expect(reconnectInPlace).not.toHaveBeenCalled();
+  });
 });
 
 const CONNECT_PARAMS: ConnectParams = {

packages/shared/src/errors.test.ts

@@ -5,6 +5,7 @@ import {
   isFatalSessionError,
   isNotAuthenticatedError,
   isRateLimitError,
+  isTransientUpstreamError,
   NotAuthenticatedError,
   serializeError,
 } from "./errors";
@@ -103,6 +104,56 @@ describe("isFatalSessionError", () => {
   it("returns false for ordinary recoverable errors", () => {
     expect(isFatalSessionError("temporary network blip")).toBe(false);
   });
+
+  it.each([
+    "Internal error: API Error: the operation timed out",
+    "Internal error: API Error: Request timeout",
+    "Internal error: API Error: terminated",
+    "Internal error: API Error: Connection error",
+    "Internal error: API Error: 529 overloaded_error",
+  ])("does not treat the transient upstream failure %j as fatal", (message) => {
+    expect(isFatalSessionError(message)).toBe(false);
+  });
+
+  it("does not treat a transient upstream failure in the details as fatal", () => {
+    expect(
+      isFatalSessionError(
+        "internal error",
+        "API Error: the operation timed out",
+      ),
+    ).toBe(false);
+  });
+});
+
+describe("isTransientUpstreamError", () => {
+  it.each([
+    "API Error: the operation timed out",
+    "API Error: terminated",
+    "API Error: Connection error",
+    "API Error: 500 internal server error",
+    "API Error: 529 overloaded_error",
+    "Internal error: API Error: request timed out",
+  ])("recognises %j", (message) => {
+    expect(isTransientUpstreamError(message)).toBe(true);
+  });
+
+  it("matches against the details when the message is generic", () => {
+    expect(
+      isTransientUpstreamError(
+        "Internal error",
+        "API Error: the operation timed out",
+      ),
+    ).toBe(true);
+  });
+
+  it.each([
+    "process exited",
+    "session not found",
+    "the operation timed out", // no "API Error:" marker — not an upstream turn failure
+    "API Error: 400 invalid_request_error",
+  ])("does not match %j", (message) => {
+    expect(isTransientUpstreamError(message)).toBe(false);
+  });
 });
 
 describe("serializeError", () => {

packages/shared/src/errors.ts

@@ -82,6 +82,22 @@ const FATAL_SESSION_ERROR_PATTERNS = [
   "session not found",
 ] as const;
 
+/**
+ * Transient upstream provider failures, as surfaced by agent adapters in
+ * "API Error: …" result strings (kept in sync with classifyAgentError in
+ * @posthog/agent). The agent process and session are healthy — a single
+ * provider request timed out, dropped, or returned a retryable status — so
+ * these must not count as fatal session errors: the fix is re-sending the
+ * prompt, never tearing the session down. Checked before the fatal patterns
+ * because the ACP layer wraps them as "Internal error: API Error: …".
+ */
+const UPSTREAM_TRANSIENT_ERROR_REGEXES = [
+  /API Error:\s*terminated\b/i,
+  /API Error:\s*Connection error\b/i,
+  /API Error:.*\b(?:timed out|timeout)\b/i,
+  /API Error:\s*(?:429|5\d\d)\b/i,
+] as const;
+
 function includesAny(
   value: string | undefined,
   patterns: readonly string[],
@@ -101,11 +117,22 @@ export function isRateLimitError(
   );
 }
 
+export function isTransientUpstreamError(
+  errorMessage: string,
+  errorDetails?: string,
+): boolean {
+  return UPSTREAM_TRANSIENT_ERROR_REGEXES.some(
+    (regex) =>
+      regex.test(errorMessage) || (!!errorDetails && regex.test(errorDetails)),
+  );
+}
+
 export function isFatalSessionError(
   errorMessage: string,
   errorDetails?: string,
 ): boolean {
   if (isRateLimitError(errorMessage, errorDetails)) return false;
+  if (isTransientUpstreamError(errorMessage, errorDetails)) return false;
   return (
     includesAny(errorMessage, FATAL_SESSION_ERROR_PATTERNS) ||
     includesAny(errorDetails, FATAL_SESSION_ERROR_PATTERNS)

packages/shared/src/index.ts

@@ -75,6 +75,7 @@ export {
   isFatalSessionError,
   isNotAuthenticatedError,
   isRateLimitError,
+  isTransientUpstreamError,
   NotAuthenticatedError,
   type SerializedError,
   serializeError,