diff --git a/src/agents/cli-runner.reliability.test.ts b/src/agents/cli-runner.reliability.test.ts index f8b0e350cb61..db33e5eac95f 100644 --- a/src/agents/cli-runner.reliability.test.ts +++ b/src/agents/cli-runner.reliability.test.ts @@ -808,8 +808,128 @@ describe("runCliAgent reliability", () => { releaseAgentEnd(); }); + it("persists approved CLI user turns before model execution", async () => { + supervisorSpawnMock.mockResolvedValueOnce( + createManagedRun({ + reason: "exit", + exitCode: 0, + exitSignal: null, + durationMs: 50, + stdout: "hello from cli", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + }), + ); + const { dir, sessionFile } = createSessionFile(); + const onUserMessagePersisted = vi.fn(); + + try { + const context = buildPreparedContext({ + sessionKey: "agent:main:main", + runId: "run-persist-cli", + }); + const result = await runPreparedCliAgent({ + ...context, + params: { + ...context.params, + agentId: "main", + sessionFile, + workspaceDir: dir, + prompt: "runtime prompt", + userTurnTranscript: { text: "display prompt" }, + onUserMessagePersisted, + }, + }); + + expect(result.payloads).toEqual([{ text: "hello from cli" }]); + expect(onUserMessagePersisted).toHaveBeenCalledOnce(); + expect(onUserMessagePersisted).toHaveBeenCalledWith( + expect.objectContaining({ + role: "user", + content: "display prompt", + }), + ); + + const messages = fs + .readFileSync(sessionFile, "utf-8") + .trim() + .split("\n") + .map((line) => JSON.parse(line) as { message?: unknown }) + .map((entry) => entry.message) + .filter(Boolean); + expect(messages).toContainEqual( + expect.objectContaining({ + role: "user", + content: "display prompt", + }), + ); + expect(JSON.stringify(messages)).not.toContain("runtime prompt"); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); + + it("persists approved CLI media user turns without caller-side transcript shaping", async () => { + supervisorSpawnMock.mockResolvedValueOnce( + createManagedRun({ + reason: "exit", + exitCode: 0, + exitSignal: null, + durationMs: 50, + stdout: "image handled", + stderr: "", + timedOut: false, + noOutputTimedOut: false, + }), + ); + const { dir, sessionFile } = createSessionFile(); + const onUserMessagePersisted = vi.fn(); + + try { + const context = buildPreparedContext({ + sessionKey: "agent:main:main", + runId: "run-persist-cli-media", + }); + await runPreparedCliAgent({ + ...context, + params: { + ...context.params, + agentId: "main", + sessionFile, + workspaceDir: dir, + prompt: "runtime image prompt", + userTurnTranscript: { + message: { + role: "user", + content: "describe this", + MediaPath: "/tmp/image.png", + MediaType: "image/png", + }, + }, + onUserMessagePersisted, + }, + }); + + expect(onUserMessagePersisted).toHaveBeenCalledWith( + expect.objectContaining({ + role: "user", + content: "describe this", + MediaPath: "/tmp/image.png", + MediaType: "image/png", + }), + ); + const transcript = fs.readFileSync(sessionFile, "utf-8"); + expect(transcript).toContain('"MediaPath":"/tmp/image.png"'); + expect(transcript).toContain('"MediaType":"image/png"'); + } finally { + fs.rmSync(dir, { recursive: true, force: true }); + } + }); + it("blocks CLI runs before llm_input and model execution when before_agent_run blocks", async () => { supervisorSpawnMock.mockClear(); + const onUserMessagePersisted = vi.fn(); let releaseAgentEnd: () => void = () => undefined; const agentEndSettled = new Promise((resolve) => { releaseAgentEnd = resolve; @@ -836,15 +956,20 @@ describe("runCliAgent reliability", () => { try { let resolved = false; + const context = buildPreparedContext({ + sessionKey: "agent:main:main", + runId: "run-blocked-cli", + }); const run = runPreparedCliAgent({ - ...buildPreparedContext({ sessionKey: "agent:main:main", runId: "run-blocked-cli" }), + ...context, params: { - ...buildPreparedContext({ sessionKey: "agent:main:main", runId: "run-blocked-cli" }) - .params, + ...context.params, agentId: "main", sessionFile, workspaceDir: dir, prompt: "secret prompt", + userTurnTranscript: { text: "secret prompt" }, + onUserMessagePersisted, }, }).then((result) => { resolved = true; @@ -869,6 +994,7 @@ describe("runCliAgent reliability", () => { expect(result.meta.livenessState).toBe("blocked"); expect(supervisorSpawnMock).not.toHaveBeenCalled(); expect(hookRunner.runLlmInput).not.toHaveBeenCalled(); + expect(onUserMessagePersisted).not.toHaveBeenCalled(); const beforeRunEvent = requireRecord( callArg(hookRunner.runBeforeAgentRun, 0, 0, "before_agent_run event"), "before_agent_run event", diff --git a/src/agents/cli-runner.ts b/src/agents/cli-runner.ts index 01e99e0b7524..602ff6e7cb4a 100644 --- a/src/agents/cli-runner.ts +++ b/src/agents/cli-runner.ts @@ -7,6 +7,7 @@ import { createSubsystemLogger } from "../logging/subsystem.js"; import { buildAgentHookContextChannelFields } from "../plugins/hook-agent-context.js"; import { resolveBlockMessage } from "../plugins/hook-decision-types.js"; import { getGlobalHookRunner } from "../plugins/hook-runner-global.js"; +import { tryAppendInlineUserTurnTranscriptMessage } from "../sessions/user-turn-transcript.js"; import { loadCliSessionContextEngineMessages, loadCliSessionHistoryMessages, @@ -126,6 +127,36 @@ async function runCliAgentEndHook( runAgentHarnessAgentEndHook(hookParams); } +async function persistApprovedCliUserTurnTranscript(params: RunCliAgentParams): Promise { + if (params.suppressNextUserMessagePersistence === true || !params.userTurnTranscript) { + return; + } + + const transcriptTarget = { + transcriptPath: params.sessionFile, + sessionId: params.sessionId, + ...(params.sessionKey ? { sessionKey: params.sessionKey } : {}), + cwd: params.workspaceDir, + errorContext: "CLI user turn transcript", + ...(params.config ? { config: params.config } : {}), + }; + const persisted = params.userTurnTranscript.message + ? await tryAppendInlineUserTurnTranscriptMessage({ + ...transcriptTarget, + message: params.userTurnTranscript.message, + }) + : await tryAppendInlineUserTurnTranscriptMessage({ + ...transcriptTarget, + input: { + text: params.userTurnTranscript.text, + timestamp: Date.now(), + }, + }); + if (persisted) { + await params.onUserMessagePersisted?.(persisted.message); + } +} + async function finalizeCliContextEngineTurn(params: { context: PreparedCliRunContext; historyMessages: unknown[]; @@ -620,6 +651,7 @@ export async function runPreparedCliAgent( } } + await persistApprovedCliUserTurnTranscript(params); runAgentHarnessLlmInputHook({ event: llmInputEvent, ctx: hookContext, diff --git a/src/agents/cli-runner/types.ts b/src/agents/cli-runner/types.ts index 2141b7ef8337..3ea9142c89df 100644 --- a/src/agents/cli-runner/types.ts +++ b/src/agents/cli-runner/types.ts @@ -10,6 +10,7 @@ import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { ContextEngine } from "../../context-engine/types.js"; import type { PromptImageOrderEntry } from "../../media/prompt-image-order.js"; import type { InputProvenance } from "../../sessions/input-provenance.js"; +import type { PersistedUserTurnMessage } from "../../sessions/user-turn-transcript.js"; import type { BootstrapContextMode } from "../bootstrap-files.js"; import type { ResolvedCliBackend } from "../cli-backends.js"; import type { ContextWindowInfo } from "../context-window-guard.js"; @@ -21,6 +22,16 @@ import type { import type { SkillSnapshot } from "../skills.js"; import type { SilentReplyPromptMode } from "../system-prompt.types.js"; +export type CliUserTurnTranscriptInput = + | { + message: PersistedUserTurnMessage; + text?: never; + } + | { + message?: never; + text: string; + }; + export type RunCliAgentParams = { sessionId: string; sessionKey?: string; @@ -32,6 +43,13 @@ export type RunCliAgentParams = { config?: OpenClawConfig; prompt: string; transcriptPrompt?: string; + /** + * Canonical user turn to persist after before_agent_run allows the prompt. + * This is transcript projection only; model input still comes from prompt/images/context. + */ + userTurnTranscript?: CliUserTurnTranscriptInput; + suppressNextUserMessagePersistence?: boolean; + onUserMessagePersisted?: (message: PersistedUserTurnMessage) => void | Promise; currentInboundEventKind?: InboundEventKind; currentInboundContext?: CurrentInboundPromptContext; inputProvenance?: InputProvenance; diff --git a/src/auto-reply/reply/agent-runner-execution.test.ts b/src/auto-reply/reply/agent-runner-execution.test.ts index 0381c253f94f..40e31d4897f2 100644 --- a/src/auto-reply/reply/agent-runner-execution.test.ts +++ b/src/auto-reply/reply/agent-runner-execution.test.ts @@ -30,7 +30,6 @@ const state = vi.hoisted(() => ({ isContextOverflowErrorMock: vi.fn((_: string | undefined) => false), isLikelyContextOverflowErrorMock: vi.fn((_: string | undefined) => false), updateSessionStoreMock: vi.fn(), - persistUserTurnTranscriptMock: vi.fn(), })); const GENERIC_RUN_FAILURE_TEXT = @@ -138,17 +137,6 @@ vi.mock("../../utils/message-channel.js", () => ({ isInternalMessageChannel: (value: unknown) => state.isInternalMessageChannelMock(value), })); -vi.mock("../../sessions/user-turn-transcript.js", async () => { - const actual = await vi.importActual( - "../../sessions/user-turn-transcript.js", - ); - return { - ...actual, - tryPersistInlineUserTurnTranscript: (params: unknown) => - state.persistUserTurnTranscriptMock(params), - }; -}); - vi.mock("../heartbeat.js", () => ({ stripHeartbeatToken: (text: string) => ({ text, @@ -1104,12 +1092,6 @@ describe("runAgentTurnWithFallback", () => { state.isLikelyContextOverflowErrorMock.mockReset(); state.isLikelyContextOverflowErrorMock.mockReturnValue(false); state.updateSessionStoreMock.mockReset(); - state.persistUserTurnTranscriptMock.mockReset(); - state.persistUserTurnTranscriptMock.mockResolvedValue({ - sessionFile: "/tmp/session.jsonl", - messageId: "user-message", - message: { role: "user", content: "persisted" }, - }); state.runWithModelFallbackMock.mockImplementation(async (params: FallbackRunnerParams) => ({ result: await params.run("anthropic", "claude"), provider: "anthropic", @@ -1784,7 +1766,7 @@ describe("runAgentTurnWithFallback", () => { }); }); - it("persists CLI-backed user turns through the shared session helper", async () => { + it("passes prepared CLI user turns to the runtime persistence boundary", async () => { state.isCliProviderMock.mockReturnValue(true); state.runWithModelFallbackMock.mockImplementationOnce(async (params: FallbackRunnerParams) => ({ result: await params.run("codex-cli", "gpt-5.4"), @@ -1826,21 +1808,21 @@ describe("runAgentTurnWithFallback", () => { }); expect(result.kind).toBe("success"); - expect(state.persistUserTurnTranscriptMock).toHaveBeenCalledOnce(); - expectMockCallArgFields(state.persistUserTurnTranscriptMock, 0, "user turn persistence", { - message: followupRun.userMessageForPersistence, - sessionId: "session", + expect(state.runCliAgentMock).toHaveBeenCalledOnce(); + expectMockCallArgFields(state.runCliAgentMock, 0, "CLI runtime", { sessionKey: "main", - sessionEntry, - sessionStore: activeSessionStore, - storePath: "/tmp/sessions.json", agentId: "agent", - cwd: "/tmp", - errorContext: "CLI user turn transcript", + sessionId: "session", + userTurnTranscript: { message: followupRun.userMessageForPersistence }, + suppressNextUserMessagePersistence: false, }); + const call = requireMockCall(state.runCliAgentMock, 0, "CLI runtime"); + expect(requireRecord(call[0], "CLI runtime").onUserMessagePersisted).toEqual( + expect.any(Function), + ); }); - it("uses clean transcript text for text-only CLI user persistence", async () => { + it("passes clean transcript text for text-only CLI user persistence", async () => { state.isCliProviderMock.mockReturnValue(true); state.runWithModelFallbackMock.mockImplementationOnce(async (params: FallbackRunnerParams) => ({ result: await params.run("codex-cli", "gpt-5.4"), @@ -1864,15 +1846,16 @@ describe("runAgentTurnWithFallback", () => { transcriptCommandBody: "display prompt", }); - expectMockCallArgFields(state.persistUserTurnTranscriptMock, 0, "user turn persistence", { + expect(state.runCliAgentMock).toHaveBeenCalledOnce(); + expectMockCallArgFields(state.runCliAgentMock, 0, "CLI runtime", { sessionId: "session", sessionKey: "main", agentId: "agent", - text: "display prompt", - errorContext: "CLI user turn transcript", + prompt: "runtime prompt with metadata", + transcriptPrompt: "display prompt", + userTurnTranscript: { text: "display prompt" }, + suppressNextUserMessagePersistence: false, }); - const call = requireMockCall(state.persistUserTurnTranscriptMock, 0, "user turn persistence"); - expect(requireRecord(call[0], "user turn persistence").input).toBeUndefined(); }); it("does not reuse or persist CLI sessions for room-event turns", async () => { diff --git a/src/auto-reply/reply/followup-runner.test.ts b/src/auto-reply/reply/followup-runner.test.ts index 6b9aa852bfae..5bec4fa37f5d 100644 --- a/src/auto-reply/reply/followup-runner.test.ts +++ b/src/auto-reply/reply/followup-runner.test.ts @@ -10,7 +10,6 @@ import type { FollowupRun, QueueSettings } from "./queue.js"; const runEmbeddedPiAgentMock = vi.fn(); const runCliAgentMock = vi.fn(); const runWithModelFallbackMock = vi.fn(); -const persistUserTurnTranscriptMock = vi.fn(); const compactEmbeddedPiSessionMock = vi.fn(); const routeReplyMock = vi.fn(); const isRoutableChannelMock = vi.fn(); @@ -353,16 +352,6 @@ async function loadFreshFollowupRunnerModuleForTest() { vi.doMock("../../agents/cli-runner.js", () => ({ runCliAgent: (params: unknown) => runCliAgentMock(params), })); - vi.doMock("../../sessions/user-turn-transcript.js", async () => { - const actual = await vi.importActual( - "../../sessions/user-turn-transcript.js", - ); - return { - ...actual, - tryPersistInlineUserTurnTranscript: (params: unknown) => - persistUserTurnTranscriptMock(params), - }; - }); vi.doMock("./queue.js", () => ({ clearFollowupQueue: clearFollowupQueueForFollowupTest, completeFollowupRunLifecycle: (run: Pick) => @@ -472,11 +461,6 @@ beforeEach(() => { clearRuntimeConfigSnapshot?.(); runEmbeddedPiAgentMock.mockReset(); runCliAgentMock.mockReset(); - persistUserTurnTranscriptMock.mockReset(); - persistUserTurnTranscriptMock.mockResolvedValue({ - message: { role: "user", content: [{ type: "text", text: "hello" }] }, - sessionFile: "/tmp/session.jsonl", - }); runWithModelFallbackMock.mockReset(); runWithModelFallbackMock.mockImplementation( async (params: { @@ -832,23 +816,63 @@ describe("createFollowupRunner runtime config", () => { expect(call.config).toBe(runtimeConfig); expect(call.cliSessionId).toBe("cli-session-1"); expect(call.messageChannel).toBe("telegram"); - expect(persistUserTurnTranscriptMock).toHaveBeenCalledOnce(); - const persistenceCall = requireLastMockCallArg( - persistUserTurnTranscriptMock, - "persist user turn transcript", - ); - expect(persistenceCall).toMatchObject({ + expect(call).toMatchObject({ sessionId: "session-cli-followup", sessionKey: "main", - sessionEntry, - sessionStore, agentId: "agent", - cwd: "/tmp", + workspaceDir: "/tmp", config: runtimeConfig, - text: "hello", - errorContext: "CLI user turn transcript", + userTurnTranscript: { text: "hello" }, + suppressNextUserMessagePersistence: false, }); - expect(persistenceCall.input).toBeUndefined(); + expect(call.onUserMessagePersisted).toEqual(expect.any(Function)); + }); + + it("passes prepared media user turns to CLI runtime dispatch", async () => { + const runtimeConfig: OpenClawConfig = { + agents: { + defaults: { + cliBackends: { + "claude-cli": { command: "claude" }, + }, + models: { + "anthropic/claude-opus-4-7": { agentRuntime: { id: "claude-cli" } }, + }, + }, + }, + }; + const userMessageForPersistence = { + role: "user", + content: "describe this", + MediaPath: "/tmp/image.png", + MediaType: "image/png", + } as never; + runCliAgentMock.mockResolvedValueOnce({ + payloads: [{ text: "done" }], + meta: {}, + }); + + const runner = createFollowupRunner({ + typing: createMockTypingController(), + typingMode: "instant", + sessionKey: "main", + defaultModel: "anthropic/claude-opus-4-7", + }); + + await runner( + createQueuedRun({ + userMessageForPersistence, + run: { + config: runtimeConfig, + provider: "anthropic", + model: "claude-opus-4-7", + }, + }), + ); + + expect(runCliAgentMock).toHaveBeenCalledOnce(); + const mediaCall = requireLastMockCallArg(runCliAgentMock, "run cli agent"); + expect(mediaCall.userTurnTranscript).toEqual({ message: userMessageForPersistence }); }); it("defers queued CLI attempt terminal lifecycle events until fallback settles", async () => { diff --git a/src/auto-reply/reply/followup-runner.ts b/src/auto-reply/reply/followup-runner.ts index 8556ea924ef7..a55543bfbcea 100644 --- a/src/auto-reply/reply/followup-runner.ts +++ b/src/auto-reply/reply/followup-runner.ts @@ -26,7 +26,6 @@ import { emitAgentEvent, registerAgentRunContext } from "../../infra/agent-event import { formatErrorMessage } from "../../infra/errors.js"; import { defaultRuntime } from "../../runtime.js"; import { shouldPreserveUserFacingSessionStateForInputProvenance } from "../../sessions/input-provenance.js"; -import { tryPersistInlineUserTurnTranscript } from "../../sessions/user-turn-transcript.js"; import { readStringValue } from "../../shared/string-coerce.js"; import { isInternalMessageChannel } from "../../utils/message-channel.js"; import type { GetReplyOptions, ReplyPayload } from "../types.js"; @@ -724,6 +723,13 @@ export function createFollowupRunner(params: { config: runtimeConfig, prompt: queued.prompt, transcriptPrompt: queued.transcriptPrompt, + userTurnTranscript: effectiveQueued.userMessageForPersistence + ? { message: effectiveQueued.userMessageForPersistence } + : { text: effectiveQueued.transcriptPrompt ?? effectiveQueued.prompt }, + suppressNextUserMessagePersistence: suppressQueuedUserPersistenceForCandidate, + onUserMessagePersisted: () => { + queuedUserMessagePersistedAcrossFallback = true; + }, currentInboundEventKind: queued.currentInboundEventKind, currentInboundContext: queued.currentInboundContext, inputProvenance: run.inputProvenance, @@ -777,25 +783,6 @@ export function createFollowupRunner(params: { })() : rawResult, }); - if (!suppressQueuedUserPersistenceForCandidate) { - const persistedUserTurn = await tryPersistInlineUserTurnTranscript({ - ...(effectiveQueued.userMessageForPersistence - ? { message: effectiveQueued.userMessageForPersistence } - : { text: effectiveQueued.transcriptPrompt ?? effectiveQueued.prompt }), - sessionId: run.sessionId, - sessionKey: replySessionKey ?? run.sessionId, - sessionEntry: activeSessionEntry, - ...(sessionStore ? { sessionStore } : {}), - ...(storePath ? { storePath } : {}), - agentId: run.agentId, - cwd: run.workspaceDir, - config: runtimeConfig, - errorContext: "CLI user turn transcript", - }); - if (persistedUserTurn) { - queuedUserMessagePersistedAcrossFallback = true; - } - } bootstrapPromptWarningSignaturesSeen = resolveBootstrapWarningSignaturesSeen( result.meta?.systemPromptReport, );