diff --git a/apps/x/apps/renderer/src/App.tsx b/apps/x/apps/renderer/src/App.tsx index 6b820f70..5ef631e2 100644 --- a/apps/x/apps/renderer/src/App.tsx +++ b/apps/x/apps/renderer/src/App.tsx @@ -980,6 +980,28 @@ function App() { const ttsRef = useRef(tts) ttsRef.current = tts + // Speak newly completed blocks from the new runtime's live stream + // (parity with the legacy text-delta voice extraction below). The store + // accumulates completed blocks in chatState.voiceSegments; we speak only + // segments that appeared after the current session became active. + const spokenVoiceRef = useRef<{ key: string | null; count: number }>({ key: null, count: 0 }) + const voiceSegments = sessionChat.chatState?.voiceSegments + useEffect(() => { + if (!voiceSegments) return + if (spokenVoiceRef.current.key !== runId) { + // Session switch: skip anything already streamed before we arrived. + spokenVoiceRef.current = { key: runId, count: voiceSegments.length } + return + } + while (spokenVoiceRef.current.count < voiceSegments.length) { + const segment = voiceSegments[spokenVoiceRef.current.count] + spokenVoiceRef.current.count += 1 + if (ttsEnabledRef.current) { + ttsRef.current.speak(segment) + } + } + }, [voiceSegments, runId]) + const voice = useVoiceMode() const voiceRef = useRef(voice) voiceRef.current = voice diff --git a/apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts b/apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts index 937ad9a2..66a781ce 100644 --- a/apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts +++ b/apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts @@ -82,6 +82,49 @@ describe('applyOverlay', () => { }) }) +describe('voice output', () => { + const delta = (d: string): Parameters[1] => + ({ type: 'text_delta', turnId: T1, modelCallIndex: 0, delta: d }) + + it('extracts completed blocks across split deltas', () => { + let overlay = emptyOverlay() + overlay = applyOverlay(overlay, delta('Sure! hello there and bye')) + expect(overlay.voiceSegments).toEqual(['hello there']) + overlay = applyOverlay(overlay, delta(' done')) + expect(overlay.voiceSegments).toEqual(['hello there', 'bye']) + }) + + it('keeps segments but resets the scan on model_call_completed', () => { + let overlay = emptyOverlay() + overlay = applyOverlay(overlay, delta('one')) + overlay = applyOverlay(overlay, completed(T1, 0, assistantText('one'))) + expect(overlay.voiceSegments).toEqual(['one']) + expect(overlay.text).toBe('') + overlay = applyOverlay(overlay, delta('two')) + expect(overlay.voiceSegments).toEqual(['one', 'two']) + }) + + it('strips voice tags from the streaming message and exposes segments on state', () => { + const turn = reduceTurn([created(T1, S1), requested(T1, 0)]) + let overlay = emptyOverlay() + overlay = applyOverlay(overlay, delta('Plan: speak this rest')) + const state = buildSessionChatState([turn], overlay) + expect(state.currentAssistantMessage).toBe('Plan: speak this rest') + expect(state.voiceSegments).toEqual(['speak this']) + }) + + it('strips voice tags from persisted assistant messages', () => { + const state = reduceTurn( + completedTurnLog(T1, S1, 'q', 'Sure. Here you go. Done.'), + ) + const items = buildTurnConversation(state) + const assistant = items.find((i) => isChatMessage(i) && i.role === 'assistant') + expect(isChatMessage(assistant!) && assistant.content).toBe('Sure. Here you go. Done.') + }) +}) + describe('buildTurnConversation', () => { it('maps user input, assistant text, and settled tool calls', () => { const call = assistantCalls(toolCallPart('tc1', 'echo', { x: 1 })) diff --git a/apps/x/apps/renderer/src/lib/session-chat/turn-view.ts b/apps/x/apps/renderer/src/lib/session-chat/turn-view.ts index fad07998..e4ca3cdd 100644 --- a/apps/x/apps/renderer/src/lib/session-chat/turn-view.ts +++ b/apps/x/apps/renderer/src/lib/session-chat/turn-view.ts @@ -34,20 +34,62 @@ export type LiveOverlay = { text: string reasoning: string toolOutput: Record + // Contents of completed … blocks seen while streaming, in + // order, monotonically growing for the lifetime of the overlay (i.e. one + // active turn). Consumers speak segments beyond what they've already + // spoken; the overlay reset on turn switch starts a fresh list. + voiceSegments: string[] + // Scan cursor into `text` — everything before it has been checked for + // complete voice blocks. + voiceScanIndex: number } -export const emptyOverlay = (): LiveOverlay => ({ text: '', reasoning: '', toolOutput: {} }) +export const emptyOverlay = (): LiveOverlay => ({ + text: '', + reasoning: '', + toolOutput: {}, + voiceSegments: [], + voiceScanIndex: 0, +}) + +// The model emits … around speakable text when voice output +// is enabled; tags are never shown to the user. +export function stripVoiceTags(text: string): string { + return text.replace(/<\/?voice>/g, '') +} + +const VOICE_BLOCK = /([\s\S]*?)<\/voice>/g // Accumulates deltas; canonical durable events supersede the buffers (the // committed transcript now contains what was streaming). export function applyOverlay(overlay: LiveOverlay, event: TurnStreamEvent): LiveOverlay { switch (event.type) { - case 'text_delta': - return { ...overlay, text: overlay.text + event.delta } + case 'text_delta': { + const text = overlay.text + event.delta + // Extract complete voice blocks past the scan cursor. Incomplete + // blocks (opening tag seen, closing not yet) stay unconsumed until a + // later delta completes them. + const segments: string[] = [] + let scanIndex = overlay.voiceScanIndex + VOICE_BLOCK.lastIndex = scanIndex + for (let m = VOICE_BLOCK.exec(text); m; m = VOICE_BLOCK.exec(text)) { + const content = m[1].trim() + if (content) segments.push(content) + scanIndex = m.index + m[0].length + } + return { + ...overlay, + text, + ...(segments.length > 0 + ? { voiceSegments: [...overlay.voiceSegments, ...segments] } + : {}), + voiceScanIndex: scanIndex, + } + } case 'reasoning_delta': return { ...overlay, reasoning: overlay.reasoning + event.delta } case 'model_call_completed': - return { ...overlay, text: '', reasoning: '' } + return { ...overlay, text: '', reasoning: '', voiceScanIndex: 0 } case 'tool_progress': { const progress = event.progress if ( @@ -140,13 +182,16 @@ export function buildTurnConversation(state: TurnState): ConversationItem[] { for (const call of state.modelCalls) { if (call.response === undefined) continue const content = call.response.content - const text = + // Voice tags are model-facing markup, never shown (parity with the + // legacy path's display-time strip). + const text = stripVoiceTags( typeof content === 'string' ? content : content .map((part) => (part.type === 'text' ? part.text : '')) .filter(Boolean) - .join('\n') + .join('\n'), + ) if (text) { items.push({ id: `${turnId}:a${call.index}`, @@ -192,6 +237,8 @@ type PermMeta = z.infer export type SessionChatState = { conversation: ConversationItem[] currentAssistantMessage: string + // See LiveOverlay.voiceSegments. + voiceSegments: string[] pendingAskHumanRequests: Map> allPermissionRequests: Map> permissionResponses: Map @@ -289,7 +336,8 @@ export function buildSessionChatState( const settled = status === 'completed' || status === 'failed' || status === 'cancelled' return { conversation, - currentAssistantMessage: overlay.text, + currentAssistantMessage: stripVoiceTags(overlay.text), + voiceSegments: overlay.voiceSegments, pendingAskHumanRequests, allPermissionRequests, permissionResponses, diff --git a/apps/x/packages/core/src/di/container.ts b/apps/x/packages/core/src/di/container.ts index 34388754..e9be949f 100644 --- a/apps/x/packages/core/src/di/container.ts +++ b/apps/x/packages/core/src/di/container.ts @@ -31,6 +31,8 @@ import { FSTurnRepo } from "../turns/fs-repo.js"; import type { ITurnRepo } from "../turns/repo.js"; import { TurnRepoContextResolver, type IContextResolver } from "../turns/context-resolver.js"; import { EmitterTurnLifecycleBus, type ITurnLifecycleBus } from "../turns/bus.js"; +import { RealUsageReporter } from "../turns/bridges/real-usage-reporter.js"; +import type { IUsageReporter } from "../turns/usage-reporter.js"; import { TurnRuntime } from "../turns/runtime.js"; import type { ITurnRuntime } from "../turns/api.js"; import type { IAgentResolver } from "../turns/agent-resolver.js"; @@ -103,6 +105,7 @@ container.register({ turnRepo: asClass(FSTurnRepo).singleton(), contextResolver: asClass(TurnRepoContextResolver).singleton(), lifecycleBus: asClass(EmitterTurnLifecycleBus).singleton(), + usageReporter: asClass(RealUsageReporter).singleton(), agentResolver: asFunction(() => new RealAgentResolver()).singleton(), modelRegistry: asFunction(() => new RealModelRegistry()).singleton(), toolRegistry: asFunction(() => new RealToolRegistry()).singleton(), diff --git a/apps/x/packages/core/src/turns/bridges/real-usage-reporter.ts b/apps/x/packages/core/src/turns/bridges/real-usage-reporter.ts new file mode 100644 index 00000000..99d004e5 --- /dev/null +++ b/apps/x/packages/core/src/turns/bridges/real-usage-reporter.ts @@ -0,0 +1,22 @@ +import { captureLlmUsage } from "../../analytics/usage.js"; +import { getCurrentUseCase } from "../../analytics/use_case.js"; +import type { IUsageReporter, ModelUsageReport } from "../usage-reporter.js"; + +// Reports each completed model call as the same `llm_usage` PostHog event +// the old run loop emitted. The use case comes from the AsyncLocalStorage +// context the caller established (headless runners wrap startHeadlessAgent +// in withUseCase); UI-driven session turns have no context and default to +// copilot_chat — matching the old createRun default. +export class RealUsageReporter implements IUsageReporter { + reportModelUsage(report: ModelUsageReport): void { + const context = getCurrentUseCase(); + captureLlmUsage({ + useCase: context?.useCase ?? "copilot_chat", + ...(context?.subUseCase ? { subUseCase: context.subUseCase } : {}), + agentName: report.agentId, + model: report.model.model, + provider: report.model.provider, + usage: report.usage, + }); + } +} diff --git a/apps/x/packages/core/src/turns/index.ts b/apps/x/packages/core/src/turns/index.ts index 1ec99055..77153361 100644 --- a/apps/x/packages/core/src/turns/index.ts +++ b/apps/x/packages/core/src/turns/index.ts @@ -13,3 +13,4 @@ export * from "./repo.js"; export * from "./runtime.js"; export * from "./stream.js"; export * from "./tool-registry.js"; +export * from "./usage-reporter.js"; diff --git a/apps/x/packages/core/src/turns/runtime.test.ts b/apps/x/packages/core/src/turns/runtime.test.ts index 170ba61b..335250e4 100644 --- a/apps/x/packages/core/src/turns/runtime.test.ts +++ b/apps/x/packages/core/src/turns/runtime.test.ts @@ -308,6 +308,7 @@ function makeRuntime(opts: { const checker = opts.checker ?? new FakePermissionChecker(); const classifier = opts.classifier ?? new FakePermissionClassifier(); const bus = new FakeBus(); + const usage = new FakeUsageReporter(); const runtime = new TurnRuntime({ turnRepo: repo, idGenerator: new FakeIdGen(opts.idStart ?? 0), @@ -319,8 +320,9 @@ function makeRuntime(opts: { permissionChecker: checker, permissionClassifier: classifier, lifecycleBus: bus, + usageReporter: usage, }); - return { runtime, repo, models, checker, classifier, bus }; + return { runtime, repo, models, checker, classifier, bus, usage }; } async function newTurn( @@ -371,6 +373,21 @@ async function advanceAndSettle( return settle(runtime.advanceTurn(turnId, input, options)); } +class FakeUsageReporter { + reports: Array<{ + agentId: string; + model: { provider: string; model: string }; + usage: Record; + }> = []; + reportModelUsage(report: { + agentId: string; + model: { provider: string; model: string }; + usage: Record; + }): void { + this.reports.push(report); + } +} + function typesOf(events: Array<{ type: string }>): string[] { return events.map((e) => e.type); } @@ -397,7 +414,7 @@ async function persisted( describe("plain model response (26.1)", () => { it("runs one model step to completion with exact persisted request", async () => { - const { runtime, repo, models, bus } = makeRuntime({ + const { runtime, repo, models, bus, usage } = makeRuntime({ models: [ respond( { type: "text_delta", delta: "do" }, @@ -433,6 +450,14 @@ describe("plain model response (26.1)", () => { // snapshot tools, encoded messages. expect(models.requests[0].systemPrompt).toBe("SYS"); expect(sentMessages(models.requests[0])).toEqual([user("hello")]); + // One usage report per completed model call, after the durable append. + expect(usage.reports).toEqual([ + { + agentId: "copilot", + model: defaultAgent.model, + usage: { inputTokens: 10, outputTokens: 5, totalTokens: 15 }, + }, + ]); // Deltas are streamed but never persisted. expect(events.filter((e) => e.type === "text_delta")).toHaveLength(2); expect(typesOf(log)).not.toContain("text_delta"); diff --git a/apps/x/packages/core/src/turns/runtime.ts b/apps/x/packages/core/src/turns/runtime.ts index 80f657b9..da2362a4 100644 --- a/apps/x/packages/core/src/turns/runtime.ts +++ b/apps/x/packages/core/src/turns/runtime.ts @@ -41,6 +41,7 @@ import { import type { ITurnLifecycleBus } from "./bus.js"; import type { IClock } from "./clock.js"; import { composeModelRequest } from "./compose-model-request.js"; +import type { IUsageReporter } from "./usage-reporter.js"; import type { IContextResolver } from "./context-resolver.js"; import type { IModelRegistry, @@ -68,6 +69,7 @@ export interface TurnRuntimeDependencies { permissionChecker: IPermissionChecker; permissionClassifier: IPermissionClassifier; lifecycleBus: ITurnLifecycleBus; + usageReporter: IUsageReporter; } // Immutable dependency container: holds no mutable per-turn state. All active @@ -83,6 +85,7 @@ export class TurnRuntime implements ITurnRuntime { private readonly permissionChecker: IPermissionChecker; private readonly permissionClassifier: IPermissionClassifier; private readonly lifecycleBus: ITurnLifecycleBus; + private readonly usageReporter: IUsageReporter; constructor({ turnRepo, @@ -95,6 +98,7 @@ export class TurnRuntime implements ITurnRuntime { permissionChecker, permissionClassifier, lifecycleBus, + usageReporter, }: TurnRuntimeDependencies) { this.turnRepo = turnRepo; this.idGenerator = idGenerator; @@ -106,6 +110,7 @@ export class TurnRuntime implements ITurnRuntime { this.permissionChecker = permissionChecker; this.permissionClassifier = permissionClassifier; this.lifecycleBus = lifecycleBus; + this.usageReporter = usageReporter; } async createTurn(input: CreateTurnInput): Promise { @@ -257,6 +262,7 @@ export class TurnRuntime implements ITurnRuntime { resolvedContext, resolvedAgent, model, + usageReporter: this.usageReporter, toolsByName, signal: controller.signal, turnRepo: this.turnRepo, @@ -285,6 +291,7 @@ class TurnAdvance { private readonly resolvedContext: Array>; private readonly resolvedAgent: z.infer; private readonly model: ResolvedModel; + private readonly usageReporter: IUsageReporter; private readonly toolsByName: Map; private readonly signal: AbortSignal; private readonly turnRepo: ITurnRepo; @@ -306,6 +313,7 @@ class TurnAdvance { resolvedContext: Array>; resolvedAgent: z.infer; model: ResolvedModel; + usageReporter: IUsageReporter; toolsByName: Map; signal: AbortSignal; turnRepo: ITurnRepo; @@ -320,6 +328,7 @@ class TurnAdvance { this.resolvedContext = init.resolvedContext; this.resolvedAgent = init.resolvedAgent; this.model = init.model; + this.usageReporter = init.usageReporter; this.toolsByName = init.toolsByName; this.signal = init.signal; this.turnRepo = init.turnRepo; @@ -1015,6 +1024,17 @@ class TurnAdvance { ? {} : { providerMetadata: completion.providerMetadata }), }); + // Analytics after the durable barrier; a reporter failure must never + // affect the turn. + try { + this.usageReporter.reportModelUsage({ + agentId: this.resolvedAgent.agentId, + model: this.resolvedAgent.model, + usage: completion.usage, + }); + } catch { + // best effort + } return undefined; } diff --git a/apps/x/packages/core/src/turns/usage-reporter.ts b/apps/x/packages/core/src/turns/usage-reporter.ts new file mode 100644 index 00000000..90bbd284 --- /dev/null +++ b/apps/x/packages/core/src/turns/usage-reporter.ts @@ -0,0 +1,19 @@ +import type { z } from "zod"; +import type { ModelDescriptor, TurnUsage } from "@x/shared/dist/turns.js"; + +// Analytics seam for the turn loop: invoked once per completed model call, +// after the durable model_call_completed append. Implementations must never +// throw into the loop and must not block it (fire-and-forget). +export interface ModelUsageReport { + agentId: string; + model: z.infer; + usage: z.infer; +} + +export interface IUsageReporter { + reportModelUsage(report: ModelUsageReport): void; +} + +export class NoopUsageReporter implements IUsageReporter { + reportModelUsage(): void {} +}