Skip to content

Commit 1a96d26

Browse files
mcaxtrmcaxtr
committed
fix: preserve steered audio for inbound TTS
1 parent d84a8b1 commit 1a96d26

12 files changed

Lines changed: 146 additions & 9 deletions

‎src/agents/embedded-agent-runner/run-state.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -29,6 +29,7 @@ export type EmbeddedAgentQueueHandle = {
2929

3030
export type EmbeddedAgentQueueMessageOptions = {
3131
steeringMode?: "all";
32+
currentInboundAudio?: boolean;
3233
debounceMs?: number;
3334
deliveryTimeoutMs?: number;
3435
waitForTranscriptCommit?: boolean;

‎src/agents/embedded-agent-runner/run/attempt.ts‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -3709,6 +3709,9 @@ export async function runEmbeddedAttempt(
37093709
activeSession.agent.steeringMode = options.steeringMode;
37103710
}
37113711
await steerActiveSessionWithOptionalDeliveryWait(activeSession, text, options);
3712+
if (options?.currentInboundAudio === true) {
3713+
params.replyOperation?.markSteeredInboundAudio();
3714+
}
37123715
},
37133716
isStreaming: () => activeSession.isStreaming,
37143717
isCompacting: () => subscription.isCompacting(),

‎src/agents/embedded-agent-runner/runs.test.ts‎

Lines changed: 23 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -285,6 +285,26 @@ describe("embedded-agent runner run registry", () => {
285285
);
286286
});
287287

