fix(acpx): retry persisted resume ids cleanly (#52209)

* fix(acpx): store agent session ID when session/load fails

When an ACP agent (e.g. Gemini CLI) rejects the acpx-generated session
ID via session/load and falls back to session/new, the agent-returned
session ID was previously discarded. This caused identity stuck at
pending forever, multi-turn failures, lost completion events, and
persistent reconcile warnings.

- Parse ACP protocol stream in runTurn() to capture agent session IDs
- Flip resolveRuntimeResumeSessionId() to prefer agentSessionId
- Add createIdentityFromHandleEvent() for handle-sourced identity
- Layer handle event identity before status in reconcile
- Add regression tests for load fallback and restart resume

Closes #52182

* ACPX: prefer decoded session ids

* ACPX: refresh runtime handle state from status

---------

Co-authored-by: Wesley <imwyvern@users.noreply.github.com>

src/acp/control-plane/manager.core.ts

@@ -1354,6 +1354,7 @@ export class AcpSessionManager {
     const runtime = backend.runtime;
     const previousMeta = params.meta;
     const previousIdentity = resolveSessionIdentityFromMeta(previousMeta);
+    let identityForEnsure = previousIdentity;
     const persistedResumeSessionId =
       mode === "persistent" ? resolveRuntimeResumeSessionId(previousIdentity) : undefined;
     const ensureSession = async (resumeSessionId?: string) =>
@@ -1385,6 +1386,19 @@ export class AcpSessionManager {
         logVerbose(
           `acp-manager: resume init failed for ${params.sessionKey}; retrying without persisted ACP session id: ${acpError.message}`,
         );
+        if (identityForEnsure) {
+          const {
+            acpxSessionId: _staleAcpxSessionId,
+            agentSessionId: _staleAgentSessionId,
+            ...retryIdentity
+          } = identityForEnsure;
+          // The persisted resume identifiers already failed, so do not merge them back into the
+          // fresh named-session handle returned by the retry path.
+          identityForEnsure = {
+            ...retryIdentity,
+            state: "pending",
+          };
+        }
         ensured = await ensureSession();
       }
     } else {
@@ -1399,13 +1413,13 @@ export class AcpSessionManager {
     });
     const nextIdentity =
       mergeSessionIdentity({
-        current: previousIdentity,
+        current: identityForEnsure,
         incoming: createIdentityFromEnsure({
           handle: ensured,
           now,
         }),
         now,
-      }) ?? previousIdentity;
+      }) ?? identityForEnsure;
     const nextHandleIdentifiers = resolveRuntimeHandleIdentifiersFromIdentity(nextIdentity);
     const nextHandle: AcpRuntimeHandle = {
       ...ensured,

src/acp/control-plane/manager.identity-reconcile.ts

@@ -2,6 +2,7 @@ import type { OpenClawConfig } from "../../config/config.js";
 import { logVerbose } from "../../globals.js";
 import { withAcpRuntimeErrorBoundary } from "../runtime/errors.js";
 import {
+  createIdentityFromHandleEvent,
   createIdentityFromStatus,
   identityEquals,
   mergeSessionIdentity,
@@ -63,15 +64,25 @@ export async function reconcileManagerRuntimeSessionIdentifiers(params: {
 
   const now = Date.now();
   const currentIdentity = resolveSessionIdentityFromMeta(params.meta);
-  const nextIdentity =
+  const eventIdentity = createIdentityFromHandleEvent({
+    handle: params.handle,
+    now,
+  });
+  const identityAfterEvent =
     mergeSessionIdentity({
       current: currentIdentity,
+      incoming: eventIdentity,
+      now,
+    }) ?? currentIdentity;
+  const nextIdentity =
+    mergeSessionIdentity({
+      current: identityAfterEvent,
       incoming: createIdentityFromStatus({
         status: runtimeStatus,
         now,
       }),
       now,
-    }) ?? currentIdentity;
+    }) ?? identityAfterEvent;
   const handleIdentifiers = resolveRuntimeHandleIdentifiersFromIdentity(nextIdentity);
   const handleChanged =
     handleIdentifiers.backendSessionId !== params.handle.backendSessionId ||

src/acp/control-plane/manager.test.ts

@@ -843,6 +843,51 @@ describe("AcpSessionManager", () => {
     );
   });
 
+  it("prefers the persisted agent session id when reopening an ACP runtime after restart", async () => {
+    const runtimeState = createRuntime();
+    hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
+      id: "acpx",
+      runtime: runtimeState.runtime,
+    });
+    const sessionKey = "agent:gemini:acp:binding:discord:default:restart";
+    hoisted.readAcpSessionEntryMock.mockImplementation((paramsUnknown: unknown) => {
+      const key = (paramsUnknown as { sessionKey?: string }).sessionKey ?? sessionKey;
+      return {
+        sessionKey: key,
+        storeSessionKey: key,
+        acp: {
+          ...readySessionMeta(),
+          agent: "gemini",
+          runtimeSessionName: key,
+          identity: {
+            state: "resolved",
+            source: "status",
+            acpxSessionId: "acpx-sid-1",
+            agentSessionId: "gemini-sid-1",
+            lastUpdatedAt: Date.now(),
+          },
+        },
+      };
+    });
+
+    const manager = new AcpSessionManager();
+    await manager.runTurn({
+      cfg: baseCfg,
+      sessionKey,
+      text: "after restart",
+      mode: "prompt",
+      requestId: "r-binding-restart-gemini",
+    });
+
+    expect(runtimeState.ensureSession).toHaveBeenCalledWith(
+      expect.objectContaining({
+        sessionKey,
+        agent: "gemini",
+        resumeSessionId: "gemini-sid-1",
+      }),
+    );
+  });
+
   it("does not resume persisted ACP identity for oneshot sessions after restart", async () => {
     const runtimeState = createRuntime();
     hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
@@ -890,7 +935,7 @@ describe("AcpSessionManager", () => {
     expect(ensureInput?.resumeSessionId).toBeUndefined();
   });
 
-  it("falls back to a fresh ensure when reopening a persisted ACP backend session id fails", async () => {
+  it("falls back to a fresh ensure without reusing stale agent session ids", async () => {
     const runtimeState = createRuntime();
     runtimeState.ensureSession.mockImplementation(async (inputUnknown: unknown) => {
       const input = inputUnknown as {
@@ -929,6 +974,7 @@ describe("AcpSessionManager", () => {
         state: "resolved",
         source: "status",
         acpxSessionId: "acpx-sid-stale",
+        agentSessionId: "agent-sid-stale",
         lastUpdatedAt: Date.now(),
       },
     };
@@ -971,13 +1017,19 @@ describe("AcpSessionManager", () => {
     expect(runtimeState.ensureSession.mock.calls[0]?.[0]).toMatchObject({
       sessionKey,
       agent: "codex",
-      resumeSessionId: "acpx-sid-stale",
+      resumeSessionId: "agent-sid-stale",
     });
     const retryInput = runtimeState.ensureSession.mock.calls[1]?.[0] as
       | { resumeSessionId?: string }
       | undefined;
     expect(retryInput?.resumeSessionId).toBeUndefined();
+    const runTurnInput = runtimeState.runTurn.mock.calls[0]?.[0] as
+      | { handle?: { agentSessionId?: string; backendSessionId?: string } }
+      | undefined;
+    expect(runTurnInput?.handle?.backendSessionId).toBe("acpx-sid-fresh");
+    expect(runTurnInput?.handle?.agentSessionId).toBeUndefined();
     expect(currentMeta.identity?.acpxSessionId).toBe("acpx-sid-fresh");
+    expect(currentMeta.identity?.agentSessionId).toBeUndefined();
   });
 
   it("enforces acp.maxConcurrentSessions when opening new runtime handles", async () => {
@@ -1812,6 +1864,83 @@ describe("AcpSessionManager", () => {
     expect(currentMeta.identity?.agentSessionId).toBe("agent-session-1");
   });
 
+  it("reconciles prompt-learned agent session IDs even when runtime status omits them", async () => {
+    const runtimeState = createRuntime();
+    runtimeState.ensureSession.mockResolvedValue({
+      sessionKey: "agent:gemini:acp:session-1",
+      backend: "acpx",
+      runtimeSessionName: "runtime-3",
+      backendSessionId: "acpx-stale",
+    });
+    runtimeState.runTurn.mockImplementation(async function* (inputUnknown: unknown) {
+      const input = inputUnknown as {
+        handle: {
+          agentSessionId?: string;
+        };
+      };
+      input.handle.agentSessionId = "gemini-session-1";
+      yield { type: "done" as const };
+    });
+    runtimeState.getStatus.mockResolvedValue({
+      summary: "status=alive",
+      details: { status: "alive" },
+    });
+    hoisted.requireAcpRuntimeBackendMock.mockReturnValue({
+      id: "acpx",
+      runtime: runtimeState.runtime,
+    });
+
+    let currentMeta: SessionAcpMeta = {
+      ...readySessionMeta(),
+      agent: "gemini",
+      identity: {
+        state: "pending",
+        source: "ensure",
+        acpxSessionId: "acpx-stale",
+        lastUpdatedAt: Date.now(),
+      },
+    };
+    const sessionKey = "agent:gemini:acp:session-1";
+    hoisted.readAcpSessionEntryMock.mockImplementation((paramsUnknown: unknown) => {
+      const key = (paramsUnknown as { sessionKey?: string }).sessionKey ?? sessionKey;
+      return {
+        sessionKey: key,
+        storeSessionKey: key,
+        acp: currentMeta,
+      };
+    });
+    hoisted.upsertAcpSessionMetaMock.mockImplementation(async (paramsUnknown: unknown) => {
+      const params = paramsUnknown as {
+        mutate: (
+          current: SessionAcpMeta | undefined,
+          entry: { acp?: SessionAcpMeta } | undefined,
+        ) => SessionAcpMeta | null | undefined;
+      };
+      const next = params.mutate(currentMeta, { acp: currentMeta });
+      if (next) {
+        currentMeta = next;
+      }
+      return {
+        sessionId: "session-1",
+        updatedAt: Date.now(),
+        acp: currentMeta,
+      };
+    });
+
+    const manager = new AcpSessionManager();
+    await manager.runTurn({
+      cfg: baseCfg,
+      sessionKey,
+      text: "learn prompt session",
+      mode: "prompt",
+      requestId: "run-prompt-learned-agent-id",
+    });
+
+    expect(currentMeta.identity?.state).toBe("resolved");
+    expect(currentMeta.identity?.agentSessionId).toBe("gemini-session-1");
+    expect(currentMeta.identity?.acpxSessionId).toBe("acpx-stale");
+  });
+
   it("skips startup identity reconciliation for already resolved sessions", async () => {
     const runtimeState = createRuntime();
     hoisted.requireAcpRuntimeBackendMock.mockReturnValue({

src/acp/runtime/session-identity.ts

@@ -77,7 +77,7 @@ export function resolveRuntimeResumeSessionId(
   if (!identity) {
     return undefined;
   }
-  return normalizeText(identity.acpxSessionId) ?? normalizeText(identity.agentSessionId);
+  return normalizeText(identity.agentSessionId) ?? normalizeText(identity.acpxSessionId);
 }
 
 export function isSessionIdentityPending(identity: SessionAcpIdentity | undefined): boolean {
@@ -175,6 +175,26 @@ export function createIdentityFromEnsure(params: {
   };
 }
 
+export function createIdentityFromHandleEvent(params: {
+  handle: AcpRuntimeHandle;
+  now: number;
+}): SessionAcpIdentity | undefined {
+  const acpxRecordId = normalizeText((params.handle as { acpxRecordId?: unknown }).acpxRecordId);
+  const acpxSessionId = normalizeText(params.handle.backendSessionId);
+  const agentSessionId = normalizeText(params.handle.agentSessionId);
+  if (!acpxRecordId && !acpxSessionId && !agentSessionId) {
+    return undefined;
+  }
+  return {
+    state: agentSessionId ? "resolved" : "pending",
+    ...(acpxRecordId ? { acpxRecordId } : {}),
+    ...(acpxSessionId ? { acpxSessionId } : {}),
+    ...(agentSessionId ? { agentSessionId } : {}),
+    source: "event",
+    lastUpdatedAt: params.now,
+  };
+}
+
 export function createIdentityFromStatus(params: {
   status: AcpRuntimeStatus | undefined;
   now: number;