fix(auto-reply): stamp CLI activity once at the lifecycle bridge seam

src/auto-reply/reply/agent-runner-cli-dispatch.test.ts

@@ -171,6 +171,95 @@ describe("runCliAgentWithLifecycle", () => {
     expect(result.payloads).toEqual([{ text: "Visible answer" }]);
   });
 
+  it("stamps onActivity for every delivered bridge event, including reasoning-only progress", async () => {
+    cliDispatchState.runCliAgentMock.mockImplementationOnce(async (params: { runId: string }) => {
+      emitAgentEvent({
+        runId: params.runId,
+        stream: "thinking",
+        data: { progressTokens: 50 },
+      });
+      emitAgentEvent({
+        runId: params.runId,
+        stream: "thinking",
+        data: { progressTokens: 50 },
+      });
+      emitAgentEvent({
+        runId: params.runId,
+        stream: "thinking",
+        data: { progressTokens: 200 },
+      });
+      return { payloads: [], meta: { durationMs: 1 } };
+    });
+    const onActivity = vi.fn();
+    const onReasoningProgress = vi.fn<(payload: ReasoningProgressPayload) => Promise<void>>(
+      async () => undefined,
+    );
+
+    await runCliAgentWithLifecycle({
+      runId: "run-activity-reasoning-progress",
+      provider: "claude-cli",
+      onActivity,
+      onReasoningProgress,
+      runParams: {
+        sessionId: "session-1",
+        sessionFile: "/tmp/session.jsonl",
+        workspaceDir: "/tmp/workspace",
+        prompt: "hello",
+        provider: "claude-cli",
+        model: "claude",
+        thinkLevel: "high",
+        timeoutMs: 1_000,
+        runId: "run-activity-reasoning-progress",
+      },
+    });
+
+    // A run in a long pure-reasoning stretch must keep stamping activity, or
+    // stale-takeover reclaims it while it is visibly thinking.
+    expect(onReasoningProgress).toHaveBeenCalledTimes(2);
+    expect(onActivity).toHaveBeenCalledTimes(2);
+  });
+
+  it("stamps onActivity for assistant text without caller callbacks for that stream", async () => {
+    cliDispatchState.runCliAgentMock.mockImplementationOnce(async (params: { runId: string }) => {
+      emitAgentEvent({
+        runId: params.runId,
+        stream: "thinking",
+        data: { text: "Deep thought", delta: "Deep thought", isReasoningSnapshot: true },
+      });
+      emitAgentEvent({
+        runId: params.runId,
+        stream: "assistant",
+        data: { text: "Visible answer", delta: "Visible answer" },
+      });
+      return { payloads: [{ text: "Visible answer" }], meta: { durationMs: 1 } };
+    });
+    const onActivity = vi.fn();
+    const onAssistantText = vi.fn<(text: string) => Promise<void>>(async () => undefined);
+
+    await runCliAgentWithLifecycle({
+      runId: "run-activity-assistant",
+      provider: "claude-cli",
+      onActivity,
+      onAssistantText,
+      runParams: {
+        sessionId: "session-1",
+        sessionFile: "/tmp/session.jsonl",
+        workspaceDir: "/tmp/workspace",
+        prompt: "hello",
+        provider: "claude-cli",
+        model: "claude",
+        thinkLevel: "high",
+        timeoutMs: 1_000,
+        runId: "run-activity-assistant",
+      },
+    });
+
+    // Reasoning text stamps even with no onReasoningText caller: the durable
+    // reasoning deliver always runs, and thinking is activity evidence.
+    expect(onAssistantText).toHaveBeenCalledTimes(1);
+    expect(onActivity).toHaveBeenCalledTimes(2);
+  });
+
   it("does not add a durable reasoning payload when the CLI emits no thinking", async () => {
     cliDispatchState.runCliAgentMock.mockResolvedValueOnce({
       payloads: [{ text: "Visible answer" }],

src/auto-reply/reply/agent-runner-cli-dispatch.ts

@@ -412,6 +412,13 @@ type RunCliAgentWithLifecycleParams = {
   emitLifecycleTerminal?: boolean;
   onAgentRunStart?: () => void;
   suppressAssistantBridge?: boolean;
+  /**
+   * Stamped before every delivered CLI progress event (assistant, reasoning,
+   * tool, commentary, fast-mode). Callers wire this to the reply operation's
+   * activity evidence; per-callback stamps at call sites drift and a missed
+   * stamp lets stale-takeover reclaim a healthy run.
+   */
+  onActivity?: () => void;
   onAssistantText?: (text: string) => Promise<void>;
   onReasoningText?: (payload: ReasoningTextPayload) => Promise<void>;
   onReasoningProgress?: (payload: ReasoningProgressPayload) => Promise<void>;
@@ -463,10 +470,13 @@ async function runCliAgentWithLifecycleInternal(
       ...(params.runParams.sessionKey ? { sessionKey: params.runParams.sessionKey } : {}),
     });
     try {
-      await params.onFastModeAutoProgress?.({
-        text: summary,
-        channelData: { openclawProgressKind: FAST_MODE_AUTO_PROGRESS_KIND },
-      });
+      if (params.onFastModeAutoProgress) {
+        params.onActivity?.();
+        await params.onFastModeAutoProgress({
+          text: summary,
+          channelData: { openclawProgressKind: FAST_MODE_AUTO_PROGRESS_KIND },
+        });
+      }
     } catch {
       // Progress hints are best-effort; a channel failure must not fail the agent turn.
     }
@@ -518,34 +528,46 @@ async function runCliAgentWithLifecycleInternal(
       },
     });
   }
+  // One activity seam for every delivered CLI event: stamping per-callback at
+  // call sites drifted (a queued-followup callback missed one), which lets
+  // stale-takeover reclaim a run that is visibly producing output.
+  const withActivity = <T>(
+    deliver: ((payload: T) => Promise<void>) | undefined,
+  ): ((payload: T) => Promise<void>) | undefined =>
+    deliver && params.onActivity
+      ? async (payload: T) => {
+          params.onActivity?.();
+          await deliver(payload);
+        }
+      : deliver;
   const assistantBridge = createAssistantTextBridge({
     runId: params.runId,
     suppressed: params.suppressAssistantBridge,
-    deliver: params.onAssistantText,
+    deliver: withActivity(params.onAssistantText),
   });
   let finalReasoningText: string | undefined;
   const reasoningBridge = createReasoningTextBridge({
     runId: params.runId,
     suppressed: params.suppressAssistantBridge,
-    deliver: async (payload) => {
+    deliver: withActivity(async (payload: ReasoningTextPayload) => {
       finalReasoningText = normalizeOptionalString(payload.text);
       await params.onReasoningText?.(payload);
-    },
+    }),
   });
   const reasoningProgressBridge = createReasoningProgressBridge({
     runId: params.runId,
     suppressed: params.suppressAssistantBridge,
-    deliver: params.onReasoningProgress,
+    deliver: withActivity(params.onReasoningProgress),
   });
   const toolBridge = createToolEventBridge({
     runId: params.runId,
     suppressed: params.suppressAssistantBridge,
-    deliver: params.onToolEvent,
+    deliver: withActivity(params.onToolEvent),
   });
   const commentaryBridge = createCommentaryEventBridge({
     runId: params.runId,
     suppressed: params.suppressAssistantBridge,
-    deliver: params.onCommentaryText,
+    deliver: withActivity(params.onCommentaryText),
   });
   const toolBoundaryBridge = createToolBoundaryBridge({
     runId: params.runId,

src/auto-reply/reply/agent-runner-execution.ts

@@ -2429,8 +2429,8 @@ async function runAgentTurnWithFallbackInternal(
                   emitLifecycleTerminal: false,
                   onAgentRunStart: notifyAgentRunStart,
                   suppressAssistantBridge: params.followupRun.run.silentExpected,
+                  onActivity: () => params.replyOperation?.recordActivity(),
                   onAssistantText: async (text) => {
-                    params.replyOperation?.recordActivity();
                     const textForTyping = await handlePartialForTyping({ text } as ReplyPayload);
                     if (textForTyping === undefined || !params.opts?.onPartialReply) {
                       return;
@@ -2439,11 +2439,9 @@ async function runAgentTurnWithFallbackInternal(
                   },
                   onReasoningText: createCliReasoningStreamBridge(params.opts?.onReasoningStream),
                   onReasoningProgress: async (payload) => {
-                    params.replyOperation?.recordActivity();
                     await params.opts?.onReasoningProgress?.(payload);
                   },
                   onToolEvent: async (payload) => {
-                    params.replyOperation?.recordActivity();
                     await cliToolSummaryTracker.noteToolEvent(payload);
                     if (payload.phase === "result") {
                       return;
@@ -2462,7 +2460,6 @@ async function runAgentTurnWithFallbackInternal(
                   onCommentaryText:
                     params.opts?.commentaryProgressEnabled === true && params.opts.onItemEvent
                       ? async (payload) => {
-                          params.replyOperation?.recordActivity();
                           await params.opts?.onItemEvent?.({
                             itemId: payload.itemId,
                             kind: "preamble",
@@ -2471,7 +2468,6 @@ async function runAgentTurnWithFallbackInternal(
                         }
                       : undefined,
                   onFastModeAutoProgress: async (payload) => {
-                    params.replyOperation?.recordActivity();
                     await params.opts?.onToolResult?.(payload);
                   },
                   onErrorBeforeLifecycle: async () => {

src/auto-reply/reply/followup-runner.ts

@@ -1094,12 +1094,12 @@ export function createFollowupRunner(params: {
                   emitLifecycleTerminal: false,
                   onAgentRunStart: () => opts?.onAgentRunStart?.(runId),
                   suppressAssistantBridge: run.silentExpected,
+                  onActivity: () => replyOperation?.recordActivity(),
                   onReasoningText: createCliReasoningStreamBridge(progressOpts?.onReasoningStream),
                   onReasoningProgress: async (payload) => {
                     await progressOpts?.onReasoningProgress?.(payload);
                   },
                   onToolEvent: async (payload) => {
-                    replyOperation?.recordActivity();
                     await cliToolSummaryTracker.noteToolEvent(payload);
                     if (payload.phase === "result") {
                       return;
@@ -1117,7 +1117,6 @@ export function createFollowupRunner(params: {
                   onCommentaryText:
                     progressOpts?.commentaryProgressEnabled === true && progressOpts.onItemEvent
                       ? async ({ text, itemId }) => {
-                          replyOperation?.recordActivity();
                           await forwardFollowupProgressEvent({
                             evt: {
                               stream: "item",
@@ -1129,7 +1128,6 @@ export function createFollowupRunner(params: {
                         }
                       : undefined,
                   onFastModeAutoProgress: async (payload) => {
-                    replyOperation?.recordActivity();
                     await enqueueProgressDelivery(async () => {
                       // Mirrors direct dispatch progress suppression: ambient
                       // room events never get automatic fast-mode notices.