fix(sessions): recover local agent turns interrupted by system sleep

packages/core/src/sessions/sessionEvents.ts

@@ -358,7 +358,11 @@ export function normalizePromptToBlocks(
   );
 }
 
-export { isFatalSessionError, isRateLimitError } from "@posthog/shared";
+export {
+  isFatalSessionError,
+  isRateLimitError,
+  isTransportError,
+} from "@posthog/shared";
 
 /**
  * Whether a list of events already contains a `session/prompt` request.

packages/core/src/sessions/sessionService.ts

@@ -22,6 +22,7 @@ import {
   isJsonRpcRequest,
   isJsonRpcResponse,
   isRateLimitError,
+  isTransportError,
   mergeConfigOptions,
   type OptimisticItem,
   type PermissionRequest,
@@ -89,7 +90,7 @@ const LOCAL_SESSION_RECONNECT_BACKOFF = {
 const LOCAL_SESSION_RECOVERY_MESSAGE =
   "Lost connection to the agent. Reconnecting…";
 const LOCAL_SESSION_RECOVERY_FAILED_MESSAGE =
-  "Connecting to to the agent has been lost. Retry, or start a new session.";
+  "Connection to the agent has been lost. Retry, or start a new session.";
 const GITHUB_AUTHORIZATION_REQUIRED_CODE = "github_authorization_required";
 const AUTO_RETRY_MAX_ATTEMPTS = 2;
 const AUTO_RETRY_DELAY_MS = 10_000;
@@ -152,6 +153,9 @@ export interface SessionTrpc {
     onSessionEvent: TrpcSubscription;
     onPermissionRequest: TrpcSubscription;
     onSessionIdleKilled: TrpcSubscription;
+    /** Optional: only hosts with a power manager detect turns stalled by
+     * system sleep. Core subscribes when the host exposes it. */
+    onSessionStalled?: TrpcSubscription;
   };
   workspace: { verify: TrpcQuery };
   cloudTask: {
@@ -559,6 +563,7 @@ export class SessionService {
     { startedAtTs: number; agentTextChunks: number; agentOutputEvents: number }
   >();
   private idleKilledSubscription: { unsubscribe: () => void } | null = null;
+  private stalledTurnSubscription: { unsubscribe: () => void } | null = null;
   /**
    * Cached preview-config-options responses keyed by `${apiHost}::${adapter}`.
    * Shared across cloud sessions so switching model/adapter reuses the list.
@@ -612,6 +617,38 @@ export class SessionService {
         },
       },
     );
+    this.stalledTurnSubscription =
+      d.trpc.agent.onSessionStalled?.subscribe(undefined, {
+        onData: (event: { taskRunId: string; taskId: string }) => {
+          this.handleStalledTurn(event.taskId, event.taskRunId);
+        },
+        onError: (err: unknown) => {
+          d.log.debug("Stalled-turn subscription error", { error: err });
+        },
+      }) ?? null;
+  }
+
+  /**
+   * The main process detected a turn that produced no agent traffic after a
+   * system resume: the agent is wedged on a connection that died during
+   * sleep. Reconnect in place — the resumed session keeps its history.
+   */
+  private handleStalledTurn(taskId: string, taskRunId: string): void {
+    const session = this.d.store.getSessionByTaskId(taskId);
+    if (!session || session.taskRunId !== taskRunId || session.isCloud) {
+      return;
+    }
+    this.d.log.warn("Recovering turn stalled by system sleep", {
+      taskId,
+      taskRunId,
+    });
+    this.startAutoRecoverLocalSession(
+      taskId,
+      taskRunId,
+      session.taskTitle,
+      "Turn stalled after system sleep",
+      "The agent lost its connection while the computer was asleep. Please retry or start a new session.",
+    );
   }
 
   /**
@@ -976,6 +1013,16 @@ export class SessionService {
       });
 
       if (result) {
+        if (sessionId && result.resumedExistingSession === false) {
+          this.d.log.warn(
+            "Reconnect could not restore prior agent context; continuing with a fresh session",
+            { taskId, taskRunId },
+          );
+          this.d.toast.info(
+            "Couldn't restore the agent's previous context, so it may not remember earlier turns. The transcript is unaffected.",
+          );
+        }
+
         // Cast and merge live configOptions with persisted values.
         // Fall back to persisted options if the agent doesn't return any
         // (e.g. after session compaction).
@@ -1237,8 +1284,16 @@ export class SessionService {
     reason: string,
     fallbackMessage: string,
   ): void {
-    void this.tryAutoRecoverLocalSession(taskId, taskRunId, reason).then(
-      (recovered) => {
+    void this.tryAutoRecoverLocalSession(taskId, taskRunId, reason)
+      .catch((error) => {
+        this.d.log.error("Local session recovery threw", {
+          taskId,
+          taskRunId,
+          error,
+        });
+        return false;
+      })
+      .then((recovered) => {
         if (recovered) {
           return;
         }
@@ -1257,8 +1312,7 @@ export class SessionService {
             "Connection lost",
           );
         }
-      },
-    );
+      });
   }
 
   private async createNewLocalSession(
@@ -1686,6 +1740,8 @@ export class SessionService {
     this.cloudRunIdleTracker.clear();
     this.idleKilledSubscription?.unsubscribe();
     this.idleKilledSubscription = null;
+    this.stalledTurnSubscription?.unsubscribe();
+    this.stalledTurnSubscription = null;
   }
 
   /**
@@ -2436,8 +2492,11 @@ export class SessionService {
         return { stopReason: "rate_limited" };
       }
 
-      if (isFatalSessionError(errorMessage, errorDetails)) {
-        this.d.log.error("Fatal prompt error, attempting recovery", {
+      if (
+        isFatalSessionError(errorMessage, errorDetails) ||
+        isTransportError(errorMessage, errorDetails)
+      ) {
+        this.d.log.error("Prompt error, attempting session recovery", {
           taskRunId: session.taskRunId,
           errorMessage,
           errorDetails,

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

@@ -48,6 +48,9 @@ function createHarness({ spyConnect = true } = {}) {
     updateSession: vi.fn(),
   };
   const log = { info: vi.fn(), warn: vi.fn(), error: vi.fn(), debug: vi.fn() };
+  let emitStalledTurn:
+    | ((event: { taskRunId: string; taskId: string }) => void)
+    | undefined;
   const deps = {
     store,
     log,
@@ -65,6 +68,17 @@ function createHarness({ spyConnect = true } = {}) {
         onSessionIdleKilled: {
           subscribe: () => ({ unsubscribe: vi.fn() }),
         },
+        onSessionStalled: {
+          subscribe: (
+            _input: unknown,
+            handlers: {
+              onData: (event: { taskRunId: string; taskId: string }) => void;
+            },
+          ) => {
+            emitStalledTurn = handlers.onData;
+            return { unsubscribe: vi.fn() };
+          },
+        },
       },
     },
   } as unknown as SessionServiceDeps;
@@ -73,7 +87,16 @@ function createHarness({ spyConnect = true } = {}) {
   const connectToTask = spyConnect
     ? vi.spyOn(service, "connectToTask").mockResolvedValue(undefined)
     : undefined;
-  return { service, sessions, connectToTask, log };
+  return {
+    service,
+    sessions,
+    connectToTask,
+    log,
+    emitStalledTurn: (event: { taskRunId: string; taskId: string }) => {
+      if (!emitStalledTurn) throw new Error("Stalled-turn handler not wired");
+      emitStalledTurn(event);
+    },
+  };
 }
 
 function reconcile(
@@ -188,3 +211,59 @@ describe("SessionService run-less local task recovery", () => {
     expect(connectToTask).toHaveBeenCalledWith({ task, repoPath: "/repo" });
   });
 });
+
+describe("SessionService stalled-turn recovery", () => {
+  it("auto-recovers a local session when the main process reports a stalled turn", async () => {
+    const { service, sessions, emitStalledTurn } = createHarness();
+    sessions["run-task-1"] = makeSession("task-1");
+    (
+      service as unknown as { localRepoPaths: Map<string, string> }
+    ).localRepoPaths.set("task-1", "/repo");
+    const reconnectInPlace = vi
+      .spyOn(
+        service as unknown as {
+          reconnectInPlace: (taskId: string, repoPath: string) => unknown;
+        },
+        "reconnectInPlace",
+      )
+      .mockResolvedValue(true);
+
+    emitStalledTurn({ taskId: "task-1", taskRunId: "run-task-1" });
+
+    await vi.waitFor(() =>
+      expect(reconnectInPlace).toHaveBeenCalledWith("task-1", "/repo"),
+    );
+  });
+
+  it("ignores stalled-turn reports for cloud sessions", async () => {
+    const { service, sessions, emitStalledTurn } = createHarness();
+    sessions["run-task-1"] = { ...makeSession("task-1"), isCloud: true };
+    const reconnectInPlace = vi.spyOn(
+      service as unknown as {
+        reconnectInPlace: (taskId: string, repoPath: string) => unknown;
+      },
+      "reconnectInPlace",
+    );
+
+    emitStalledTurn({ taskId: "task-1", taskRunId: "run-task-1" });
+    await Promise.resolve();
+
+    expect(reconnectInPlace).not.toHaveBeenCalled();
+  });
+
+  it("ignores stalled-turn reports for superseded runs", async () => {
+    const { service, sessions, emitStalledTurn } = createHarness();
+    sessions["run-task-1"] = makeSession("task-1");
+    const reconnectInPlace = vi.spyOn(
+      service as unknown as {
+        reconnectInPlace: (taskId: string, repoPath: string) => unknown;
+      },
+      "reconnectInPlace",
+    );
+
+    emitStalledTurn({ taskId: "task-1", taskRunId: "run-old" });
+    await Promise.resolve();
+
+    expect(reconnectInPlace).not.toHaveBeenCalled();
+  });
+});

packages/host-router/src/routers/agent.router.ts

@@ -200,6 +200,16 @@ export const agentRouter = router({
     }
   }),
 
+  onSessionStalled: publicProcedure.subscription(async function* (opts) {
+    const service = opts.ctx.container.get<AgentService>(AGENT_SERVICE);
+    for await (const event of service.toIterable(
+      AgentServiceEvent.SessionStalled,
+      { signal: opts.signal },
+    )) {
+      yield event;
+    }
+  }),
+
   onAgentFileActivity: publicProcedure.subscription(async function* (opts) {
     const service = opts.ctx.container.get<AgentService>(AGENT_SERVICE);
     for await (const event of service.toIterable(

packages/shared/src/errors.test.ts

@@ -5,6 +5,7 @@ import {
   isFatalSessionError,
   isNotAuthenticatedError,
   isRateLimitError,
+  isTransportError,
   NotAuthenticatedError,
   serializeError,
 } from "./errors";
@@ -105,6 +106,38 @@ describe("isFatalSessionError", () => {
   });
 });
 
+describe("isTransportError", () => {
+  it.each([
+    "read ECONNRESET",
+    "connect ECONNREFUSED 127.0.0.1:443",
+    "write EPIPE",
+    "request timed out: ETIMEDOUT",
+    "getaddrinfo ENOTFOUND api.anthropic.com",
+    "socket hang up",
+    "TypeError: fetch failed",
+    "Connection reset by peer",
+    "stream closed unexpectedly",
+    "UND_ERR_SOCKET: other side closed",
+  ])("treats %j as a transport error", (message) => {
+    expect(isTransportError(message)).toBe(true);
+  });
+
+  it("matches in details when the message is generic", () => {
+    expect(isTransportError("Unknown error", "read ECONNRESET")).toBe(true);
+  });
+
+  it("does not treat a rate-limit error as a transport error", () => {
+    expect(isTransportError("network error", "rate limit exceeded")).toBe(
+      false,
+    );
+  });
+
+  it("returns false for non-transport errors", () => {
+    expect(isTransportError("invalid prompt")).toBe(false);
+    expect(isTransportError("")).toBe(false);
+  });
+});
+
 describe("serializeError", () => {
   it("captures name, message and code from an Error", () => {
     const err = Object.assign(new TypeError("boom"), { code: "ERR_X" });

packages/shared/src/errors.ts

@@ -82,6 +82,29 @@ const FATAL_SESSION_ERROR_PATTERNS = [
   "session not found",
 ] as const;
 
+// Network/socket failures cut a turn short without killing the agent process —
+// typical after system sleep resets the agent's upstream connection. They are
+// recoverable by reconnecting the session, unlike FATAL_SESSION_ERROR_PATTERNS
+// which indicate the agent itself is gone.
+const TRANSPORT_ERROR_PATTERNS = [
+  "econnreset",
+  "econnrefused",
+  "epipe",
+  "etimedout",
+  "enetdown",
+  "enetunreach",
+  "ehostunreach",
+  "enotfound",
+  "socket hang up",
+  "network error",
+  "fetch failed",
+  "connection reset",
+  "connection closed",
+  "connection error",
+  "stream closed",
+  "other side closed",
+] as const;
+
 function includesAny(
   value: string | undefined,
   patterns: readonly string[],
@@ -111,3 +134,14 @@ export function isFatalSessionError(
     includesAny(errorDetails, FATAL_SESSION_ERROR_PATTERNS)
   );
 }
+
+export function isTransportError(
+  errorMessage: string,
+  errorDetails?: string,
+): boolean {
+  if (isRateLimitError(errorMessage, errorDetails)) return false;
+  return (
+    includesAny(errorMessage, TRANSPORT_ERROR_PATTERNS) ||
+    includesAny(errorDetails, TRANSPORT_ERROR_PATTERNS)
+  );
+}

packages/shared/src/index.ts

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