diff --git a/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts b/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts index 677745ff78d8..aed0dd27bca9 100644 --- a/src/auto-reply/reply/agent-runner-cli-dispatch.test.ts +++ b/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>( + 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>(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" }], diff --git a/src/auto-reply/reply/agent-runner-cli-dispatch.ts b/src/auto-reply/reply/agent-runner-cli-dispatch.ts index 78a80f476145..f85ce88231d4 100644 --- a/src/auto-reply/reply/agent-runner-cli-dispatch.ts +++ b/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; onReasoningText?: (payload: ReasoningTextPayload) => Promise; onReasoningProgress?: (payload: ReasoningProgressPayload) => Promise; @@ -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 = ( + deliver: ((payload: T) => Promise) | undefined, + ): ((payload: T) => Promise) | 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, diff --git a/src/auto-reply/reply/agent-runner-execution.ts b/src/auto-reply/reply/agent-runner-execution.ts index 6f28937def73..d1d4fb317970 100644 --- a/src/auto-reply/reply/agent-runner-execution.ts +++ b/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 () => { diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index 00692e5cf910..57755767ee68 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/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.