From 0bd2d99694dec182b0abf1a430156c4fb9fbc358 Mon Sep 17 00:00:00 2001 From: pandec Date: Mon, 24 Aug 2026 15:57:27 +0200 Subject: [PATCH 1/7] feat(voice): let agents reply with a voice recording Agents could only produce text; the fork's message-listening TTS was a manual per-message action. This adds a voice_reply MCP tool on the injected t3-code server: the agent writes a script for the ear, the server synthesizes it through ElevenLabs and stages the MP3, and provider-runtime ingestion attaches it to the turn's final assistant message when the turn completes (a voice-only turn publishes the transcript as the message). The projector owns the speech row via the message-sent event, so live clients get the update and replay rebuilds it. Web and desktop render the player as the main message content with the written reply behind a toggle; mobile does the same. MCP credentials now carry per-capability scopes, gated by ELEVENLABS_API_KEY and a new Settings -> Extras switch. Implemented by Claude Fable 5 via Claude Code. --- README.md | 1 + .../src/features/threads/ThreadFeed.tsx | 139 ++++++++++--- .../OrchestrationEngineHarness.integration.ts | 2 + apps/server/src/mcp/McpHttpServer.ts | 11 +- apps/server/src/mcp/McpInvocationContext.ts | 2 +- .../server/src/mcp/McpSessionRegistry.test.ts | 5 + apps/server/src/mcp/McpSessionRegistry.ts | 3 +- .../server/src/mcp/toolkits/voice/handlers.ts | 22 ++ apps/server/src/mcp/toolkits/voice/tools.ts | 27 +++ .../Layers/ProjectionPipeline.ts | 99 +++++++++ .../Layers/ProjectionSnapshotQuery.test.ts | 1 + .../Layers/ProjectionSnapshotQuery.ts | 11 +- .../Layers/ProviderRuntimeIngestion.test.ts | 174 ++++++++++++++++ .../Layers/ProviderRuntimeIngestion.ts | 74 +++++++ apps/server/src/orchestration/decider.ts | 1 + apps/server/src/persistence/Migrations.ts | 2 + .../050_ProjectionMessageSpeechOrigin.ts | 11 + .../src/provider/Layers/ProviderService.ts | 30 ++- apps/server/src/server.ts | 5 + apps/server/src/voice/AgentVoiceReply.ts | 195 ++++++++++++++++++ apps/server/src/voice/MessageSpeech.ts | 34 ++- apps/server/src/voice/elevenLabsTts.ts | 50 +++++ .../src/components/chat/MessagesTimeline.tsx | 69 +++++-- .../settings/ExtrasSettingsPanel.tsx | 35 +++- packages/contracts/src/orchestration.ts | 10 +- packages/contracts/src/previewAutomation.ts | 2 +- packages/contracts/src/settings.test.ts | 23 ++- packages/contracts/src/settings.ts | 5 + packages/contracts/src/voice.ts | 57 +++++ packages/shared/src/serverSettings.test.ts | 13 +- 30 files changed, 1025 insertions(+), 88 deletions(-) create mode 100644 apps/server/src/mcp/toolkits/voice/handlers.ts create mode 100644 apps/server/src/mcp/toolkits/voice/tools.ts create mode 100644 apps/server/src/persistence/Migrations/050_ProjectionMessageSpeechOrigin.ts create mode 100644 apps/server/src/voice/AgentVoiceReply.ts create mode 100644 apps/server/src/voice/elevenLabsTts.ts diff --git a/README.md b/README.md index 6c91af2c8f83..6e859a9bbfd9 100644 --- a/README.md +++ b/README.md @@ -45,6 +45,7 @@ This is a personal fork of [pingdotgg/t3code](https://github.com/pingdotgg/t3cod - **Voice dictation** — ElevenLabs-powered voice transcription in the composer, including mobile. - **Message listening** — optional spoken versions of assistant messages with playback controls; per-message summaries and speech artifacts are persisted, with mobile playback support and server-side ElevenLabs model and voice overrides. +- **Agent voice replies** — agents get a `voice_reply` tool that turns a script they write for the ear into an ElevenLabs recording, attached to their final message when the turn completes. Web and mobile then lead with the player and fold the written reply behind a "Show written reply" toggle; a turn that ends with no written message publishes the transcript as the message text. Needs `ELEVENLABS_API_KEY` on the server; the switch in Settings → Extras (on by default) withholds the tool from newly started agent sessions. ### Agents & skills diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 5943a1dfee0e..85129f3e6e6a 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -1238,31 +1238,44 @@ function renderFeedEntry( } const enterAnimated = isFreshTimestamp(message.createdAt); + const agentVoiceReply = message.speech?.origin === "agent" ? message.speech : null; + const writtenReply = + message.text.trim().length > 0 ? ( + hasNativeSelectableMarkdownText() ? ( + + ) : ( + + {message.text} + + ) + ) : null; return ( - {message.text.trim().length > 0 ? ( - hasNativeSelectableMarkdownText() ? ( - - ) : ( - - {message.text} - - ) - ) : null} + {agentVoiceReply !== null ? ( + + {writtenReply} + + ) : ( + writtenReply + )} {attachments.map((attachment) => { return ( + setTranscriptExpanded((current) => !current)} + onRetry={null} + primary + /> + {hasWrittenReply ? ( + setWrittenReplyExpanded((current) => !current)} + > + + {writtenReplyExpanded ? "Hide written reply" : "Show written reply"} + + + + ) : null} + {writtenReplyExpanded ? props.children : null} + + ); +} + function AssistantMessageMetaAndArtifacts(props: { readonly environmentId: EnvironmentId; readonly messageId: MessageId; @@ -1349,6 +1410,10 @@ function AssistantMessageMetaAndArtifacts(props: { ); const speech = sessionArtifacts.speech ?? props.persistedSpeech; const summary = sessionArtifacts.summary ?? props.persistedSummary; + // Agent voice replies render their own player above the message; the meta + // row must not offer a second one (or a regeneration that would replace the + // agent's recording with a synthesized listening version). + const isAgentVoiceReply = speech !== null && speech.origin === "agent"; const prepareSpeech = useCallback(async () => { if (preparing) return; @@ -1453,7 +1518,7 @@ function AssistantMessageMetaAndArtifacts(props: { )} ) : null} - {props.textToSpeechAvailable || speech !== null ? ( + {(props.textToSpeechAvailable || speech !== null) && !isAgentVoiceReply ? ( ) : null} - {speech !== null && expanded ? ( + {speech !== null && expanded && !isAgentVoiceReply ? ( void; - readonly onRetry: () => void; + /** null hides the regenerate action (agent recordings cannot be re-made client-side). */ + readonly onRetry: (() => void) | null; + /** Agent voice replies render the player as the message's main content. */ + readonly primary?: boolean; }) { const { blocked, speed } = useListeningPlaybackSnapshot(); // The transport sits on `bg-foreground`, so its glyph has to come from the @@ -1610,7 +1678,12 @@ function AssistantSpeechPlayer(props: { ]); return ( - + - Listening version + + {props.primary ? "Voice reply" : "Listening version"} + {audioUrlState._tag === "Failure" ? ( - The audio file is unavailable. Regenerate it to listen again. + {props.onRetry === null + ? "The audio file is unavailable." + : "The audio file is unavailable. Regenerate it to listen again."} - - Regenerate - + {props.onRetry !== null ? ( + + Regenerate + + ) : null} ) : audioUrl === null ? ( @@ -1689,7 +1768,7 @@ function AssistantSpeechPlayer(props: { onPress={props.onToggleTranscript} > - View listening transcript + {props.primary ? "View transcript" : "View listening transcript"} const issued = yield* registry.issue({ threadId: ThreadId.make(`thread-${hostname}`), providerInstanceId: ProviderInstanceId.make("codex"), + capabilities: new Set(["preview"] as const), }); expect(issued.config.endpoint).toBe(expectedEndpoint); } @@ -81,6 +83,7 @@ it.effect("expires credentials once their session stops showing signs of life", const issued = yield* registry.issue({ threadId: ThreadId.make("thread-2"), providerInstanceId: ProviderInstanceId.make("claude"), + capabilities: new Set(["preview"] as const), }); const token = issued.config.authorizationHeader.replace(/^Bearer\s+/, ""); timestamp += 101; @@ -96,6 +99,7 @@ it.effect("keeps a credential alive across turns that never touch an MCP tool", const issued = yield* registry.issue({ threadId, providerInstanceId: ProviderInstanceId.make("claude"), + capabilities: new Set(["preview"] as const), }); const token = issued.config.authorizationHeader.replace(/^Bearer\s+/, ""); @@ -117,6 +121,7 @@ it.effect("does not keep credentials of other threads alive", () => const issued = yield* registry.issue({ threadId: ThreadId.make("thread-4"), providerInstanceId: ProviderInstanceId.make("codex"), + capabilities: new Set(["preview"] as const), }); const token = issued.config.authorizationHeader.replace(/^Bearer\s+/, ""); diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts index f19a4f4e8c49..cd664ad530e5 100644 --- a/apps/server/src/mcp/McpSessionRegistry.ts +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -14,6 +14,7 @@ import * as McpProviderSession from "./McpProviderSession.ts"; export interface McpCredentialRequest { readonly threadId: ThreadId; readonly providerInstanceId: ProviderInstanceId; + readonly capabilities: ReadonlySet; } export interface McpIssuedCredential { @@ -128,7 +129,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( threadId: ThreadId.make(request.threadId), providerSessionId, providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), - capabilities: new Set(["preview"]), + capabilities: new Set(request.capabilities), issuedAt, }; yield* SynchronizedRef.update(state, ({ records }) => { diff --git a/apps/server/src/mcp/toolkits/voice/handlers.ts b/apps/server/src/mcp/toolkits/voice/handlers.ts new file mode 100644 index 000000000000..2fad16fb5e3a --- /dev/null +++ b/apps/server/src/mcp/toolkits/voice/handlers.ts @@ -0,0 +1,22 @@ +import * as Effect from "effect/Effect"; + +import { AgentVoiceReply } from "../../../voice/AgentVoiceReply.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { VoiceToolkit } from "./tools.ts"; + +export const VoiceToolkitHandlersLive = VoiceToolkit.toLayer({ + voice_reply: (input) => + Effect.gen(function* () { + const scope = yield* McpInvocationContext.requireMcpCapability("voice"); + const agentVoiceReply = yield* AgentVoiceReply; + const staged = yield* agentVoiceReply.stage({ + threadId: scope.threadId, + script: input.script, + }); + return { + status: "staged" as const, + transcriptChars: staged.transcript.length, + audioSizeBytes: staged.sizeBytes, + }; + }), +}); diff --git a/apps/server/src/mcp/toolkits/voice/tools.ts b/apps/server/src/mcp/toolkits/voice/tools.ts new file mode 100644 index 000000000000..6cd76425ca53 --- /dev/null +++ b/apps/server/src/mcp/toolkits/voice/tools.ts @@ -0,0 +1,27 @@ +import { + AgentVoiceReplyError, + AgentVoiceReplyInput, + AgentVoiceReplyResult, + PreviewAutomationUnavailableError, +} from "@t3tools/contracts"; +import * as Schema from "effect/Schema"; +import { Tool, Toolkit } from "effect/unstable/ai"; + +import { AgentVoiceReply } from "../../../voice/AgentVoiceReply.ts"; +import * as McpInvocationContext from "../../McpInvocationContext.ts"; + +export const VoiceReplyTool = Tool.make("voice_reply", { + description: + "Deliver your reply for this turn as a spoken recording. The audio is generated from your script and attached to your final message, where the user hears it as the primary form of your reply (your written text stays available behind a toggle). Write the script for the ear, not the eye: conversational tone, short sentences, no markdown, no code, no URLs or file paths — describe such things in words instead. Still write your normal text reply after calling this. Call at most once per turn, shortly before you finish; calling again replaces the previous recording. The recording is published only when the turn completes normally.", + parameters: AgentVoiceReplyInput, + success: AgentVoiceReplyResult, + failure: Schema.Union([AgentVoiceReplyError, PreviewAutomationUnavailableError]), + dependencies: [McpInvocationContext.McpInvocationContext, AgentVoiceReply], +}) + .annotate(Tool.Title, "Reply with a voice recording") + .annotate(Tool.Readonly, false) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, false) + .annotate(Tool.OpenWorld, false); + +export const VoiceToolkit = Toolkit.make(VoiceReplyTool); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 2738380be10d..27995ad998d3 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -2,6 +2,7 @@ import { ApprovalRequestId, type ChatAttachment, type MessageId, + type MessageSpeechAttachment, type OrchestrationEvent, type OrchestrationSessionStatus, ThreadId, @@ -544,6 +545,97 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti const path = yield* Path.Path; const serverConfig = yield* ServerConfig; + /** + * Materializes an agent-staged voice recording carried on a + * `thread.message-sent` event. The MP3 was written to the attachments + * directory before the command was dispatched, so this only records + * metadata — which also makes event replay rebuild the row correctly. + */ + const upsertMessageSpeechFromEvent = Effect.fn("upsertMessageSpeechFromEvent")( + function* (input: { + readonly messageId: MessageId; + readonly threadId: ThreadId; + readonly speech: MessageSpeechAttachment; + }) { + const { messageId, threadId, speech } = input; + const previousRows = yield* sql<{ readonly speechId: string }>` + SELECT speech_id AS "speechId" + FROM projection_message_speech + WHERE message_id = ${messageId} + LIMIT 1 + `.pipe( + Effect.catchTag("SqlError", (sqlError) => + Effect.fail( + toPersistenceSqlError("ProjectionPipeline.upsertMessageSpeech:query")(sqlError), + ), + ), + ); + yield* sql` + INSERT INTO projection_message_speech ( + message_id, + thread_id, + speech_id, + transcript, + mime_type, + size_bytes, + source_text_hash, + script_recipe_hash, + voice_id, + tts_model, + origin, + created_at + ) VALUES ( + ${messageId}, + ${threadId}, + ${speech.speechId}, + ${speech.transcript}, + ${speech.mimeType}, + ${speech.sizeBytes}, + ${speech.sourceTextHash}, + ${"agent-voice-reply"}, + ${speech.voiceId}, + ${speech.ttsModel}, + ${speech.origin}, + ${speech.createdAt} + ) + ON CONFLICT(message_id) DO UPDATE SET + thread_id = excluded.thread_id, + speech_id = excluded.speech_id, + transcript = excluded.transcript, + mime_type = excluded.mime_type, + size_bytes = excluded.size_bytes, + source_text_hash = excluded.source_text_hash, + script_recipe_hash = excluded.script_recipe_hash, + voice_id = excluded.voice_id, + tts_model = excluded.tts_model, + origin = excluded.origin, + created_at = excluded.created_at + `.pipe( + Effect.catchTag("SqlError", (sqlError) => + Effect.fail( + toPersistenceSqlError("ProjectionPipeline.upsertMessageSpeech:upsert")(sqlError), + ), + ), + ); + const previous = previousRows[0]; + if (previous && previous.speechId !== speech.speechId) { + // Same guard as the stale-speech prune: only remove a file whose id + // provably belongs to this thread's attachment namespace. + const threadSegment = toSafeThreadAttachmentSegment(threadId); + const relativePath = `${previous.speechId}.mp3`; + const parsedId = parseAttachmentIdFromRelativePath(relativePath); + const parsedThreadSegment = parsedId + ? parseThreadSegmentFromAttachmentId(parsedId) + : null; + if (threadSegment && parsedThreadSegment === threadSegment) { + yield* fileSystem + .remove(path.join(serverConfig.attachmentsDir, relativePath), { force: true }) + .pipe(Effect.ignore); + } + } + }, + ); + const applyProjectsProjection: ProjectorDefinition["apply"] = Effect.fn( "applyProjectsProjection", )(function* (event, _attachmentSideEffects) { @@ -1173,6 +1265,13 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti eventSequence: event.sequence, }); } + if (event.payload.speech !== undefined && event.payload.role === "assistant") { + yield* upsertMessageSpeechFromEvent({ + messageId: event.payload.messageId, + threadId: event.payload.threadId, + speech: event.payload.speech, + }); + } return; } diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index eb2db4547ade..5888b262ca70 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -546,6 +546,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { transcript: "A persisted transcript.", mimeType: "audio/mpeg", sizeBytes: 42, + origin: "user", createdAt: "2026-02-24T00:00:05.500Z", }, }, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 2bbe9727707c..4ad3eb062f22 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -4,6 +4,7 @@ import { IsoDateTime, MessageId, MessageInputOrigin, + MessageSpeechOrigin, NonNegativeInt, OrchestrationCheckpointFile, OrchestrationProposedPlanId, @@ -125,6 +126,7 @@ const ProjectionThreadMessageArtifactDbRowSchema = Schema.Struct({ speechSizeBytes: Schema.NullOr(NonNegativeInt), speechCreatedAt: Schema.NullOr(IsoDateTime), speechSourceTextHash: Schema.NullOr(Schema.String), + speechOrigin: Schema.NullOr(MessageSpeechOrigin), }); const ProjectionThreadProposedPlanDbRowSchema = ProjectionThreadProposedPlan; const ProjectionThreadDbRowSchema = ProjectionThread.mapFields( @@ -446,7 +448,10 @@ function mapMessageRow( row.speechMimeType === "audio/mpeg" && row.speechSizeBytes !== null && row.speechCreatedAt !== null && - row.speechSourceTextHash === currentSourceTextHash + // An agent recording's transcript is authored independently of the + // message text, so the staleness gate that protects user-requested + // listening versions does not apply to it. + (row.speechOrigin === "agent" || row.speechSourceTextHash === currentSourceTextHash) ? { speech: { messageId: row.messageId, @@ -454,6 +459,7 @@ function mapMessageRow( transcript: row.speechTranscript, mimeType: row.speechMimeType, sizeBytes: row.speechSizeBytes, + origin: row.speechOrigin ?? "user", createdAt: row.speechCreatedAt, }, } @@ -1215,7 +1221,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { speech.mime_type AS "speechMimeType", speech.size_bytes AS "speechSizeBytes", speech.created_at AS "speechCreatedAt", - speech.source_text_hash AS "speechSourceTextHash" + speech.source_text_hash AS "speechSourceTextHash", + speech.origin AS "speechOrigin" FROM projection_thread_messages AS messages LEFT JOIN projection_message_summary AS summary ON summary.message_id = messages.message_id diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index a64df87b453d..c77964f1c6d0 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -23,6 +23,7 @@ import { DEFAULT_PROVIDER_INTERACTION_MODE, EventId, MessageId, + type MessageSpeechAttachment, type OrchestrationCommand, ProjectId, ProviderItemId, @@ -62,6 +63,7 @@ import { ProviderRuntimeIngestionService } from "../Services/ProviderRuntimeInge import { ProjectionSnapshotQuery } from "../Services/ProjectionSnapshotQuery.ts"; import { ServerConfig } from "../../config.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; +import * as AgentVoiceReply from "../../voice/AgentVoiceReply.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; function makeTestServerSettingsLayer(overrides: Partial = {}) { @@ -279,6 +281,7 @@ describe("ProviderRuntimeIngestion", () => { async function createHarness(options?: { serverSettings?: Partial; + agentVoiceReply?: AgentVoiceReply.AgentVoiceReplyShape; providerInstanceHealth?: ProviderInstanceHealthShape; threadTitle?: string; }) { @@ -323,6 +326,11 @@ describe("ProviderRuntimeIngestion", () => { ? ProviderInstanceHealthLive : Layer.succeed(ProviderInstanceHealth, options.providerInstanceHealth), ), + Layer.provideMerge( + options?.agentVoiceReply === undefined + ? AgentVoiceReply.layerNoop + : Layer.succeed(AgentVoiceReply.AgentVoiceReply, options.agentVoiceReply), + ), Layer.provideMerge(NodeServices.layer), ); runtime = ManagedRuntime.make(layer); @@ -1337,6 +1345,172 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.streaming).toBe(false); }); + function makeStagedVoiceReply( + overrides: Partial = {}, + ): MessageSpeechAttachment { + return { + speechId: "thread-1-agent-voice", + transcript: "spoken summary", + mimeType: "audio/mpeg", + sizeBytes: 1234, + sourceTextHash: "hash-of-spoken-summary", + voiceId: "voice-1", + ttsModel: "eleven_flash_v2_5", + origin: "agent", + createdAt: "2026-01-01T00:00:00.000Z", + ...overrides, + } as MessageSpeechAttachment; + } + + function makeFakeAgentVoiceReply(initialStaged: MessageSpeechAttachment | null) { + let staged = initialStaged; + const removedAudio: string[] = []; + const shape: AgentVoiceReply.AgentVoiceReplyShape = { + available: true, + stage: () => Effect.die(new Error("stage is not exercised by ingestion tests")), + takeStaged: () => + Effect.sync(() => { + const current = staged ?? undefined; + staged = null; + return current; + }), + discardStaged: () => + Effect.sync(() => { + if (staged !== null) removedAudio.push(staged.speechId); + staged = null; + }), + removeStagedAudio: (entry) => + Effect.sync(() => { + removedAudio.push(entry.speechId); + }), + }; + return { shape, removedAudio, hasStaged: () => staged !== null }; + } + + it("attaches a staged agent voice reply to the turn's final assistant message", async () => { + const staged = makeStagedVoiceReply(); + const fake = makeFakeAgentVoiceReply(staged); + const harness = await createHarness({ agentVoiceReply: fake.shape }); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-voice-delta"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-voice"), + itemId: asItemId("item-voice"), + payload: { streamKind: "assistant_text", delta: "written reply" }, + }); + harness.emit({ + type: "item.completed", + eventId: asEventId("evt-voice-item-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-voice"), + itemId: asItemId("item-voice"), + payload: { itemType: "assistant_message", status: "completed" }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-voice-turn-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-voice"), + status: "completed", + }); + + const thread = await waitForThread(harness.readModel, (entry) => + entry.messages.some( + (message: ProviderRuntimeTestMessage) => + message.id === "assistant:item-voice" && message.speech !== undefined, + ), + ); + const message = thread.messages.find( + (entry: ProviderRuntimeTestMessage) => entry.id === "assistant:item-voice", + ); + expect(message?.text).toBe("written reply"); + expect(message?.speech?.origin).toBe("agent"); + expect(message?.speech?.speechId).toBe(staged.speechId); + expect(message?.speech?.transcript).toBe("spoken summary"); + expect(fake.removedAudio).toEqual([]); + }); + + it("publishes the transcript as the message when a voice-only turn completes", async () => { + const staged = makeStagedVoiceReply({ speechId: "thread-1-agent-voice-only" }); + const fake = makeFakeAgentVoiceReply(staged); + const harness = await createHarness({ agentVoiceReply: fake.shape }); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-voice-only-turn-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-voice-only"), + status: "completed", + }); + + const thread = await waitForThread(harness.readModel, (entry) => + entry.messages.some( + (message: ProviderRuntimeTestMessage) => + message.role === "assistant" && message.speech?.origin === "agent", + ), + ); + const message = thread.messages.find( + (entry: ProviderRuntimeTestMessage) => entry.speech?.origin === "agent", + ); + expect(message?.text).toBe("spoken summary"); + expect(message?.streaming).toBe(false); + expect(message?.speech?.speechId).toBe("thread-1-agent-voice-only"); + }); + + it("drops a staged voice reply when the turn does not complete normally", async () => { + const staged = makeStagedVoiceReply({ speechId: "thread-1-agent-voice-dropped" }); + const fake = makeFakeAgentVoiceReply(staged); + const harness = await createHarness({ agentVoiceReply: fake.shape }); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-voice-dropped-delta"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-voice-dropped"), + itemId: asItemId("item-voice-dropped"), + payload: { streamKind: "assistant_text", delta: "partial reply" }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-voice-dropped-turn-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-voice-dropped"), + status: "failed", + errorMessage: "turn failed", + }); + + const thread = await waitForThread( + harness.readModel, + (entry) => + entry.session?.status === "error" && + entry.messages.some( + (message: ProviderRuntimeTestMessage) => + message.id === "assistant:item-voice-dropped" && !message.streaming, + ), + ); + expect(fake.removedAudio).toEqual(["thread-1-agent-voice-dropped"]); + expect( + thread.messages.every((message: ProviderRuntimeTestMessage) => message.speech === undefined), + ).toBe(true); + }); + it("preserves completed tool metadata on projected tool activities", async () => { const harness = await createHarness(); const now = "2026-01-01T00:00:00.000Z"; diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index a6854255058c..74077136a867 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -55,6 +55,7 @@ import { projectActivityPayload } from "../ActivityPayloadProjection.ts"; import { forkParked } from "../../serverActivation.ts"; import { ServerSettingsService } from "../../serverSettings.ts"; import { canReplaceThreadTitle } from "../threadTitles.ts"; +import { AgentVoiceReply } from "../../voice/AgentVoiceReply.ts"; const providerTurnKey = (threadId: ThreadId, turnId: TurnId) => `${threadId}:${turnId}`; const providerTaskKey = (threadId: ThreadId, taskId: string) => `${threadId}:${taskId}`; @@ -175,6 +176,19 @@ function findMessageById( return undefined; } +function findLastAssistantMessageForTurn( + messages: ReadonlyArray, + turnId: TurnId, +): OrchestrationMessage | undefined { + for (let index = messages.length - 1; index >= 0; index -= 1) { + const message = messages[index]; + if (message?.role === "assistant" && sameId(message.turnId, turnId)) { + return message; + } + } + return undefined; +} + function findProposedPlanById( proposedPlans: ReadonlyArray< Pick @@ -931,6 +945,7 @@ const make = Effect.gen(function* () { const providerInstanceRegistry = yield* ProviderInstanceRegistry; const projectionTurnRepository = yield* ProjectionTurnRepository; const serverSettingsService = yield* ServerSettingsService; + const agentVoiceReply = yield* AgentVoiceReply; const providerCommandId = (event: ProviderRuntimeEvent, tag: string) => crypto.randomUUIDv4.pipe( Effect.map((uuid) => CommandId.make(`provider:${event.eventId}:${tag}:${uuid}`)), @@ -2066,9 +2081,68 @@ const make = Effect.gen(function* () { turnId, updatedAt: now, }); + + // A recording staged through the voice_reply MCP tool attaches to + // the turn's last assistant message once the turn completes + // normally; any other outcome throws the recording away, since an + // interrupted or failed turn no longer matches its script. + const stagedVoiceReply = yield* agentVoiceReply.takeStaged(thread.id); + if (stagedVoiceReply) { + if (normalizeRuntimeTurnState(event.payload.state) === "completed") { + const finalizedMessageIds = Array.from(assistantMessageIds); + const lastFinalizedMessageId = finalizedMessageIds[finalizedMessageIds.length - 1]; + const lastProjectedMessageId = findLastAssistantMessageForTurn(messages, turnId)?.id; + const targetMessageId = lastFinalizedMessageId ?? lastProjectedMessageId; + if (targetMessageId !== undefined) { + yield* orchestrationEngine.dispatch({ + type: "thread.message.assistant.complete", + commandId: yield* providerCommandId(event, "agent-voice-reply-attach"), + threadId: thread.id, + messageId: targetMessageId, + turnId, + speech: stagedVoiceReply, + createdAt: now, + }); + } else { + // A voice-only turn: surface the transcript as the message + // text so the recording has a durable, searchable home. + const voiceMessageId = MessageId.make(`assistant:voice-reply:${event.eventId}`); + yield* orchestrationEngine.dispatch({ + type: "thread.message.assistant.delta", + commandId: yield* providerCommandId(event, "agent-voice-reply-message"), + threadId: thread.id, + messageId: voiceMessageId, + delta: stagedVoiceReply.transcript, + turnId, + createdAt: now, + }); + yield* orchestrationEngine.dispatch({ + type: "thread.message.assistant.complete", + commandId: yield* providerCommandId(event, "agent-voice-reply-complete"), + threadId: thread.id, + messageId: voiceMessageId, + turnId, + speech: stagedVoiceReply, + createdAt: now, + }); + } + } else { + yield* agentVoiceReply.removeStagedAudio(stagedVoiceReply); + } + } + } else { + // An untargeted completion cannot prove which turn it ends, so a + // staged recording has nothing safe to bind to. + yield* agentVoiceReply.discardStaged(thread.id); } } + if (event.type === "session.exited") { + // A staged voice reply cannot outlive its session: the turn that + // staged it will never complete now. + yield* agentVoiceReply.discardStaged(thread.id); + } + if (event.type === "session.exited" && shouldApplyThreadLifecycle) { yield* clearTurnStateForSession(thread.id); } diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index a8187b4b6b36..7c166282111d 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1495,6 +1495,7 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" messageId: command.messageId, role: "assistant", text: "", + ...(command.speech !== undefined ? { speech: command.speech } : {}), turnId: command.turnId ?? null, streaming: false, createdAt: command.createdAt, diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 3142ac5eaddb..bd8da6952e55 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -67,6 +67,7 @@ import Migration0047 from "./Migrations/047_ProjectionProjectsDefaultThreadEnvMo import Migration0048 from "./Migrations/048_ProjectionProjectFaviconPath.ts"; // Upstream shipped this as 041; renumbered after the fork's migration history. import Migration0049 from "./Migrations/049_AuthSessionClientConnection.ts"; +import Migration0050 from "./Migrations/050_ProjectionMessageSpeechOrigin.ts"; /** * Migration loader with all migrations defined inline. @@ -128,6 +129,7 @@ export const migrationEntries = [ [47, "ProjectionProjectsDefaultThreadEnvMode", Migration0047], [48, "ProjectionProjectFaviconPath", Migration0048], [49, "AuthSessionClientConnection", Migration0049], + [50, "ProjectionMessageSpeechOrigin", Migration0050], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/050_ProjectionMessageSpeechOrigin.ts b/apps/server/src/persistence/Migrations/050_ProjectionMessageSpeechOrigin.ts new file mode 100644 index 000000000000..a94c3ff70100 --- /dev/null +++ b/apps/server/src/persistence/Migrations/050_ProjectionMessageSpeechOrigin.ts @@ -0,0 +1,11 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql` + ALTER TABLE projection_message_speech + ADD COLUMN origin TEXT NOT NULL DEFAULT 'user' + `; +}); diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 9afaa7ddbacb..8f1fabab7da9 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -26,12 +26,14 @@ import { type ProviderSession, } from "@t3tools/contracts"; import { causeErrorTag } from "@t3tools/shared/observability"; +import * as Config from "effect/Config"; import * as DateTime from "effect/DateTime"; import * as Deferred from "effect/Deferred"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as PubSub from "effect/PubSub"; +import * as Redacted from "effect/Redacted"; import * as Ref from "effect/Ref"; import * as Schema from "effect/Schema"; import * as SchemaIssue from "effect/SchemaIssue"; @@ -64,6 +66,7 @@ import { type EventNdjsonLogger } from "./EventNdjsonLogger.ts"; import * as ProviderEventLoggers from "./ProviderEventLoggers.ts"; import * as AnalyticsService from "../../telemetry/AnalyticsService.ts"; import { isExistingDirectory } from "../../pathExpansion.ts"; +import type * as McpInvocationContext from "../../mcp/McpInvocationContext.ts"; import * as McpProviderSession from "../../mcp/McpProviderSession.ts"; import * as McpSessionRegistry from "../../mcp/McpSessionRegistry.ts"; import * as ServerSettings from "../../serverSettings.ts"; @@ -440,19 +443,34 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( * "off" silently becoming "on" would violate the user's stated choice, * whereas the reverse costs an agent one toolset and is visible immediately. */ - const agentBrowserAccessEnabled = serverSettings.getSettings.pipe( - Effect.map((settings) => settings.enableAgentBrowserAccess), + // Whether the server can synthesize agent voice replies at all. Read once; + // per-session enablement additionally consults the settings toggle below. + const elevenLabsApiKey = yield* Config.redacted("ELEVENLABS_API_KEY").pipe(Config.option); + const agentVoiceReplyAvailable = + Option.isSome(elevenLabsApiKey) && Redacted.value(elevenLabsApiKey.value).trim().length > 0; + + const mcpSessionCapabilities = serverSettings.getSettings.pipe( + Effect.map( + (settings): ReadonlySet => + new Set([ + ...(settings.enableAgentBrowserAccess ? (["preview"] as const) : []), + ...(agentVoiceReplyAvailable && settings.voice.enableAgentVoiceReplies + ? (["voice"] as const) + : []), + ]), + ), Effect.catch((cause) => Effect.logWarning( - "Could not read server settings; withholding agent browser access for this session.", + "Could not read server settings; withholding agent MCP toolsets for this session.", { cause }, - ).pipe(Effect.as(false)), + ).pipe(Effect.as>(new Set())), ), ); const prepareMcpSession = (threadId: ThreadId, providerInstanceId: ProviderInstanceId) => Effect.gen(function* () { - if (!(yield* agentBrowserAccessEnabled)) { + const capabilities = yield* mcpSessionCapabilities; + if (capabilities.size === 0) { // Revoke as well as clear. Every other prepare path reaches // `issueActiveMcpCredential`, which revokes the thread first, so // skipping it here would leave a previously issued bearer token valid @@ -463,7 +481,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( yield* Effect.sync(() => McpProviderSession.clearMcpProviderSession(threadId)); return undefined; } - const credential = yield* issueMcpCredential({ threadId, providerInstanceId }); + const credential = yield* issueMcpCredential({ threadId, providerInstanceId, capabilities }); if (credential) { yield* Effect.sync(() => McpProviderSession.setMcpProviderSession(credential.config)); } diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 804055375a53..028f3f54c6b7 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -132,6 +132,7 @@ import { orchestrationHttpApiLayer } from "./orchestration/http.ts"; import * as TurnStartBootstrap from "./orchestration/Services/TurnStartBootstrap.ts"; import * as VoiceTranscription from "./voice/VoiceTranscription.ts"; import * as MessageSpeech from "./voice/MessageSpeech.ts"; +import * as AgentVoiceReply from "./voice/AgentVoiceReply.ts"; import { voiceHttpApiLayer } from "./voice/http.ts"; import * as MessageSummary from "./messageArtifacts/MessageSummary.ts"; import { messageArtifactsHttpApiLayer } from "./messageArtifacts/http.ts"; @@ -430,6 +431,10 @@ const RuntimeCoreDependenciesLive = Layer.mergeAll( ), ), CheckpointingLayerLive, + // Shared between the voice_reply MCP handler (stages recordings) and + // provider-runtime ingestion (attaches them at turn completion), so it + // must be one instance below both. + AgentVoiceReply.layer, ), ), // Shared bootstrap program for thread.turn.start commands, consumed by both diff --git a/apps/server/src/voice/AgentVoiceReply.ts b/apps/server/src/voice/AgentVoiceReply.ts new file mode 100644 index 000000000000..d81d2d8a09e4 --- /dev/null +++ b/apps/server/src/voice/AgentVoiceReply.ts @@ -0,0 +1,195 @@ +// @effect-diagnostics nodeBuiltinImport:off +import * as NodeCrypto from "node:crypto"; + +import { + AGENT_VOICE_REPLY_MAX_SCRIPT_CHARS, + AgentVoiceReplyError, + type MessageSpeechAttachment, + type ThreadId, +} from "@t3tools/contracts"; +import * as Config from "effect/Config"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FileSystem from "effect/FileSystem"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Redacted from "effect/Redacted"; +import * as SynchronizedRef from "effect/SynchronizedRef"; +import { HttpClient } from "effect/unstable/http"; + +import { createAttachmentId } from "../attachmentStore.ts"; +import { resolveAttachmentRelativePath } from "../attachmentPaths.ts"; +import * as ServerConfig from "../config.ts"; +import { ServerSettingsService } from "../serverSettings.ts"; +import { SPEECH_MIME_TYPE, synthesizeElevenLabsSpeech } from "./elevenLabsTts.ts"; +import { + DEFAULT_ELEVENLABS_TTS_MODEL, + DEFAULT_ELEVENLABS_TTS_VOICE_ID, + getElevenLabsTtsCharacterLimit, + resolveMessageSpeechVoiceSetting, +} from "./MessageSpeech.ts"; + +/** + * Agent-staged voice replies. The `voice_reply` MCP tool synthesizes a + * recording mid-turn and parks it here; provider-runtime ingestion collects it + * when the turn completes and attaches it to the turn's final assistant + * message. One staged reply per thread — a second call replaces the first. + * + * The MP3 is written to the attachments directory at stage time so the later + * attach command can stay metadata-only, mirroring how user image attachments + * are persisted by the normalizer before their event is recorded. + */ +export interface AgentVoiceReplyShape { + readonly available: boolean; + readonly stage: (input: { + readonly threadId: ThreadId; + readonly script: string; + }) => Effect.Effect; + /** Removes and returns the staged reply without touching its audio file. */ + readonly takeStaged: (threadId: ThreadId) => Effect.Effect; + /** Removes the staged reply and deletes its audio file, if any. */ + readonly discardStaged: (threadId: ThreadId) => Effect.Effect; + /** Deletes the audio file behind a reply that will never be attached. */ + readonly removeStagedAudio: (staged: MessageSpeechAttachment) => Effect.Effect; +} + +export class AgentVoiceReply extends Context.Service()( + "t3/voice/AgentVoiceReply", +) {} + +/** Inert instance for tests and harnesses that do not exercise voice replies. */ +export const layerNoop = Layer.succeed(AgentVoiceReply, { + available: false, + stage: () => Effect.fail(new AgentVoiceReplyError({ reason: "unavailable" })), + takeStaged: () => Effect.succeed(undefined), + discardStaged: () => Effect.void, + removeStagedAudio: () => Effect.void, +}); + +export const layer = Layer.effect( + AgentVoiceReply, + Effect.gen(function* () { + const apiKey = yield* Config.redacted("ELEVENLABS_API_KEY").pipe(Config.option); + const envTtsModel = yield* Config.string("ELEVENLABS_TTS_MODEL").pipe( + Config.withDefault(DEFAULT_ELEVENLABS_TTS_MODEL), + ); + const envVoiceId = yield* Config.string("ELEVENLABS_TTS_VOICE_ID").pipe( + Config.withDefault(DEFAULT_ELEVENLABS_TTS_VOICE_ID), + ); + const available = Option.isSome(apiKey) && Redacted.value(apiKey.value).trim().length > 0; + const httpClient = yield* HttpClient.HttpClient; + const fileSystem = yield* FileSystem.FileSystem; + const serverConfig = yield* ServerConfig.ServerConfig; + const serverSettings = yield* ServerSettingsService; + const staged = yield* SynchronizedRef.make>( + new Map(), + ); + + const resolveSpeechPath = (speechId: string) => + resolveAttachmentRelativePath({ + attachmentsDir: serverConfig.attachmentsDir, + relativePath: `${speechId}.mp3`, + }); + + const removeAudioFile = (speechId: string) => { + const path = resolveSpeechPath(speechId); + return path ? fileSystem.remove(path, { force: true }).pipe(Effect.ignore) : Effect.void; + }; + + const stage: AgentVoiceReplyShape["stage"] = Effect.fn("AgentVoiceReply.stage")( + function* (input) { + if (!available || Option.isNone(apiKey)) { + return yield* new AgentVoiceReplyError({ reason: "unavailable" }); + } + + const settings = yield* serverSettings.getSettings.pipe( + Effect.mapError(() => new AgentVoiceReplyError({ reason: "storage_failed" })), + ); + const ttsModel = resolveMessageSpeechVoiceSetting( + settings.voice.ttsModelId, + envTtsModel, + DEFAULT_ELEVENLABS_TTS_MODEL, + ); + const voiceId = resolveMessageSpeechVoiceSetting( + settings.voice.ttsVoiceId, + envVoiceId, + DEFAULT_ELEVENLABS_TTS_VOICE_ID, + ); + + const script = input.script.trim(); + const characterLimit = Math.min( + AGENT_VOICE_REPLY_MAX_SCRIPT_CHARS, + getElevenLabsTtsCharacterLimit(ttsModel), + ); + if (script.length === 0 || script.length > characterLimit) { + return yield* new AgentVoiceReplyError({ reason: "script_too_long" }); + } + + const audioBytes = yield* synthesizeElevenLabsSpeech({ + httpClient, + apiKey: apiKey.value, + voiceId, + ttsModel, + text: script, + }).pipe(Effect.mapError(() => new AgentVoiceReplyError({ reason: "provider_failed" }))); + + const speechId = createAttachmentId(input.threadId); + const speechPath = speechId ? resolveSpeechPath(speechId) : null; + if (!speechId || !speechPath) { + return yield* new AgentVoiceReplyError({ reason: "storage_failed" }); + } + yield* fileSystem.makeDirectory(serverConfig.attachmentsDir, { recursive: true }).pipe( + Effect.andThen(fileSystem.writeFile(speechPath, audioBytes)), + Effect.mapError(() => new AgentVoiceReplyError({ reason: "storage_failed" })), + ); + + const createdAt = DateTime.formatIso(yield* DateTime.now); + const attachment: MessageSpeechAttachment = { + speechId, + transcript: script as MessageSpeechAttachment["transcript"], + mimeType: SPEECH_MIME_TYPE, + sizeBytes: audioBytes.byteLength as MessageSpeechAttachment["sizeBytes"], + sourceTextHash: NodeCrypto.createHash("sha256") + .update(script, "utf8") + .digest("hex") as MessageSpeechAttachment["sourceTextHash"], + voiceId: voiceId as MessageSpeechAttachment["voiceId"], + ttsModel: ttsModel as MessageSpeechAttachment["ttsModel"], + origin: "agent", + createdAt: createdAt as MessageSpeechAttachment["createdAt"], + }; + + const replaced = yield* SynchronizedRef.modify(staged, (entries) => { + const previous = entries.get(input.threadId); + const next = new Map(entries); + next.set(input.threadId, attachment); + return [previous, next] as const; + }); + if (replaced) { + yield* removeAudioFile(replaced.speechId); + } + return attachment; + }, + ); + + const takeStaged: AgentVoiceReplyShape["takeStaged"] = (threadId) => + SynchronizedRef.modify(staged, (entries) => { + const current = entries.get(threadId); + if (!current) return [undefined, entries] as const; + const next = new Map(entries); + next.delete(threadId); + return [current, next] as const; + }); + + return AgentVoiceReply.of({ + available, + stage, + takeStaged, + discardStaged: (threadId) => + takeStaged(threadId).pipe( + Effect.flatMap((entry) => (entry ? removeAudioFile(entry.speechId) : Effect.void)), + ), + removeStagedAudio: (entry) => removeAudioFile(entry.speechId), + }); + }), +); diff --git a/apps/server/src/voice/MessageSpeech.ts b/apps/server/src/voice/MessageSpeech.ts index c28d88f1eb56..b7717ed78b0b 100644 --- a/apps/server/src/voice/MessageSpeech.ts +++ b/apps/server/src/voice/MessageSpeech.ts @@ -17,7 +17,7 @@ import * as Option from "effect/Option"; import * as Redacted from "effect/Redacted"; import * as Schema from "effect/Schema"; import * as SqlClient from "effect/unstable/sql/SqlClient"; -import { HttpBody, HttpClient, HttpClientResponse } from "effect/unstable/http"; +import { HttpClient } from "effect/unstable/http"; import { createAttachmentId } from "../attachmentStore.ts"; import { resolveAttachmentRelativePath } from "../attachmentPaths.ts"; @@ -25,14 +25,12 @@ import * as ServerConfig from "../config.ts"; import { ServerSettingsService } from "../serverSettings.ts"; import { TextGeneration } from "../textGeneration/TextGeneration.ts"; import { makeMessageArtifactLockCoordinator } from "../messageArtifacts/lock.ts"; +import { SPEECH_MIME_TYPE, synthesizeElevenLabsSpeech } from "./elevenLabsTts.ts"; export { makeMessageArtifactLockCoordinator as makeMessageSpeechLockCoordinator }; -const ELEVENLABS_TEXT_TO_SPEECH_URL = "https://api.elevenlabs.io/v1/text-to-speech"; -const ELEVENLABS_TEXT_TO_SPEECH_TIMEOUT = "120 seconds"; export const DEFAULT_ELEVENLABS_TTS_MODEL = "eleven_flash_v2_5"; export const DEFAULT_ELEVENLABS_TTS_VOICE_ID = "JBFqnCBsd6RMkjVDRZzb"; -const SPEECH_MIME_TYPE = "audio/mpeg" as const; const SPEECH_SCRIPT_RECIPE_VERSION = 2; interface MessageSpeechCacheRow { @@ -322,24 +320,13 @@ export const layer = Layer.effect( return yield* new MessageSpeechError({ reason: "script_failed" }); } - const audioBuffer = yield* httpClient - .post( - `${ELEVENLABS_TEXT_TO_SPEECH_URL}/${encodeURIComponent(voiceId)}?output_format=mp3_44100_128`, - { - headers: { "xi-api-key": Redacted.value(apiKey.value) }, - body: HttpBody.jsonUnsafe({ text: transcript, model_id: ttsModel }), - }, - ) - .pipe( - Effect.flatMap(HttpClientResponse.filterStatusOk), - Effect.flatMap((response) => response.arrayBuffer), - Effect.timeout(ELEVENLABS_TEXT_TO_SPEECH_TIMEOUT), - Effect.mapError(() => new MessageSpeechError({ reason: "provider_failed" })), - ); - const audioBytes = new Uint8Array(audioBuffer); - if (audioBytes.byteLength === 0) { - return yield* new MessageSpeechError({ reason: "provider_failed" }); - } + const audioBytes = yield* synthesizeElevenLabsSpeech({ + httpClient, + apiKey: apiKey.value, + voiceId, + ttsModel, + text: transcript, + }).pipe(Effect.mapError(() => new MessageSpeechError({ reason: "provider_failed" }))); const speechId = createAttachmentId(message.threadId); const speechPath = speechId ? resolveSpeechPath(speechId) : null; @@ -368,6 +355,7 @@ export const layer = Layer.effect( script_recipe_hash, voice_id, tts_model, + origin, created_at ) SELECT @@ -381,6 +369,7 @@ export const layer = Layer.effect( ${scriptRecipeHash}, ${voiceId}, ${ttsModel}, + ${"user"}, ${createdAt} WHERE EXISTS ( SELECT 1 @@ -404,6 +393,7 @@ export const layer = Layer.effect( script_recipe_hash = excluded.script_recipe_hash, voice_id = excluded.voice_id, tts_model = excluded.tts_model, + origin = excluded.origin, created_at = excluded.created_at RETURNING message_id AS "messageId", diff --git a/apps/server/src/voice/elevenLabsTts.ts b/apps/server/src/voice/elevenLabsTts.ts new file mode 100644 index 000000000000..7ccb99a978fd --- /dev/null +++ b/apps/server/src/voice/elevenLabsTts.ts @@ -0,0 +1,50 @@ +import * as Effect from "effect/Effect"; +import * as Redacted from "effect/Redacted"; +import * as Schema from "effect/Schema"; +import { HttpBody, type HttpClient, HttpClientResponse } from "effect/unstable/http"; + +const ELEVENLABS_TEXT_TO_SPEECH_URL = "https://api.elevenlabs.io/v1/text-to-speech"; +const ELEVENLABS_TEXT_TO_SPEECH_TIMEOUT = "120 seconds"; + +export const SPEECH_MIME_TYPE = "audio/mpeg" as const; + +export class ElevenLabsTtsError extends Schema.TaggedErrorClass()( + "ElevenLabsTtsError", + { + reason: Schema.Literals(["request_failed", "empty_audio"]), + }, +) {} + +/** + * One ElevenLabs text-to-speech request, shared by the on-demand listening + * version and agent voice replies. Fails with `ElevenLabsTtsError` on any + * transport, status, or empty-body problem so callers can map it onto their + * own error vocabulary. + */ +export const synthesizeElevenLabsSpeech = (input: { + readonly httpClient: HttpClient.HttpClient; + readonly apiKey: Redacted.Redacted; + readonly voiceId: string; + readonly ttsModel: string; + readonly text: string; +}): Effect.Effect => + input.httpClient + .post( + `${ELEVENLABS_TEXT_TO_SPEECH_URL}/${encodeURIComponent(input.voiceId)}?output_format=mp3_44100_128`, + { + headers: { "xi-api-key": Redacted.value(input.apiKey) }, + body: HttpBody.jsonUnsafe({ text: input.text, model_id: input.ttsModel }), + }, + ) + .pipe( + Effect.flatMap(HttpClientResponse.filterStatusOk), + Effect.flatMap((response) => response.arrayBuffer), + Effect.timeout(ELEVENLABS_TEXT_TO_SPEECH_TIMEOUT), + Effect.mapError(() => new ElevenLabsTtsError({ reason: "request_failed" })), + Effect.flatMap((buffer) => { + const bytes = new Uint8Array(buffer); + return bytes.byteLength === 0 + ? Effect.fail(new ElevenLabsTtsError({ reason: "empty_audio" })) + : Effect.succeed(bytes); + }), + ); diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 869a1c13b80f..6056d2c47ea1 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -1264,7 +1264,10 @@ function AssistantTimelineRow({ row }: { row: Extract("idle"); - const [speechExpanded, setSpeechExpanded] = useState(false); + // null = untouched: agent voice replies start expanded, listening versions + // start collapsed. + const [speechExpandedState, setSpeechExpandedState] = useState(null); + const [writtenReplyExpanded, setWrittenReplyExpanded] = useState(false); const [summaryPhase, setSummaryPhase] = useState<"idle" | "preparing">("idle"); const [summaryExpanded, setSummaryExpanded] = useState(false); const readSessionArtifacts = useCallback( @@ -1287,6 +1290,11 @@ function AssistantTimelineRow({ row }: { row: Extract { if (speech !== null) { - setSpeechExpanded((expanded) => !expanded); + setSpeechExpandedState(!speechExpanded); return; } await prepareSpeech(); - }, [prepareSpeech, speech]); + }, [prepareSpeech, speech, speechExpanded]); const onToggleSummary = useCallback(async () => { if (summary !== null) { @@ -1372,14 +1380,35 @@ function AssistantTimelineRow({ row }: { row: Extract
- + {isAgentVoiceReply && speech !== null && speechExpanded ? ( + void prepareSpeech()} + primary + /> + ) : null} + {isAgentVoiceReply && speechExpanded ? ( + + ) : null} + {showMessageText ? ( + + ) : null}
) : null} - {speech !== null && speechExpanded ? ( + {speech !== null && speechExpanded && !isAgentVoiceReply ? ( void; + /** Agent voice replies render the player as the message's main content. */ + primary?: boolean; }) { const audioRef = useRef(null); const { blocked, speed } = useListeningPlaybackSnapshot(); @@ -1528,10 +1560,15 @@ function AssistantSpeechPlayer({ ); return ( -
+
- Listening version + {primary ? "Voice reply" : "Listening version"}
{audioUrlState._tag === "Failure" ? (
@@ -1573,7 +1610,7 @@ function AssistantSpeechPlayer({ )}
- View listening transcript + {primary ? "View transcript" : "View listening transcript"}

{speech.transcript} diff --git a/apps/web/src/components/settings/ExtrasSettingsPanel.tsx b/apps/web/src/components/settings/ExtrasSettingsPanel.tsx index ca52861fe47e..d2f2e5406852 100644 --- a/apps/web/src/components/settings/ExtrasSettingsPanel.tsx +++ b/apps/web/src/components/settings/ExtrasSettingsPanel.tsx @@ -941,10 +941,39 @@ function VoiceExtrasSection() { } /> + + updateSettings({ + voice: { + enableAgentVoiceReplies: DEFAULT_UNIFIED_SETTINGS.voice.enableAgentVoiceReplies, + }, + }) + } + /> + ) : null + } + control={ + + updateSettings({ voice: { enableAgentVoiceReplies: Boolean(checked) } }) + } + aria-label="Allow agent voice replies" + /> + } + /> +

- Speech playback needs ELEVENLABS_API_KEY in the server's environment. When set, - these fields override the server's ELEVENLABS_TTS_MODEL and{" "} - ELEVENLABS_TTS_VOICE_ID environment variables. + Speech playback and agent voice replies need ELEVENLABS_API_KEY in the server's + environment. When set, these fields override the server's ELEVENLABS_TTS_MODEL{" "} + and ELEVENLABS_TTS_VOICE_ID environment variables.

); diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index 0162d7c25dc9..2bb38619fd84 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -23,7 +23,11 @@ import { TurnId, } from "./baseSchemas.ts"; import { ProviderDriverKind, ProviderInstanceId } from "./providerInstance.ts"; -import { MessageSpeechSynthesisResult, MessageSummaryResult } from "./voice.ts"; +import { + MessageSpeechAttachment, + MessageSpeechSynthesisResult, + MessageSummaryResult, +} from "./voice.ts"; export const ORCHESTRATION_WS_METHODS = { dispatchCommand: "orchestration.dispatchCommand", @@ -1128,6 +1132,9 @@ const ThreadMessageAssistantCompleteCommand = Schema.Struct({ threadId: ThreadId, messageId: MessageId, turnId: Schema.optional(TurnId), + // An agent-staged voice recording to attach to the completed message. The + // MP3 already sits in the attachments directory; this is metadata only. + speech: Schema.optional(MessageSpeechAttachment), createdAt: IsoDateTime, }); @@ -1403,6 +1410,7 @@ export const ThreadMessageSentPayload = Schema.Struct({ text: Schema.String, attachments: Schema.optional(Schema.Array(ChatAttachment)), inputOrigin: Schema.optional(MessageInputOrigin), + speech: Schema.optional(MessageSpeechAttachment), turnId: Schema.NullOr(TurnId), streaming: Schema.Boolean, createdAt: IsoDateTime, diff --git a/packages/contracts/src/previewAutomation.ts b/packages/contracts/src/previewAutomation.ts index e33615fa4c05..5b2d028c78f7 100644 --- a/packages/contracts/src/previewAutomation.ts +++ b/packages/contracts/src/previewAutomation.ts @@ -634,7 +634,7 @@ export type PreviewAutomationResponse = typeof PreviewAutomationResponse.Type; export class PreviewAutomationUnavailableError extends Schema.TaggedErrorClass()( "PreviewAutomationUnavailableError", { - capability: Schema.Literal("preview"), + capability: Schema.Literals(["preview", "voice"]), environmentId: EnvironmentId, threadId: ThreadId, providerSessionId: TrimmedNonEmptyString, diff --git a/packages/contracts/src/settings.test.ts b/packages/contracts/src/settings.test.ts index dadb7e53805f..3404e4b9d9c1 100644 --- a/packages/contracts/src/settings.test.ts +++ b/packages/contracts/src/settings.test.ts @@ -361,19 +361,38 @@ describe("ClientSettings extras", () => { describe("ServerSettings.voice", () => { it("defaults to unset, so the server keeps its env/default resolution", () => { - expect(DEFAULT_SERVER_SETTINGS.voice).toEqual({ ttsModelId: "", ttsVoiceId: "" }); - expect(decodeServerSettings({}).voice).toEqual({ ttsModelId: "", ttsVoiceId: "" }); + expect(DEFAULT_SERVER_SETTINGS.voice).toEqual({ + ttsModelId: "", + ttsVoiceId: "", + enableAgentVoiceReplies: true, + }); + expect(decodeServerSettings({}).voice).toEqual({ + ttsModelId: "", + ttsVoiceId: "", + enableAgentVoiceReplies: true, + }); }); it("trims values in both the settings and the patch", () => { expect(decodeServerSettings({ voice: { ttsModelId: " eleven_v3 " } }).voice).toEqual({ ttsModelId: "eleven_v3", ttsVoiceId: "", + enableAgentVoiceReplies: true, }); expect(decodeServerSettingsPatch({ voice: { ttsVoiceId: " abc " } })).toEqual({ voice: { ttsVoiceId: "abc" }, }); }); + + it("round-trips the agent voice replies toggle through the patch", () => { + expect( + decodeServerSettings({ voice: { enableAgentVoiceReplies: false } }).voice + .enableAgentVoiceReplies, + ).toBe(false); + expect(decodeServerSettingsPatch({ voice: { enableAgentVoiceReplies: false } })).toEqual({ + voice: { enableAgentVoiceReplies: false }, + }); + }); }); describe("ServerSettings.providerInstances (slice-2 invariant)", () => { diff --git a/packages/contracts/src/settings.ts b/packages/contracts/src/settings.ts index 29946425c2b6..68a7c3ab5b26 100644 --- a/packages/contracts/src/settings.ts +++ b/packages/contracts/src/settings.ts @@ -858,6 +858,10 @@ export type ObservabilitySettings = typeof ObservabilitySettings.Type; export const VoiceSettings = Schema.Struct({ ttsModelId: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), ttsVoiceId: TrimmedString.pipe(Schema.withDecodingDefault(Effect.succeed(""))), + // Exposes the voice_reply MCP tool to agent sessions so they can answer + // with a staged recording. Only effective while the server has an + // ELEVENLABS_API_KEY; defaults to on so setting the key is enough. + enableAgentVoiceReplies: Schema.Boolean.pipe(Schema.withDecodingDefault(Effect.succeed(true))), }); export type VoiceSettings = typeof VoiceSettings.Type; @@ -1244,6 +1248,7 @@ export const ServerSettingsPatch = Schema.Struct({ Schema.Struct({ ttsModelId: Schema.optionalKey(TrimmedString), ttsVoiceId: Schema.optionalKey(TrimmedString), + enableAgentVoiceReplies: Schema.optionalKey(Schema.Boolean), }), ), providers: Schema.optionalKey( diff --git a/packages/contracts/src/voice.ts b/packages/contracts/src/voice.ts index 093698317609..9b890ac21b0e 100644 --- a/packages/contracts/src/voice.ts +++ b/packages/contracts/src/voice.ts @@ -38,6 +38,15 @@ export const MESSAGE_SPEECH_MAX_SCRIPT_CHARS = 40_000; export const MESSAGE_SUMMARY_MAX_SOURCE_CHARS = 120_000; export const MESSAGE_SUMMARY_MAX_TEXT_CHARS = 12_000; +/** + * Who produced a message's speech artifact. "user" is the on-demand listening + * version a client requested; "agent" is a recording the agent staged itself + * through the voice_reply MCP tool. Agent recordings are presented as the + * primary form of the message; user ones stay an opt-in secondary artifact. + */ +export const MessageSpeechOrigin = Schema.Literals(["user", "agent"]); +export type MessageSpeechOrigin = typeof MessageSpeechOrigin.Type; + export const MessageSpeechSynthesisRequest = Schema.Struct({ messageId: MessageId, }); @@ -49,10 +58,58 @@ export const MessageSpeechSynthesisResult = Schema.Struct({ transcript: TrimmedNonEmptyString.check(Schema.isMaxLength(MESSAGE_SPEECH_MAX_SCRIPT_CHARS)), mimeType: Schema.Literal("audio/mpeg"), sizeBytes: NonNegativeInt, + // Optional so payloads persisted before agent voice replies still decode; + // absent means "user". + origin: Schema.optional(MessageSpeechOrigin), createdAt: IsoDateTime, }); export type MessageSpeechSynthesisResult = typeof MessageSpeechSynthesisResult.Type; +export const AGENT_VOICE_REPLY_MAX_SCRIPT_CHARS = 10_000; + +/** + * Speech metadata carried on an assistant-message completion. The audio bytes + * live in the server attachments directory under `.mp3` (written + * before the command is dispatched, mirroring how user image attachments are + * persisted by the normalizer); the event stream only ever sees metadata. + */ +export const MessageSpeechAttachment = Schema.Struct({ + speechId: TrimmedNonEmptyString, + transcript: TrimmedNonEmptyString.check(Schema.isMaxLength(MESSAGE_SPEECH_MAX_SCRIPT_CHARS)), + mimeType: Schema.Literal("audio/mpeg"), + sizeBytes: NonNegativeInt, + sourceTextHash: TrimmedNonEmptyString, + voiceId: TrimmedNonEmptyString, + ttsModel: TrimmedNonEmptyString, + origin: MessageSpeechOrigin, + createdAt: IsoDateTime, +}); +export type MessageSpeechAttachment = typeof MessageSpeechAttachment.Type; + +export const AgentVoiceReplyInput = Schema.Struct({ + script: TrimmedNonEmptyString.check(Schema.isMaxLength(AGENT_VOICE_REPLY_MAX_SCRIPT_CHARS)), +}); +export type AgentVoiceReplyInput = typeof AgentVoiceReplyInput.Type; + +export const AgentVoiceReplyResult = Schema.Struct({ + status: Schema.Literal("staged"), + transcriptChars: NonNegativeInt, + audioSizeBytes: NonNegativeInt, +}); +export type AgentVoiceReplyResult = typeof AgentVoiceReplyResult.Type; + +export class AgentVoiceReplyError extends Schema.TaggedErrorClass()( + "AgentVoiceReplyError", + { + reason: Schema.Literals([ + "unavailable", + "script_too_long", + "provider_failed", + "storage_failed", + ]), + }, +) {} + export const MessageSummaryRequest = Schema.Struct({ messageId: MessageId, }); diff --git a/packages/shared/src/serverSettings.test.ts b/packages/shared/src/serverSettings.test.ts index a2898de52c84..a40bd0daf9ca 100644 --- a/packages/shared/src/serverSettings.test.ts +++ b/packages/shared/src/serverSettings.test.ts @@ -66,18 +66,27 @@ describe("serverSettings helpers", () => { it("merges the voice group one field at a time", () => { const current = { ...DEFAULT_SERVER_SETTINGS, - voice: { ttsModelId: "eleven_v3", ttsVoiceId: "voice-a" }, + voice: { ttsModelId: "eleven_v3", ttsVoiceId: "voice-a", enableAgentVoiceReplies: true }, }; - // A patch touching one field leaves the other alone... + // A patch touching one field leaves the others alone... expect(applyServerSettingsPatch(current, { voice: { ttsVoiceId: "voice-b" } }).voice).toEqual({ ttsModelId: "eleven_v3", ttsVoiceId: "voice-b", + enableAgentVoiceReplies: true, }); // ...and an empty string clears a field back to "unset". expect(applyServerSettingsPatch(current, { voice: { ttsModelId: "" } }).voice).toEqual({ ttsModelId: "", ttsVoiceId: "voice-a", + enableAgentVoiceReplies: true, + }); + expect( + applyServerSettingsPatch(current, { voice: { enableAgentVoiceReplies: false } }).voice, + ).toEqual({ + ttsModelId: "eleven_v3", + ttsVoiceId: "voice-a", + enableAgentVoiceReplies: false, }); expect(applyServerSettingsPatch(current, {}).voice).toEqual(current.voice); }); From aec9804029c45e5c707d643da2a09ab976fd055a Mon Sep 17 00:00:00 2001 From: pandec Date: Mon, 24 Aug 2026 16:27:38 +0200 Subject: [PATCH 2/7] fix(voice): harden staged-recording lifecycle after review Bind each staged recording to the turn that was active when voice_reply ran, so a stale completion for another turn can no longer consume or delete it and an interrupted turn cannot leak its recording into the next one; turn.aborted now discards a matching entry. The staged entry is cleared only after the attach commands land instead of being taken up front. The projector no longer deletes replaced audio files (unsafe under transaction rollback and event replay), and the web primary player drops the Regenerate action that would have replaced an agent recording with re-synthesized text. Review findings by gpt-5.6-sol; fixes by Claude Fable 5 via Claude Code. --- .../Layers/ProjectionPipeline.ts | 32 +------ .../Layers/ProviderRuntimeIngestion.test.ts | 60 ++++++++++--- .../Layers/ProviderRuntimeIngestion.ts | 48 +++++++--- apps/server/src/voice/AgentVoiceReply.ts | 87 +++++++++++++------ .../src/components/chat/MessagesTimeline.tsx | 23 +++-- 5 files changed, 169 insertions(+), 81 deletions(-) diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index 27995ad998d3..91bea7df5445 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -550,6 +550,10 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti * `thread.message-sent` event. The MP3 was written to the attachments * directory before the command was dispatched, so this only records * metadata — which also makes event replay rebuild the row correctly. + * Deliberately no file deletion here: a projector apply can run inside a + * transaction that later rolls back, and replay revisits old events, so + * destroying a replaced recording's file from this path would be unsafe. + * A superseded MP3 is left for thread-deletion cleanup instead. */ const upsertMessageSpeechFromEvent = Effect.fn("upsertMessageSpeechFromEvent")( function* (input: { @@ -558,18 +562,6 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti readonly speech: MessageSpeechAttachment; }) { const { messageId, threadId, speech } = input; - const previousRows = yield* sql<{ readonly speechId: string }>` - SELECT speech_id AS "speechId" - FROM projection_message_speech - WHERE message_id = ${messageId} - LIMIT 1 - `.pipe( - Effect.catchTag("SqlError", (sqlError) => - Effect.fail( - toPersistenceSqlError("ProjectionPipeline.upsertMessageSpeech:query")(sqlError), - ), - ), - ); yield* sql` INSERT INTO projection_message_speech ( message_id, @@ -617,22 +609,6 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti ), ), ); - const previous = previousRows[0]; - if (previous && previous.speechId !== speech.speechId) { - // Same guard as the stale-speech prune: only remove a file whose id - // provably belongs to this thread's attachment namespace. - const threadSegment = toSafeThreadAttachmentSegment(threadId); - const relativePath = `${previous.speechId}.mp3`; - const parsedId = parseAttachmentIdFromRelativePath(relativePath); - const parsedThreadSegment = parsedId - ? parseThreadSegmentFromAttachmentId(parsedId) - : null; - if (threadSegment && parsedThreadSegment === threadSegment) { - yield* fileSystem - .remove(path.join(serverConfig.attachmentsDir, relativePath), { force: true }) - .pipe(Effect.ignore); - } - } }, ); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts index c77964f1c6d0..9dd7af3ea7c1 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.test.ts @@ -1362,34 +1362,33 @@ describe("ProviderRuntimeIngestion", () => { } as MessageSpeechAttachment; } - function makeFakeAgentVoiceReply(initialStaged: MessageSpeechAttachment | null) { - let staged = initialStaged; + function makeFakeAgentVoiceReply( + initialStaged: MessageSpeechAttachment | null, + stagedTurnId: TurnId | null = null, + ) { + let staged: AgentVoiceReply.StagedAgentVoiceReply | null = + initialStaged === null ? null : { turnId: stagedTurnId, attachment: initialStaged }; const removedAudio: string[] = []; const shape: AgentVoiceReply.AgentVoiceReplyShape = { available: true, stage: () => Effect.die(new Error("stage is not exercised by ingestion tests")), - takeStaged: () => + peekStaged: () => Effect.sync(() => staged ?? undefined), + clearStaged: () => Effect.sync(() => { - const current = staged ?? undefined; staged = null; - return current; }), discardStaged: () => Effect.sync(() => { - if (staged !== null) removedAudio.push(staged.speechId); + if (staged !== null) removedAudio.push(staged.attachment.speechId); staged = null; }), - removeStagedAudio: (entry) => - Effect.sync(() => { - removedAudio.push(entry.speechId); - }), }; return { shape, removedAudio, hasStaged: () => staged !== null }; } it("attaches a staged agent voice reply to the turn's final assistant message", async () => { const staged = makeStagedVoiceReply(); - const fake = makeFakeAgentVoiceReply(staged); + const fake = makeFakeAgentVoiceReply(staged, asTurnId("turn-voice")); const harness = await createHarness({ agentVoiceReply: fake.shape }); const now = "2026-01-01T00:00:00.000Z"; @@ -1469,6 +1468,45 @@ describe("ProviderRuntimeIngestion", () => { expect(message?.speech?.speechId).toBe("thread-1-agent-voice-only"); }); + it("leaves a staged voice reply alone when a different turn completes", async () => { + const staged = makeStagedVoiceReply({ speechId: "thread-1-agent-voice-other-turn" }); + const fake = makeFakeAgentVoiceReply(staged, asTurnId("turn-actually-mine")); + const harness = await createHarness({ agentVoiceReply: fake.shape }); + const now = "2026-01-01T00:00:00.000Z"; + + harness.emit({ + type: "content.delta", + eventId: asEventId("evt-other-turn-delta"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-someone-else"), + itemId: asItemId("item-other-turn"), + payload: { streamKind: "assistant_text", delta: "unrelated reply" }, + }); + harness.emit({ + type: "turn.completed", + eventId: asEventId("evt-other-turn-completed"), + provider: ProviderDriverKind.make("codex"), + createdAt: now, + threadId: asThreadId("thread-1"), + turnId: asTurnId("turn-someone-else"), + status: "completed", + }); + + const thread = await waitForThread(harness.readModel, (entry) => + entry.messages.some( + (message: ProviderRuntimeTestMessage) => + message.id === "assistant:item-other-turn" && !message.streaming, + ), + ); + expect( + thread.messages.every((message: ProviderRuntimeTestMessage) => message.speech === undefined), + ).toBe(true); + expect(fake.hasStaged()).toBe(true); + expect(fake.removedAudio).toEqual([]); + }); + it("drops a staged voice reply when the turn does not complete normally", async () => { const staged = makeStagedVoiceReply({ speechId: "thread-1-agent-voice-dropped" }); const fake = makeFakeAgentVoiceReply(staged); diff --git a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts index 74077136a867..17e57d402e0d 100644 --- a/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts +++ b/apps/server/src/orchestration/Layers/ProviderRuntimeIngestion.ts @@ -2083,12 +2083,20 @@ const make = Effect.gen(function* () { }); // A recording staged through the voice_reply MCP tool attaches to - // the turn's last assistant message once the turn completes - // normally; any other outcome throws the recording away, since an - // interrupted or failed turn no longer matches its script. - const stagedVoiceReply = yield* agentVoiceReply.takeStaged(thread.id); - if (stagedVoiceReply) { + // the turn's last assistant message once ITS turn completes + // normally. A completion for some other turn (a stale event, or a + // steered/superseded turn) leaves the entry alone, and a failed or + // interrupted outcome throws the recording away, since it no longer + // matches what actually happened. The entry is cleared only after + // the attach commands land, so a failed dispatch does not silently + // lose the recording. + const stagedVoiceReply = yield* agentVoiceReply.peekStaged(thread.id); + if ( + stagedVoiceReply && + (stagedVoiceReply.turnId === null || sameId(stagedVoiceReply.turnId, turnId)) + ) { if (normalizeRuntimeTurnState(event.payload.state) === "completed") { + const speech = stagedVoiceReply.attachment; const finalizedMessageIds = Array.from(assistantMessageIds); const lastFinalizedMessageId = finalizedMessageIds[finalizedMessageIds.length - 1]; const lastProjectedMessageId = findLastAssistantMessageForTurn(messages, turnId)?.id; @@ -2100,7 +2108,7 @@ const make = Effect.gen(function* () { threadId: thread.id, messageId: targetMessageId, turnId, - speech: stagedVoiceReply, + speech, createdAt: now, }); } else { @@ -2112,7 +2120,7 @@ const make = Effect.gen(function* () { commandId: yield* providerCommandId(event, "agent-voice-reply-message"), threadId: thread.id, messageId: voiceMessageId, - delta: stagedVoiceReply.transcript, + delta: speech.transcript, turnId, createdAt: now, }); @@ -2122,17 +2130,35 @@ const make = Effect.gen(function* () { threadId: thread.id, messageId: voiceMessageId, turnId, - speech: stagedVoiceReply, + speech, createdAt: now, }); } + yield* agentVoiceReply.clearStaged(thread.id); } else { - yield* agentVoiceReply.removeStagedAudio(stagedVoiceReply); + yield* agentVoiceReply.discardStaged(thread.id); } } } else { - // An untargeted completion cannot prove which turn it ends, so a - // staged recording has nothing safe to bind to. + // An untargeted completion cannot prove which turn it ends; only a + // wildcard entry (staged with no observable turn) is dropped. + const stagedVoiceReply = yield* agentVoiceReply.peekStaged(thread.id); + if (stagedVoiceReply && stagedVoiceReply.turnId === null) { + yield* agentVoiceReply.discardStaged(thread.id); + } + } + } + + if (event.type === "turn.aborted") { + // An aborted turn's recording no longer matches what happened; drop + // it when it belongs to the aborted turn (or cannot be attributed). + const stagedVoiceReply = yield* agentVoiceReply.peekStaged(thread.id); + if ( + stagedVoiceReply && + (stagedVoiceReply.turnId === null || + eventTurnId === undefined || + sameId(stagedVoiceReply.turnId, eventTurnId)) + ) { yield* agentVoiceReply.discardStaged(thread.id); } } diff --git a/apps/server/src/voice/AgentVoiceReply.ts b/apps/server/src/voice/AgentVoiceReply.ts index d81d2d8a09e4..aff8a232fddc 100644 --- a/apps/server/src/voice/AgentVoiceReply.ts +++ b/apps/server/src/voice/AgentVoiceReply.ts @@ -6,6 +6,7 @@ import { AgentVoiceReplyError, type MessageSpeechAttachment, type ThreadId, + type TurnId, } from "@t3tools/contracts"; import * as Config from "effect/Config"; import * as Context from "effect/Context"; @@ -17,6 +18,7 @@ import * as Option from "effect/Option"; import * as Redacted from "effect/Redacted"; import * as SynchronizedRef from "effect/SynchronizedRef"; import { HttpClient } from "effect/unstable/http"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; import { createAttachmentId } from "../attachmentStore.ts"; import { resolveAttachmentRelativePath } from "../attachmentPaths.ts"; @@ -30,15 +32,28 @@ import { resolveMessageSpeechVoiceSetting, } from "./MessageSpeech.ts"; +/** + * A recording staged by the voice_reply MCP tool, bound to the turn that was + * active when it was staged. `turnId` is null only when the thread had no + * observable active turn at stage time; ingestion then treats it as "attach + * to the next completed turn". + */ +export interface StagedAgentVoiceReply { + readonly turnId: TurnId | null; + readonly attachment: MessageSpeechAttachment; +} + /** * Agent-staged voice replies. The `voice_reply` MCP tool synthesizes a * recording mid-turn and parks it here; provider-runtime ingestion collects it - * when the turn completes and attaches it to the turn's final assistant + * when its turn completes and attaches it to that turn's final assistant * message. One staged reply per thread — a second call replaces the first. * * The MP3 is written to the attachments directory at stage time so the later * attach command can stay metadata-only, mirroring how user image attachments - * are persisted by the normalizer before their event is recorded. + * are persisted by the normalizer before their event is recorded. The entry + * itself is removed only after the attach commands land (`clearStaged`), so a + * failed dispatch does not silently drop the recording from memory. */ export interface AgentVoiceReplyShape { readonly available: boolean; @@ -46,12 +61,12 @@ export interface AgentVoiceReplyShape { readonly threadId: ThreadId; readonly script: string; }) => Effect.Effect; - /** Removes and returns the staged reply without touching its audio file. */ - readonly takeStaged: (threadId: ThreadId) => Effect.Effect; - /** Removes the staged reply and deletes its audio file, if any. */ + /** Reads the staged reply without removing it. */ + readonly peekStaged: (threadId: ThreadId) => Effect.Effect; + /** Removes the staged reply, keeping its audio file (it has been attached). */ + readonly clearStaged: (threadId: ThreadId) => Effect.Effect; + /** Removes the staged reply and deletes its audio file. */ readonly discardStaged: (threadId: ThreadId) => Effect.Effect; - /** Deletes the audio file behind a reply that will never be attached. */ - readonly removeStagedAudio: (staged: MessageSpeechAttachment) => Effect.Effect; } export class AgentVoiceReply extends Context.Service()( @@ -62,9 +77,9 @@ export class AgentVoiceReply extends Context.Service Effect.fail(new AgentVoiceReplyError({ reason: "unavailable" })), - takeStaged: () => Effect.succeed(undefined), + peekStaged: () => Effect.succeed(undefined), + clearStaged: () => Effect.void, discardStaged: () => Effect.void, - removeStagedAudio: () => Effect.void, }); export const layer = Layer.effect( @@ -80,9 +95,10 @@ export const layer = Layer.effect( const available = Option.isSome(apiKey) && Redacted.value(apiKey.value).trim().length > 0; const httpClient = yield* HttpClient.HttpClient; const fileSystem = yield* FileSystem.FileSystem; + const sql = yield* SqlClient.SqlClient; const serverConfig = yield* ServerConfig.ServerConfig; const serverSettings = yield* ServerSettingsService; - const staged = yield* SynchronizedRef.make>( + const staged = yield* SynchronizedRef.make>( new Map(), ); @@ -97,6 +113,32 @@ export const layer = Layer.effect( return path ? fileSystem.remove(path, { force: true }).pipe(Effect.ignore) : Effect.void; }; + /** + * Best-effort read of the thread's active turn so the staged reply can be + * bound to it. A missing or unreadable session degrades to null rather + * than failing the stage: the recording then attaches to the thread's + * next completed turn. + */ + const resolveActiveTurnId = (threadId: ThreadId) => + sql<{ readonly activeTurnId: string | null }>` + SELECT active_turn_id AS "activeTurnId" + FROM projection_thread_sessions + WHERE thread_id = ${threadId} + LIMIT 1 + `.pipe( + Effect.map((rows) => (rows[0]?.activeTurnId ?? null) as TurnId | null), + Effect.orElseSucceed((): TurnId | null => null), + ); + + const takeEntry = (threadId: ThreadId) => + SynchronizedRef.modify(staged, (entries) => { + const current = entries.get(threadId); + if (!current) return [undefined, entries] as const; + const next = new Map(entries); + next.delete(threadId); + return [current, next] as const; + }); + const stage: AgentVoiceReplyShape["stage"] = Effect.fn("AgentVoiceReply.stage")( function* (input) { if (!available || Option.isNone(apiKey)) { @@ -126,6 +168,7 @@ export const layer = Layer.effect( return yield* new AgentVoiceReplyError({ reason: "script_too_long" }); } + const turnId = yield* resolveActiveTurnId(input.threadId); const audioBytes = yield* synthesizeElevenLabsSpeech({ httpClient, apiKey: apiKey.value, @@ -162,34 +205,28 @@ export const layer = Layer.effect( const replaced = yield* SynchronizedRef.modify(staged, (entries) => { const previous = entries.get(input.threadId); const next = new Map(entries); - next.set(input.threadId, attachment); + next.set(input.threadId, { turnId, attachment }); return [previous, next] as const; }); if (replaced) { - yield* removeAudioFile(replaced.speechId); + yield* removeAudioFile(replaced.attachment.speechId); } return attachment; }, ); - const takeStaged: AgentVoiceReplyShape["takeStaged"] = (threadId) => - SynchronizedRef.modify(staged, (entries) => { - const current = entries.get(threadId); - if (!current) return [undefined, entries] as const; - const next = new Map(entries); - next.delete(threadId); - return [current, next] as const; - }); - return AgentVoiceReply.of({ available, stage, - takeStaged, + peekStaged: (threadId) => + SynchronizedRef.get(staged).pipe(Effect.map((entries) => entries.get(threadId))), + clearStaged: (threadId) => takeEntry(threadId).pipe(Effect.asVoid), discardStaged: (threadId) => - takeStaged(threadId).pipe( - Effect.flatMap((entry) => (entry ? removeAudioFile(entry.speechId) : Effect.void)), + takeEntry(threadId).pipe( + Effect.flatMap((entry) => + entry ? removeAudioFile(entry.attachment.speechId) : Effect.void, + ), ), - removeStagedAudio: (entry) => removeAudioFile(entry.speechId), }); }), ); diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 6056d2c47ea1..97c725816734 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -1384,7 +1384,7 @@ function AssistantTimelineRow({ row }: { row: Extract void prepareSpeech()} + onRetry={null} primary /> ) : null} @@ -1532,7 +1532,12 @@ function AssistantSpeechPlayer({ }: { environmentId: EnvironmentId; speech: MessageSpeechSynthesisResult; - onRetry: () => void; + /** + * null hides the regenerate action. Agent recordings must not offer it: + * regenerating would synthesize the written text as a user listening + * version and destroy the agent's recording. + */ + onRetry: (() => void) | null; /** Agent voice replies render the player as the message's main content. */ primary?: boolean; }) { @@ -1572,10 +1577,16 @@ function AssistantSpeechPlayer({
{audioUrlState._tag === "Failure" ? (
-

The audio file is unavailable. Regenerate it to listen again.

- +

+ {onRetry === null + ? "The audio file is unavailable." + : "The audio file is unavailable. Regenerate it to listen again."} +

+ {onRetry !== null ? ( + + ) : null}
) : audioUrlState._tag === "Loading" ? (
From dd5a3c8af9a44cbad7df499df70e46bb8c58d237 Mon Sep 17 00:00:00 2001 From: pandec Date: Mon, 24 Aug 2026 16:39:38 +0200 Subject: [PATCH 3/7] fix(web): name the toggle after the voice reply it collapses --- apps/web/src/components/chat/MessagesTimeline.tsx | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 97c725816734..7a470cb7b0bd 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -1458,7 +1458,11 @@ function AssistantTimelineRow({ row }: { row: Extract ) : null} From e0e168aba500d25200b54d8c367eda4ef4a8e01c Mon Sep 17 00:00:00 2001 From: pandec Date: Mon, 24 Aug 2026 16:58:27 +0200 Subject: [PATCH 4/7] fix(clients): harden voice-reply presentation after review Written replies now surface when the recording's audio file is missing, voice-only turns no longer offer a written reply that duplicates the transcript, mobile no longer downloads every visible recording on mount (the player fetches on first play), empty assistant messages that carry only a recording render on mobile instead of disappearing, VoiceOver labels say voice reply, and the Settings toggle is searchable. --- .../src/features/threads/ThreadFeed.tsx | 101 ++++++++++++------ .../src/components/chat/MessagesTimeline.tsx | 32 +++++- .../settings/ExtrasSettingsPanel.tsx | 1 + .../src/components/settings/settingsSearch.ts | 5 + 4 files changed, 103 insertions(+), 36 deletions(-) diff --git a/apps/mobile/src/features/threads/ThreadFeed.tsx b/apps/mobile/src/features/threads/ThreadFeed.tsx index 85129f3e6e6a..99e59813b766 100644 --- a/apps/mobile/src/features/threads/ThreadFeed.tsx +++ b/apps/mobile/src/features/threads/ThreadFeed.tsx @@ -1231,14 +1231,15 @@ function renderFeedEntry( ); } - // Skip empty assistant messages (no text, no attachments) — they would - // render as an orphaned timestamp and break adjacent activity-group merging. - if (message.text.trim().length === 0 && attachments.length === 0) { + const agentVoiceReply = message.speech?.origin === "agent" ? message.speech : null; + // Skip empty assistant messages (no text, no attachments, no voice + // reply) — they would render as an orphaned timestamp and break adjacent + // activity-group merging. + if (message.text.trim().length === 0 && attachments.length === 0 && agentVoiceReply === null) { return null; } const enterAnimated = isFreshTimestamp(message.createdAt); - const agentVoiceReply = message.speech?.origin === "agent" ? message.speech : null; const writtenReply = message.text.trim().length > 0 ? ( hasNativeSelectableMarkdownText() ? ( @@ -1271,7 +1272,9 @@ function renderFeedEntry( speech={agentVoiceReply} iconSubtleColor={iconSubtleColor} > - {writtenReply} + {/* A voice-only turn's text is the transcript itself; the + player's "View transcript" already covers it. */} + {message.text.trim() === agentVoiceReply.transcript.trim() ? null : writtenReply} ) : ( writtenReply @@ -1337,6 +1340,10 @@ function AssistantAgentVoiceReply(props: { }) { const [transcriptExpanded, setTranscriptExpanded] = useState(false); const [writtenReplyExpanded, setWrittenReplyExpanded] = useState(false); + // When the recording's file is gone, the written reply becomes the message + // content and is forced visible instead of hiding behind the toggle. + const [audioUnavailable, setAudioUnavailable] = useState(false); + const onAudioUnavailable = useCallback(() => setAudioUnavailable(true), []); const hasWrittenReply = props.children !== null; return ( @@ -1348,9 +1355,10 @@ function AssistantAgentVoiceReply(props: { transcriptExpanded={transcriptExpanded} onToggleTranscript={() => setTranscriptExpanded((current) => !current)} onRetry={null} + onAudioUnavailable={onAudioUnavailable} primary /> - {hasWrittenReply ? ( + {hasWrittenReply && !audioUnavailable ? ( ) : null} - {writtenReplyExpanded ? props.children : null} + {hasWrittenReply && (writtenReplyExpanded || audioUnavailable) ? props.children : null} ); } @@ -1602,6 +1610,8 @@ function AssistantSpeechPlayer(props: { readonly onToggleTranscript: () => void; /** null hides the regenerate action (agent recordings cannot be re-made client-side). */ readonly onRetry: (() => void) | null; + /** Lets the row fall back to the written reply when the audio is gone. */ + readonly onAudioUnavailable?: () => void; /** Agent voice replies render the player as the message's main content. */ readonly primary?: boolean; }) { @@ -1617,8 +1627,19 @@ function AssistantSpeechPlayer(props: { attachmentId: props.speech.speechId, }); const audioUrl = audioUrlState._tag === "Success" ? audioUrlState.url : null; - const player = useAudioPlayer(audioUrl, { updateInterval: 250 }); + // Primary players mount for every voice-reply row in the feed, so they must + // not fetch their MP3 until the user asks to play — a thread can hold many + // recordings and the app may be on a remote or cellular link. Secondary + // players only mount after an explicit expand, which is consent enough. + const [activated, setActivated] = useState(props.primary !== true); + const [pendingPlay, setPendingPlay] = useState(false); + const player = useAudioPlayer(activated ? audioUrl : null, { updateInterval: 250 }); const status = useAudioPlayerStatus(player); + const audioUnavailable = audioUrlState._tag === "Failure"; + const onAudioUnavailableProp = props.onAudioUnavailable; + useEffect(() => { + if (audioUnavailable) onAudioUnavailableProp?.(); + }, [audioUnavailable, onAudioUnavailableProp]); const progress = status.duration > 0 ? Math.min(1, status.currentTime / status.duration) : 0; const pausePlayer = useCallback(() => { @@ -1650,32 +1671,48 @@ function AssistantSpeechPlayer(props: { [pausePlayer, props.speech.speechId], ); + const startPlayback = useCallback( + () => + startListeningPlayback({ + id: props.speech.speechId, + pause: pausePlayer, + restartFromBeginning: status.duration > 0 && status.currentTime >= status.duration - 0.1, + seekToBeginning: () => player.seekTo(0), + prepareAudioMode: () => + setAudioModeAsync({ allowsRecording: false, playsInSilentMode: true }), + applyPlaybackRate, + play: () => player.play(), + }), + [ + applyPlaybackRate, + pausePlayer, + player, + props.speech.speechId, + status.currentTime, + status.duration, + ], + ); + const onTogglePlayback = useCallback(async () => { if (status.playing) { pausePlayer(); return; } if (blocked) return; - await startListeningPlayback({ - id: props.speech.speechId, - pause: pausePlayer, - restartFromBeginning: status.duration > 0 && status.currentTime >= status.duration - 0.1, - seekToBeginning: () => player.seekTo(0), - prepareAudioMode: () => - setAudioModeAsync({ allowsRecording: false, playsInSilentMode: true }), - applyPlaybackRate, - play: () => player.play(), - }); - }, [ - applyPlaybackRate, - blocked, - pausePlayer, - player, - props.speech.speechId, - status.currentTime, - status.duration, - status.playing, - ]); + if (!activated) { + setActivated(true); + setPendingPlay(true); + return; + } + await startPlayback(); + }, [activated, blocked, pausePlayer, startPlayback, status.playing]); + + // First tap on a deferred player: wait for the source to attach, then play. + useEffect(() => { + if (!pendingPlay || !activated || audioUrl === null) return; + setPendingPlay(false); + void startPlayback(); + }, [activated, audioUrl, pendingPlay, startPlayback]); return ( ) : null} - ) : audioUrl === null ? ( + ) : audioUrl === null && activated ? ( Loading audio… @@ -1720,10 +1757,10 @@ function AssistantSpeechPlayer(props: { accessibilityRole="button" accessibilityLabel={ blocked - ? "Play listening version unavailable while recording" + ? `Play ${props.primary ? "voice reply" : "listening version"} unavailable while recording` : status.playing - ? "Pause listening version" - : "Play listening version" + ? `Pause ${props.primary ? "voice reply" : "listening version"}` + : `Play ${props.primary ? "voice reply" : "listening version"}` } accessibilityState={{ disabled: blocked }} className={cn( diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index 7a470cb7b0bd..5fe9847e3ee5 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -1292,9 +1292,22 @@ function AssistantTimelineRow({ row }: { row: Extract setVoiceReplyAudioUnavailable(true), []); + // A voice-only turn publishes its transcript as the message text; a + // "written reply" that merely duplicates the transcript is not offered. + const hasDistinctWrittenReply = + isAgentVoiceReply && + speech !== null && + row.message.text.trim().length > 0 && + row.message.text.trim() !== speech.transcript.trim(); + // Never leave the row content-less: when the player is hidden or its audio + // cannot load, the written reply always shows. + const showMessageText = + !isAgentVoiceReply || + !speechExpanded || + voiceReplyAudioUnavailable || + (writtenReplyExpanded && hasDistinctWrittenReply); const canShowSpeech = speech !== null || @@ -1385,10 +1398,14 @@ function AssistantTimelineRow({ row }: { row: Extract ) : null} - {isAgentVoiceReply && speechExpanded ? ( + {isAgentVoiceReply && + speechExpanded && + hasDistinctWrittenReply && + !voiceReplyAudioUnavailable ? (