diff --git a/src/agents/agent-tools.ts b/src/agents/agent-tools.ts index 72251aed2780..f0c4dc97102f 100644 --- a/src/agents/agent-tools.ts +++ b/src/agents/agent-tools.ts @@ -470,6 +470,8 @@ export function createOpenClawCodingTools(options?: { currentMessageId?: string | number; /** True when the current inbound turn carried audio media. */ currentInboundAudio?: boolean; + /** Dynamic audio state for runs that can accept steered input after tool creation. */ + hasCurrentInboundAudio?: () => boolean; /** Group id for channel-level tool policy resolution. */ groupId?: string | null; /** Group channel label (e.g. #general) for channel-level tool policy resolution. */ @@ -1010,6 +1012,7 @@ export function createOpenClawCodingTools(options?: { currentThreadTs: options?.currentThreadTs, currentMessageId: options?.currentMessageId, currentInboundAudio: options?.currentInboundAudio, + hasCurrentInboundAudio: options?.hasCurrentInboundAudio, modelProvider: options?.modelProvider, modelId: options?.modelId, replyToMode: options?.replyToMode, diff --git a/src/agents/embedded-agent-runner/run/attempt.ts b/src/agents/embedded-agent-runner/run/attempt.ts index aa45849a6e1c..1b1ef82f1100 100644 --- a/src/agents/embedded-agent-runner/run/attempt.ts +++ b/src/agents/embedded-agent-runner/run/attempt.ts @@ -1389,6 +1389,13 @@ export async function runEmbeddedAttempt( currentThreadTs: params.currentThreadTs, currentMessageId: params.currentMessageId, currentInboundAudio: params.currentInboundAudio, + ...(params.replyOperation + ? { + hasCurrentInboundAudio: () => + params.currentInboundAudio === true || + params.replyOperation?.acceptedSteeredInboundAudio === true, + } + : {}), includeCoreTools: toolConstructionPlan.includeCoreTools, includeToolSearchControls: toolSearchControlsEnabledForRun, toolSearchCatalogExecutor: (toolParams) => { diff --git a/src/agents/openclaw-tools.ts b/src/agents/openclaw-tools.ts index 007abaf00c28..65c966f1cc34 100644 --- a/src/agents/openclaw-tools.ts +++ b/src/agents/openclaw-tools.ts @@ -123,6 +123,8 @@ export function createOpenClawTools( currentMessageId?: string | number; /** True when the current inbound turn carried audio media. */ currentInboundAudio?: boolean; + /** Dynamic audio state for runs that can accept steered input after tool creation. */ + hasCurrentInboundAudio?: () => boolean; /** Reply-to mode for auto-threading. */ replyToMode?: "off" | "first" | "all" | "batched"; /** Mutable ref to track if a reply was sent (for "first" mode). */ @@ -358,6 +360,7 @@ export function createOpenClawTools( currentChannelProvider: options?.agentChannel, currentThreadTs: options?.currentThreadTs, currentInboundAudio: options?.currentInboundAudio, + hasCurrentInboundAudio: options?.hasCurrentInboundAudio, agentThreadId: options?.agentThreadId, currentMessageId: options?.currentMessageId, replyToMode: options?.replyToMode, diff --git a/src/agents/tools/message-tool.test.ts b/src/agents/tools/message-tool.test.ts index 5664ea3806d3..0d1da3bc6dd9 100644 --- a/src/agents/tools/message-tool.test.ts +++ b/src/agents/tools/message-tool.test.ts @@ -805,6 +805,24 @@ describe("message tool secret scoping", () => { expect(input?.sourceReplyDeliveryMode).toBe("message_tool_only"); }); + it("reads steered inbound audio when the message action runs", async () => { + mockSendResult(); + let hasCurrentInboundAudio = false; + const tool = createMessageTool({ + currentInboundAudio: false, + hasCurrentInboundAudio: () => hasCurrentInboundAudio, + sourceReplyDeliveryMode: "message_tool_only", + currentChannelProvider: "whatsapp", + agentSessionKey: "agent:main:whatsapp:direct:123456789", + runMessageAction: mocks.runMessageAction as never, + }); + hasCurrentInboundAudio = true; + + await tool.execute("call1", { action: "send", message: "hi" }); + + expect(lastRunMessageActionInput()?.inboundAudio).toBe(true); + }); + it("adds a current-run idempotency key when the model omits one", async () => { mockSendResult(); diff --git a/src/agents/tools/message-tool.ts b/src/agents/tools/message-tool.ts index 1c2615ac9d51..0618e2a27a67 100644 --- a/src/agents/tools/message-tool.ts +++ b/src/agents/tools/message-tool.ts @@ -890,6 +890,7 @@ type MessageToolOptions = { agentThreadId?: string | number; currentMessageId?: string | number; currentInboundAudio?: boolean; + hasCurrentInboundAudio?: () => boolean; replyToMode?: "off" | "first" | "all" | "batched"; hasRepliedRef?: { value: boolean }; sameChannelThreadRequired?: boolean; @@ -1503,7 +1504,7 @@ export function createMessageTool(options?: MessageToolOptions): AnyAgentTool { sandboxRoot: options?.sandboxRoot, sourceReplyDeliveryMode: sourceReplySinkDeliveryMode, inboundEventKind: options?.inboundEventKind, - inboundAudio: options?.currentInboundAudio, + inboundAudio: options?.hasCurrentInboundAudio?.() ?? options?.currentInboundAudio, abortSignal: signal, }); } catch (error) { diff --git a/src/auto-reply/reply/agent-runner-direct-runtime-config.test.ts b/src/auto-reply/reply/agent-runner-direct-runtime-config.test.ts index 8ec011d94f5d..f8028b28c0dd 100644 --- a/src/auto-reply/reply/agent-runner-direct-runtime-config.test.ts +++ b/src/auto-reply/reply/agent-runner-direct-runtime-config.test.ts @@ -135,7 +135,9 @@ function createReplyOperation(): ReplyOperation { abortByUser: vi.fn(), abortForRestart: vi.fn(), terminalRecovery: false, + acceptedSteeredInboundAudio: false, markTerminalRecovery: vi.fn(), + markAcceptedSteeredInboundAudio: vi.fn(), }; } diff --git a/src/auto-reply/reply/agent-runner-execution.test.ts b/src/auto-reply/reply/agent-runner-execution.test.ts index ca5d686b5974..c1aa9f87c994 100644 --- a/src/auto-reply/reply/agent-runner-execution.test.ts +++ b/src/auto-reply/reply/agent-runner-execution.test.ts @@ -466,6 +466,7 @@ function createMockReplyOperation(): { abortSignal: new AbortController().signal, resetTriggered: false, terminalRecovery: false, + acceptedSteeredInboundAudio: false, phase: "running", result: null, hasOwnedSessionId: vi.fn((sessionId: string) => sessionId === "session"), @@ -482,6 +483,7 @@ function createMockReplyOperation(): { abortByUser: vi.fn(() => true), abortForRestart: vi.fn(() => true), markTerminalRecovery: vi.fn(), + markAcceptedSteeredInboundAudio: vi.fn(), }, }; } diff --git a/src/auto-reply/reply/agent-runner-memory.test.ts b/src/auto-reply/reply/agent-runner-memory.test.ts index b9a323ad961d..16a4cafe5c4f 100644 --- a/src/auto-reply/reply/agent-runner-memory.test.ts +++ b/src/auto-reply/reply/agent-runner-memory.test.ts @@ -47,6 +47,7 @@ function createReplyOperation(): TestReplyOperation { abortSignal: new AbortController().signal, resetTriggered: false, terminalRecovery: false, + acceptedSteeredInboundAudio: false, phase: "queued", result: null, hasOwnedSessionId: vi.fn((sessionId: string) => sessionId === "session"), @@ -65,6 +66,7 @@ function createReplyOperation(): TestReplyOperation { abortByUser: vi.fn(() => true), abortForRestart: vi.fn(() => true), markTerminalRecovery: vi.fn(), + markAcceptedSteeredInboundAudio: vi.fn(), }; } diff --git a/src/auto-reply/reply/agent-runner.media-paths.test.ts b/src/auto-reply/reply/agent-runner.media-paths.test.ts index b666990c818c..fb40656aca5a 100644 --- a/src/auto-reply/reply/agent-runner.media-paths.test.ts +++ b/src/auto-reply/reply/agent-runner.media-paths.test.ts @@ -7,7 +7,10 @@ import type { EmbeddedAgentQueueMessageOutcome } from "../../agents/embedded-age import type { OpenClawConfig } from "../../config/types.openclaw.js"; import type { TemplateContext } from "../templating.js"; import type { FollowupRun, QueueSettings } from "./queue.js"; -import type { ReplyOperation } from "./reply-run-registry.js"; +import { + createReplyOperation as createRegisteredReplyOperation, + type ReplyOperation, +} from "./reply-run-registry.js"; import { createMockFollowupRun, createMockTypingController } from "./test-helpers.js"; const runEmbeddedAgentMock = vi.fn(); @@ -439,6 +442,45 @@ describe("runReplyAgent media path normalization", () => { expect(enqueueFollowupRunMock).not.toHaveBeenCalled(); }); + it("latches audio only after the active reply operation accepts the steer", async () => { + const operation = createRegisteredReplyOperation({ + sessionKey: "agent:main:whatsapp:direct:chat-1", + sessionId: "session", + resetTriggered: false, + }); + operation.setPhase("running"); + expect(operation.acceptedSteeredInboundAudio).toBe(false); + queueEmbeddedAgentMessageWithOutcomeAsyncMock.mockImplementation(async (sessionId: string) => ({ + queued: true, + sessionId, + target: "embedded_run", + gatewayHealth: "live", + })); + + await runReplyAgent( + makeRunReplyAgentParams({ + replyOperation: operation, + sessionKey: "agent:main:whatsapp:direct:chat-1", + resolvedQueue: { mode: "steer" } as QueueSettings, + shouldSteer: true, + shouldFollowup: true, + isActive: true, + followupRun: { + ...createMockFollowupRun({ prompt: "summarize the audio" }), + currentInboundAudio: true, + } as unknown as FollowupRun, + }), + ); + + expect(operation.acceptedSteeredInboundAudio).toBe(true); + expect(queueEmbeddedAgentMessageWithOutcomeAsyncMock).toHaveBeenLastCalledWith( + "session", + "summarize the audio", + { steeringMode: "all" }, + ); + expect(enqueueFollowupRunMock).not.toHaveBeenCalled(); + }); + it("queues active prompts in followup mode without steering", async () => { await runReplyAgent( makeRunReplyAgentParams({ diff --git a/src/auto-reply/reply/agent-runner.ts b/src/auto-reply/reply/agent-runner.ts index f249e2e59483..c159f747e53c 100644 --- a/src/auto-reply/reply/agent-runner.ts +++ b/src/auto-reply/reply/agent-runner.ts @@ -1261,9 +1261,9 @@ export async function runReplyAgent(params: { }; if (effectiveShouldSteer && isActive) { - const steerSessionId = - (sessionKey ? replyRunRegistry.resolveSessionId(sessionKey) : undefined) ?? - followupRun.run.sessionId; + const activeReplyOperation = + providedReplyOperation ?? (sessionKey ? replyRunRegistry.get(sessionKey) : undefined); + const steerSessionId = activeReplyOperation?.sessionId ?? followupRun.run.sessionId; const steerOutcome = await queueEmbeddedAgentMessageWithOutcomeAsync( steerSessionId, followupRun.prompt, @@ -1276,6 +1276,9 @@ export async function runReplyAgent(params: { }, ); if (steerOutcome.queued) { + if (followupRun.currentInboundAudio === true) { + activeReplyOperation?.markAcceptedSteeredInboundAudio(); + } await touchActiveSessionEntry(); typing.cleanup(); return undefined; diff --git a/src/auto-reply/reply/dispatch-from-config.test.ts b/src/auto-reply/reply/dispatch-from-config.test.ts index cf55b332371d..a9e55e13ac86 100644 --- a/src/auto-reply/reply/dispatch-from-config.test.ts +++ b/src/auto-reply/reply/dispatch-from-config.test.ts @@ -1600,6 +1600,43 @@ describe("dispatchReplyFromConfig", () => { ); }); + it("uses accepted steered inbound audio for final TTS", async () => { + setNoAbort(); + ttsMocks.state.synthesizeFinalAudio = true; + const dispatcher = createDispatcher(); + const ctx = buildTestCtx({ + Provider: "whatsapp", + Surface: "whatsapp", + SessionKey: "agent:main:whatsapp:direct:chat-1", + BodyForAgent: "text turn", + }); + const replyResolver = vi.fn(async (_ctx: MsgContext, opts?: GetReplyOptions) => { + const operation = ( + opts as + | { + replyOperation?: ReturnType; + } + | undefined + )?.replyOperation; + expect(operation?.acceptedSteeredInboundAudio).toBe(false); + operation?.markAcceptedSteeredInboundAudio(); + return { text: "reply to steered audio" } satisfies ReplyPayload; + }); + + await dispatchReplyFromConfig({ + ctx, + cfg: automaticDirectReplyConfig, + dispatcher, + replyResolver, + }); + + const finalTtsCall = ttsMocks.maybeApplyTtsToPayload.mock.calls.find( + ([params]) => (params as { kind?: string }).kind === "final", + )?.[0] as { inboundAudio?: boolean } | undefined; + expect(finalTtsCall?.inboundAudio).toBe(true); + expect(firstFinalReplyPayload(dispatcher)?.mediaUrl).toBe("https://example.com/tts-synth.opus"); + }); + it("passes reply policy to routed block delivery", async () => { setNoAbort(); mocks.routeReply.mockClear(); diff --git a/src/auto-reply/reply/dispatch-from-config.ts b/src/auto-reply/reply/dispatch-from-config.ts index e45c3fa76542..6884fb184d55 100644 --- a/src/auto-reply/reply/dispatch-from-config.ts +++ b/src/auto-reply/reply/dispatch-from-config.ts @@ -1462,6 +1462,8 @@ export async function dispatchReplyFromConfig( let dispatchLifecycleAbortController: AbortController | undefined; let preDispatchLifecycleInterrupted = false; const dispatchLifecycleWork = new Set>(); + const hasInboundAudioForTts = () => + inboundAudio || dispatchReplyOperation?.acceptedSteeredInboundAudio === true; const trackDispatchLifecycleWork = (work: Promise) => { if (!dispatchReplyOperation && !preDispatchLifecycleAdmission) { return; @@ -2800,7 +2802,7 @@ export async function dispatchReplyFromConfig( cfg, channel: deliveryChannel, kind: "final", - inboundAudio, + inboundAudio: hasInboundAudioForTts(), ttsAuto: sessionTtsAuto, agentId: sessionAgentId, accountId: replyRoute.accountId, @@ -3525,7 +3527,7 @@ export async function dispatchReplyFromConfig( cfg, channel: deliveryChannel, kind: "tool", - inboundAudio, + inboundAudio: hasInboundAudioForTts(), ttsAuto: sessionTtsAuto, agentId: sessionAgentId, accountId: replyRoute.accountId, @@ -3774,7 +3776,7 @@ export async function dispatchReplyFromConfig( cfg, channel: deliveryChannel, kind: "block", - inboundAudio, + inboundAudio: hasInboundAudioForTts(), ttsAuto: sessionTtsAuto, agentId: sessionAgentId, accountId: replyRoute.accountId, @@ -3981,7 +3983,7 @@ export async function dispatchReplyFromConfig( cfg, channel: deliveryChannel, kind: "final", - inboundAudio, + inboundAudio: hasInboundAudioForTts(), ttsAuto: sessionTtsAuto, agentId: sessionAgentId, accountId: replyRoute.accountId, diff --git a/src/auto-reply/reply/reply-run-registry.ts b/src/auto-reply/reply/reply-run-registry.ts index 8f231059e31a..35c57d46cc03 100644 --- a/src/auto-reply/reply/reply-run-registry.ts +++ b/src/auto-reply/reply/reply-run-registry.ts @@ -1,10 +1,10 @@ // Tracks active reply runs so stop, queue, and status commands can coordinate. import { normalizeOptionalString } from "@openclaw/normalization-core/string-coerce"; -import { createAbortError } from "../../infra/abort-signal.js"; import { createAgentRunRestartAbortError, isAgentRunRestartAbortReason, } from "../../agents/run-termination.js"; +import { createAbortError } from "../../infra/abort-signal.js"; import { markDiagnosticEmbeddedRunEnded, markDiagnosticEmbeddedRunStarted, @@ -81,6 +81,11 @@ export type ReplyOperation = { * sibling recovery already in flight, not the proven stale leftover. */ readonly terminalRecovery: boolean; + /** + * Sticky fact for audio accepted into this operation after its originating turn. + * Final delivery reads it because the original dispatch context cannot change. + */ + readonly acceptedSteeredInboundAudio: boolean; readonly phase: ReplyOperationPhase; readonly result: ReplyOperationResult | null; /** True when this operation has owned the supplied session ID. */ @@ -88,6 +93,7 @@ export type ReplyOperation = { setPhase(next: "queued" | "preflight_compacting" | "memory_flushing" | "running"): void; /** Mark this operation as an in-flight terminal-session recovery. */ markTerminalRecovery(): void; + markAcceptedSteeredInboundAudio(): void; updateSessionId(nextSessionId: string): void; attachBackend(handle: ReplyBackendHandle): void; detachBackend(handle: ReplyBackendHandle): void; @@ -464,6 +470,7 @@ export function createReplyOperation(params: { let stateCleared = false; let retainFailureUntilComplete = false; let terminalRecovery = false; + let acceptedSteeredInboundAudio = false; const upstreamAbortSignal = params.upstreamAbortSignal; let upstreamAbortHandler: (() => void) | undefined; const detachUpstreamAbort = () => { @@ -546,6 +553,9 @@ export function createReplyOperation(params: { get terminalRecovery() { return terminalRecovery; }, + get acceptedSteeredInboundAudio() { + return acceptedSteeredInboundAudio; + }, get phase() { return phase; }, @@ -565,6 +575,9 @@ export function createReplyOperation(params: { markTerminalRecovery() { terminalRecovery = true; }, + markAcceptedSteeredInboundAudio() { + acceptedSteeredInboundAudio = true; + }, updateSessionId(nextSessionId) { if (result) { return;