288+
it("passes current inbound audio only on accepted async queue delivery", async () => {
289+
const queueMessage = vi.fn(async () => {});
290+
setActiveEmbeddedRun("session-audio", {
291+
...createRunHandle(),
292+
queueMessage,
293+
});
294+
295+
const outcome = await queueEmbeddedAgentMessageWithOutcomeAsync(
296+
"session-audio",
297+
"continue from audio",
298+
{ steeringMode: "all", currentInboundAudio: true },
299+
);
300+
301+
expect(outcome.queued).toBe(true);
302+
expect(queueMessage).toHaveBeenCalledWith("continue from audio", {
303+
steeringMode: "all",
304+
currentInboundAudio: true,
305+
});
306+
});
307+
288308
it("rejects transcript-commit waits for active handles without support", async () => {
289309
const queueMessage = vi.fn(async () => {});
290310
setActiveEmbeddedRun("session-no-transcript-wait", {
@@ -342,7 +362,9 @@ describe("embedded-agent runner run registry", () => {
342362
});
343363
expect(outcome.enqueuedAtMs).toEqual(expect.any(Number));
344364
expect(outcome.deliveredAtMs).toBeUndefined();
345-
expect(queueMessage).toHaveBeenCalledWith("completion from child");
365+
expect(queueMessage).toHaveBeenCalledWith("completion from child", {
366+
waitForTranscriptCommit: true,
367+
});
346368
});
347369

348370
it("force-clears an aborted run that does not drain", async () => {
@@ -588,5 +610,4 @@ describe("embedded-agent runner run registry", () => {
588610
clearActiveEmbeddedRun("session-snapshot", handle);
589611
expect(getActiveEmbeddedRunSnapshot("session-snapshot")).toBeUndefined();
590612
});
591-
592613
});

‎src/agents/embedded-agent-runner/runs.ts‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -369,7 +369,7 @@ function prepareEmbeddedAgentQueueMessage(
369369
): PreparedEmbeddedAgentQueueMessage {
370370
const handle = ACTIVE_EMBEDDED_RUNS.get(sessionId);
371371
if (!handle) {
372-
const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text);
372+
const queuedReplyRunMessage = queueReplyRunMessage(sessionId, text, options);
373373
if (queuedReplyRunMessage) {
374374
logMessageQueued({ sessionId, source: "embedded-agent-runner" });
375375
return {

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

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -423,8 +423,10 @@ function createMockReplyOperation(): {
423423
resetTriggered: false,
424424
phase: "running",
425425
result: null,
426+
currentInboundAudio: false,
426427
setPhase: vi.fn(),
427428
updateSessionId: updateSessionIdMock,
429+
markSteeredInboundAudio: vi.fn(),
428430
attachBackend: vi.fn(),
429431
detachBackend: vi.fn(),
430432
retainFailureUntilComplete: retainFailureUntilCompleteMock,

‎src/auto-reply/reply/agent-runner-memory.test.ts‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -48,8 +48,10 @@ function createReplyOperation(): TestReplyOperation {
4848
resetTriggered: false,
4949
phase: "queued",
5050
result: null,
51+
currentInboundAudio: false,
5152
setPhase: vi.fn<ReplyOperation["setPhase"]>(),
5253
updateSessionId: vi.fn<ReplyOperation["updateSessionId"]>(),
54+
markSteeredInboundAudio: vi.fn(),
5355
attachBackend: vi.fn(),
5456
detachBackend: vi.fn(),
5557
retainFailureUntilComplete: vi.fn(),

‎src/auto-reply/reply/agent-runner.media-paths.test.ts‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -434,6 +434,38 @@ describe("runReplyAgent media path normalization", () => {
434434
expect(enqueueFollowupRunMock).not.toHaveBeenCalled();
435435
});
436436

437+
it("passes steered inbound audio metadata to the active run queue", async () => {
438+
queueEmbeddedAgentMessageWithOutcomeAsyncMock.mockImplementation(async (sessionId: string) => ({
439+
queued: true,
440+
sessionId,
441+
target: "embedded_run",
442+
gatewayHealth: "live",
443+
}));
444+
445+
await runReplyAgent(
446+
makeRunReplyAgentParams({
447+
resolvedQueue: { mode: "steer" } as QueueSettings,
448+
shouldSteer: true,
449+
shouldFollowup: true,
450+
isStreaming: true,
451+
followupRun: {
452+
...createMockFollowupRun({ prompt: "summarize the audio" }),
453+
currentInboundAudio: true,
454+
} as unknown as FollowupRun,
455+
}),
456+
);
457+
458+
expect(queueEmbeddedAgentMessageWithOutcomeAsyncMock).toHaveBeenLastCalledWith(
459+
"session",
460+
"summarize the audio",
461+
{
462+
steeringMode: "all",
463+
currentInboundAudio: true,
464+
},
465+
);
466+
expect(enqueueFollowupRunMock).not.toHaveBeenCalled();
467+
});
468+
437469
it("queues active prompts in followup mode without steering", async () => {
438470
await runReplyAgent(
439471
makeRunReplyAgentParams({

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

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1287,6 +1287,7 @@ export async function runReplyAgent(params: {
12871287
followupRun.prompt,
12881288
{
12891289
steeringMode: "all",
1290+
...(followupRun.currentInboundAudio === true ? { currentInboundAudio: true } : {}),
12901291
...(resolvedQueue.debounceMs !== undefined ? { debounceMs: resolvedQueue.debounceMs } : {}),
12911292
},
12921293
);

‎src/auto-reply/reply/dispatch-from-config.test.ts‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1600,6 +1600,43 @@ describe("dispatchReplyFromConfig", () => {
16001600
expect(replyResolver).toHaveBeenCalledTimes(1);
16011601
});
16021602

1603+
it("uses steered inbound audio marked on the reply operation for final TTS", async () => {
1604+
setNoAbort();
1605+
ttsMocks.state.synthesizeFinalAudio = true;
1606+
const cfg = automaticDirectReplyConfig;
1607+
const dispatcher = createDispatcher();
1608+
const ctx = buildTestCtx({
1609+
Provider: "whatsapp",
1610+
Surface: "whatsapp",
1611+
SessionKey: "agent:main:whatsapp:direct:chat-1",
1612+
BodyForAgent: "text turn",
1613+
});
1614+
const replyResolver = vi.fn(async (_ctx: MsgContext, opts?: GetReplyOptions) => {
1615+
const operation = (
1616+
opts as
1617+
| {
1618+
replyOperation?: {
1619+
currentInboundAudio: boolean;
1620+
markSteeredInboundAudio(): void;
1621+
};
1622+
}
1623+
| undefined
1624+
)?.replyOperation;
1625+
expect(operation?.currentInboundAudio).toBe(false);
1626+
operation?.markSteeredInboundAudio();
1627+
return { text: "reply to steered audio" } satisfies ReplyPayload;
1628+
});
1629+
1630+
await dispatchReplyFromConfig({ ctx, cfg, dispatcher, replyResolver });
1631+
1632+
const finalTtsCall = ttsMocks.maybeApplyTtsToPayload.mock.calls.find(
1633+
([params]) => (params as { kind?: string }).kind === "final",
1634+
)?.[0] as { inboundAudio?: boolean } | undefined;
1635+
expect(finalTtsCall?.inboundAudio).toBe(true);
1636+
const finalPayload = firstFinalReplyPayload(dispatcher);
1637+
expect(finalPayload?.mediaUrl).toBe("https://example.com/tts-synth.opus");
1638+
});
1639+
16031640
it("lets a different Slack DM routed thread reach reply resolution while another thread is active", async () => {
16041641
setNoAbort();
16051642
const sessionKey = "agent:main:slack:direct:U1";

‎src/auto-reply/reply/dispatch-from-config.ts‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1310,6 +1310,8 @@ export async function dispatchReplyFromConfig(
13101310
let dispatchReplyOperation: ReplyOperation | undefined;
13111311
let dispatchAbortOperation: ReplyOperation | undefined;
13121312
let preDispatchAbortOperation: ReplyOperation | undefined;
1313+
const resolveFinalTtsInboundAudio = () =>
1314+
inboundAudio || dispatchReplyOperation?.currentInboundAudio === true;
13131315
type DispatchReplyOperationAcquisition = { status: "ready" } | { status: "busy" };
13141316
const ensureDispatchReplyOperation = async (
13151317
phase: "pre_dispatch" | "dispatch",
@@ -2351,7 +2353,7 @@ export async function dispatchReplyFromConfig(
23512353
cfg,
23522354
channel: deliveryChannel,
23532355
kind: "final",
2354-
inboundAudio,
2356+
inboundAudio: resolveFinalTtsInboundAudio(),
23552357
ttsAuto: sessionTtsAuto,
23562358
agentId: sessionAgentId,
23572359
accountId: replyRoute.accountId,
@@ -3410,7 +3412,7 @@ export async function dispatchReplyFromConfig(
34103412
cfg,
34113413
channel: deliveryChannel,
34123414
kind: "final",
3413-
inboundAudio,
3415+
inboundAudio: resolveFinalTtsInboundAudio(),
34143416
ttsAuto: sessionTtsAuto,
34153417
agentId: sessionAgentId,
34163418
accountId: replyRoute.accountId,

0 commit comments

Comments
 (0)