fix: preserve close state during live checkpoints

Author: steipete

CHANGELOG.md

@@ -13,6 +13,7 @@ Repo: https://github.com/openclaw/acpx
 ### Fixes
 
 - Runtime/embedding: preserve structured ACP `tool_call_update` details on public runtime events, including content, output, locations, kind, and raw payload fields, so embedders can display live tool progress. (#306) Thanks @joeia26.
+- CLI/sessions: checkpoint live assistant and tool updates while prompt turns are still running, so `sessions read` and `sessions history` can show in-flight progress instead of only the submitted prompt. (#314) Thanks @AndroidPoet.
 
 ## 2026.5.5 (v0.7.0)
 

src/cli/session/runtime.ts

@@ -409,12 +409,29 @@ async function runSessionPrompt(options: RunSessionPromptOptions): Promise<Sessi
       await eventWriter.appendMessages(batch, { checkpoint });
     });
   };
+  const preserveClosedState = async (): Promise<void> => {
+    const latest = await resolveSessionRecord(record.acpxRecordId).catch(() => undefined);
+    if (!latest?.closed) {
+      return;
+    }
+
+    record.closed = true;
+    record.closedAt = latest.closedAt ?? record.closedAt ?? isoNow();
+    record.pid = latest.pid;
+    if (latest.acpx) {
+      record.acpx = {
+        ...record.acpx,
+        ...latest.acpx,
+      };
+    }
+  };
   const liveCheckpoint = new LiveSessionCheckpoint({
     save: async () => {
       await flushPendingMessages(false);
       record.lastUsedAt = isoNow();
       applyConversation(record, conversation);
       record.acpx = acpxState;
+      await preserveClosedState();
       await eventWriter.checkpoint();
     },
     onError: (error) => {
@@ -720,6 +737,9 @@ async function runSessionPrompt(options: RunSessionPromptOptions): Promise<Sessi
     await flushPendingMessages(false).catch(() => {
       // best effort on close
     });
+    await preserveClosedState().catch(() => {
+      // best effort on close
+    });
     await closeEventWriter(true).catch(() => {
       // best effort on close
     });

src/runtime/engine/manager.ts

@@ -544,6 +544,7 @@ export class AcpRuntimeManager {
             runtimeRecord.lastUsedAt = isoNow();
             runtimeRecord.acpx = acpxState;
             applyConversation(runtimeRecord, runtimeConversation);
+            await this.refreshClosedState(runtimeRecord);
             await this.options.sessionStore.save(runtimeRecord);
           },
         });

test/integration.test.ts

@@ -3017,11 +3017,18 @@ test("integration: sessions read shows assistant updates before the prompt finis
       try {
         const history = await waitFor(async () => {
           const result = await runCli(
-            [...baseAgentArgs(cwd), "--format", "quiet", "sessions", "read"],
+            [...baseAgentArgs(cwd), "--format", "json", "sessions", "read"],
             homeDir,
           );
           assert.equal(result.code, 0, result.stderr);
-          return result.stdout.includes("foreground-live-update") ? result.stdout : null;
+          const payload = JSON.parse(result.stdout.trim()) as {
+            entries?: Array<{ role?: string; textPreview?: string }>;
+          };
+          const assistantEntry = payload.entries?.find(
+            (entry) =>
+              entry.role === "assistant" && entry.textPreview?.includes("foreground-live-update"),
+          );
+          return assistantEntry ? result.stdout : null;
         }, 5_000);
 
         assert.equal(promptChild.exitCode, null, "prompt should still be running");
@@ -3079,11 +3086,18 @@ test("integration: --no-wait stdin prompt checkpoints live assistant updates", a
 
       const history = await waitFor(async () => {
         const result = await runCli(
-          [...baseAgentArgs(cwd), "--format", "quiet", "sessions", "read"],
+          [...baseAgentArgs(cwd), "--format", "json", "sessions", "read"],
           homeDir,
         );
         assert.equal(result.code, 0, result.stderr);
-        return result.stdout.includes("background-live-update") ? result.stdout : null;
+        const payload = JSON.parse(result.stdout.trim()) as {
+          entries?: Array<{ role?: string; textPreview?: string }>;
+        };
+        const assistantEntry = payload.entries?.find(
+          (entry) =>
+            entry.role === "assistant" && entry.textPreview?.includes("background-live-update"),
+        );
+        return assistantEntry ? result.stdout : null;
       }, 5_000);
 
       assert.match(history, /background-live-update/);
@@ -3100,6 +3114,91 @@ test("integration: --no-wait stdin prompt checkpoints live assistant updates", a
   });
 });
 
+test("integration: sessions close stays closed after live checkpoints", async () => {
+  await withTempHome(async (homeDir) => {
+    const cwd = await fs.mkdtemp(path.join(os.tmpdir(), "acpx-integration-cwd-"));
+
+    try {
+      const created = await runCli(
+        [...baseAgentArgs(cwd), "--format", "json", "sessions", "new"],
+        homeDir,
+      );
+      assert.equal(created.code, 0, created.stderr);
+      const createdPayload = JSON.parse(created.stdout.trim()) as {
+        acpxRecordId?: string;
+      };
+      const sessionId = createdPayload.acpxRecordId;
+      assert.equal(typeof sessionId, "string");
+
+      const promptChild = spawn(
+        process.execPath,
+        [
+          CLI_PATH,
+          ...baseAgentArgs(cwd),
+          "--format",
+          "quiet",
+          "prompt",
+          "stream-sleep 5000 close-live-update",
+        ],
+        {
+          env: {
+            ...process.env,
+            HOME: homeDir,
+          },
+          stdio: ["ignore", "pipe", "pipe"],
+        },
+      );
+
+      try {
+        await waitFor(async () => {
+          const result = await runCli(
+            [...baseAgentArgs(cwd), "--format", "json", "sessions", "read"],
+            homeDir,
+          );
+          assert.equal(result.code, 0, result.stderr);
+          const payload = JSON.parse(result.stdout.trim()) as {
+            entries?: Array<{ role?: string; textPreview?: string }>;
+          };
+          const assistantEntry = payload.entries?.find(
+            (entry) =>
+              entry.role === "assistant" && entry.textPreview?.includes("close-live-update"),
+          );
+          return assistantEntry ? true : null;
+        }, 5_000);
+
+        const closed = await runCli(
+          [...baseAgentArgs(cwd), "--format", "json", "sessions", "close"],
+          homeDir,
+        );
+        assert.equal(closed.code, 0, closed.stderr);
+        if (promptChild.exitCode == null && promptChild.signalCode == null) {
+          await awaitChildClose(promptChild).catch(() => {});
+        }
+
+        const recordPath = path.join(
+          homeDir,
+          ".acpx",
+          "sessions",
+          `${encodeURIComponent(sessionId as string)}.json`,
+        );
+        const storedRecord = JSON.parse(await fs.readFile(recordPath, "utf8")) as {
+          closed?: boolean;
+          closed_at?: string;
+        };
+        assert.equal(storedRecord.closed, true);
+        assert.equal(typeof storedRecord.closed_at, "string");
+      } finally {
+        if (promptChild.exitCode == null && promptChild.signalCode == null) {
+          promptChild.kill("SIGKILL");
+          await awaitChildClose(promptChild).catch(() => {});
+        }
+      }
+    } finally {
+      await fs.rm(cwd, { recursive: true, force: true });
+    }
+  });
+});
+
 test("integration: session remains resumable after queue owner exits and agent has exited", async () => {
   await withTempHome(async (homeDir) => {
     const cwd = await fs.mkdtemp(path.join(os.tmpdir(), "acpx-integration-cwd-"));

test/runtime-manager.test.ts

@@ -802,6 +802,93 @@ test("AcpRuntimeManager does not pool a persistent client after active close", a
   assert.equal(typeof closed?.closedAt, "string");
 });
 
+test("AcpRuntimeManager live checkpoints preserve active close state", async () => {
+  const record = makeSessionRecord({
+    acpxRecordId: "active-close-checkpoint-session",
+    acpSessionId: "active-close-checkpoint-sid",
+    agentCommand: "codex --acp",
+    cwd: "/workspace",
+  });
+  const store = new InMemorySessionStore([record]);
+  let handlers: FakeClientHandlers = {};
+  let promptActive = false;
+  let resolvePromptStart!: () => void;
+  let resolvePrompt!: (value: { stopReason: string }) => void;
+  const promptStarted = new Promise<void>((resolve) => {
+    resolvePromptStart = resolve;
+  });
+  const promptResult = new Promise<{ stopReason: string }>((resolve) => {
+    resolvePrompt = resolve;
+  });
+  const client: FakeClient = {
+    start: async () => {},
+    close: async () => {
+      promptActive = false;
+    },
+    createSession: async () => ({ sessionId: "unused" }),
+    loadSession: async () => ({ agentSessionId: "unused" }),
+    hasReusableSession: (sessionId) => sessionId === "active-close-checkpoint-sid",
+    supportsLoadSession: () => true,
+    supportsCloseSession: () => true,
+    closeSession: async () => {},
+    loadSessionWithOptions: async () => ({ agentSessionId: "active-close-checkpoint-agent-id" }),
+    getAgentLifecycleSnapshot: () => ({ running: promptActive }),
+    prompt: async () => {
+      promptActive = true;
+      handlers.onSessionUpdate?.({
+        sessionId: "active-close-checkpoint-sid",
+        update: {
+          sessionUpdate: "agent_message_chunk",
+          content: { type: "text", text: "live checkpoint" },
+        },
+      });
+      resolvePromptStart();
+      return await promptResult;
+    },
+    requestCancelActivePrompt: async () => {
+      promptActive = false;
+      return true;
+    },
+    hasActivePrompt: () => promptActive,
+    setSessionMode: async () => {},
+    setSessionConfigOption: async () => {},
+    clearEventHandlers: () => {
+      handlers = {};
+    },
+    setEventHandlers: (nextHandlers) => {
+      handlers = nextHandlers;
+    },
+  };
+  const manager = new AcpRuntimeManager(
+    createRuntimeOptions({ cwd: "/workspace", sessionStore: store }),
+    {
+      clientFactory: () => client as never,
+    },
+  );
+  const handle = createHandle("active-close-checkpoint-session");
+
+  const turn = manager.startTurn({
+    handle,
+    text: "hello",
+    mode: "prompt",
+    sessionMode: "persistent",
+    requestId: "req-active-close-checkpoint",
+  });
+  const eventsPromise = collectEvents(turn.events);
+  await promptStarted;
+
+  await manager.close(handle, { discardPersistentState: true });
+  await new Promise((resolve) => setTimeout(resolve, 650));
+
+  const checkpointed = await store.load("active-close-checkpoint-session");
+  assert.equal(checkpointed?.closed, true);
+  assert.equal(checkpointed?.acpx?.reset_on_next_ensure, true);
+
+  resolvePrompt({ stopReason: "cancelled" });
+  await eventsPromise;
+  await turn.result;
+});
+
 test("AcpRuntimeManager accepts a session reply even when the prompt RPC times out", async () => {
   const record = makeSessionRecord({
     acpxRecordId: "late-reply-session",