feat(voice): a per-packet call timeline from the agent's voice.json

package.json

@@ -17,6 +17,7 @@
 		"moq/js/publish",
 		"moq/js/qmux",
 		"moq/js/signals",
+		"moq/js/ui",
 		"moq/js/video",
 		"moq/js/watch"
 	],
@@ -46,6 +47,7 @@
 		"@astrojs/rss": "^4.0.19",
 		"@astrojs/solid-js": "7.1.0",
 		"@biomejs/biome": "^2.5.15",
+		"@microsoft/api-extractor": "7.59.4",
 		"@tailwindcss/forms": "^0.5.11",
 		"@tailwindcss/typography": "^0.5.20",
 		"@tailwindcss/vite": "^4.3.3",

quest/m1/README.md

@@ -9,4 +9,3 @@ moves and buffers the call and why it beats WebRTC for voice agents.
 
 - [Drain the voice goodbye instead of waiting](/quest/m1/voice-goodbye-drain.md) - hanging up closes through the draining `close()`, not a fixed wait
 - [The voice agent publishes the full voice.json](/quest/m1/voice-timeline-track.md) - condition: the agent's side of the call, timed, on wtf
-- [A live timeline of the voice call](/quest/m1/voice-buffers.md) - every frame, buffer, VAD decision, and model step on one axis, marking frames late for a typical WebRTC jitter buffer

src/components/voice-demo.tsx

@@ -9,7 +9,9 @@ import type MoqPublish from "@moq/publish/element";
 import "@moq/watch/element";
 import type MoqWatch from "@moq/watch/element";
 import { createEffect, createSignal, For, Index, onCleanup, Show } from "solid-js";
+import { type AgentRecord, CallLog } from "./voice/call";
 import Cat from "./voice/cat";
+import Timeline from "./voice/timeline";
 
 type Turnstile = {
 	render: (
@@ -63,6 +65,8 @@ const TURNSTILE_SCRIPT = "https://challenges.cloudflare.com/turnstile/v0/api.js?
 // The agent subscribes to this fixed-name track on our broadcast and publishes its own beside
 // its audio. Neither is in the catalog.
 const RTVI_TRACK = "transcript.json.z";
+// The agent's own measurements of the call (moq.pro agent/VOICE.md), a window of records.
+const TIMELINE_TRACK = "voice.json";
 const RTVI_VERSION = "2.1.0";
 
 // How often a caller waiting for pickup asks whether an edge refused it.
@@ -122,8 +126,9 @@ type Call = {
 	ticker?: ReturnType<typeof setInterval>;
 	liveAt?: number;
 	epoch: string;
-	log: Message[];
+	messages: Message[];
 	producer?: Json.Stream.Producer<Message>;
+	log: CallLog;
 };
 
 let turnstilePromise: Promise<Turnstile> | undefined;
@@ -199,6 +204,8 @@ export default function VoiceDemo() {
 	const [transcript, setTranscript] = createSignal<Turn[]>([]);
 	// The agent's decoded audio, before volume, for the cat to lip-sync to.
 	const [voice, setVoice] = createSignal<AudioNode>();
+	// The last call's measurements, kept after it ends so the reader can look them over.
+	const [timeline, setTimeline] = createSignal<CallLog>();
 
 	// Bumped by every start and teardown, so an await that resumes after Stop sees it is stale
 	// and backs out instead of resurrecting the call.
@@ -221,8 +228,8 @@ export default function VoiceDemo() {
 	};
 
 	const send = (call: Call, message: Message) => {
-		const record = { ...message, seq: call.log.length, epoch: call.epoch };
-		call.log.push(record);
+		const record = { ...message, seq: call.messages.length, epoch: call.epoch };
+		call.messages.push(record);
 		call.producer?.append({ value: record });
 	};
 
@@ -236,6 +243,7 @@ export default function VoiceDemo() {
 		if (call) {
 			for (const timer of call.timers) clearTimeout(timer);
 			clearInterval(call.ticker);
+			call.log.ended = performance.now();
 			// The agent hangs up at once on this, instead of waiting for us to redial.
 			send(call, { label: "moq-transport", type: "session-ending" });
 			const unmount = () => setMounted((m) => (m === call ? undefined : m));
@@ -308,7 +316,8 @@ export default function VoiceDemo() {
 		const epoch = Array.from(crypto.getRandomValues(new Uint8Array(8)), (b) => b.toString(16).padStart(2, "0")).join(
 			"",
 		);
-		const call: Call = { gen, session, timers: [], epoch, log: [] };
+		const call: Call = { gen, session, timers: [], epoch, messages: [], log: new CallLog() };
+		setTimeline(call.log);
 		current = call;
 		// Queued until the publisher's broadcast exists, so the agent reads it first.
 		send(call, {
@@ -420,7 +429,7 @@ export default function VoiceDemo() {
 				track: broadcast.createTrack(RTVI_TRACK),
 				compression: "deflate",
 			});
-			for (const record of call.log) producer.append({ value: record });
+			for (const record of call.messages) producer.append({ value: record });
 			call.producer = producer;
 			effect.cleanup(() => {
 				if (call.producer === producer) call.producer = undefined;
@@ -438,11 +447,35 @@ export default function VoiceDemo() {
 			setVoice(root);
 			effect.cleanup(() => setVoice((v) => (v === root ? undefined : v)));
 		});
+		effects.run((effect) => {
+			const media = effect.get(el.audio.out.timestamp);
+			const playing = !effect.get(el.audio.out.stalled);
+			call.log.playhead(performance.now(), media, playing);
+		});
+		effects.spawn(async () => {
+			const arrivals = el.audio.arrivals();
+			effects.cleanup(() => arrivals.close());
+			for (;;) {
+				const next = await arrivals.next();
+				if (!next) return;
+				if (next.kind === "frame") call.log.received(next.timestamp, next.arrival, next.late);
+				else call.log.gap(next.timestamp, next.duration);
+			}
+		});
 		effects.run((effect) => {
 			const active = effect.get(el.broadcast.out.active);
 			if (!active) return;
 			onPickup(call);
 
+			const timeline = active.track(TIMELINE_TRACK).subscribe();
+			effect.cleanup(() => timeline.close());
+			const records = new Json.Window.Consumer<AgentRecord>({ track: timeline });
+			effect.spawn(async () => {
+				for await (const { value } of records) {
+					if ("push" in value) call.log.agent(value.push.value);
+				}
+			});
+
 			const track = active.track(RTVI_TRACK).subscribe();
 			effect.cleanup(() => track.close());
 			const consumer = new Json.Stream.Consumer<Message>({ track, compression: "deflate" });
@@ -556,6 +589,10 @@ export default function VoiceDemo() {
 				</ol>
 			</Show>
 
+			<Show when={timeline()} keyed>
+				{(log) => <Timeline call={log} />}
+			</Show>
+
 			<Show when={mounted()} keyed>
 				{(call) => {
 					const effects = new Signals.Effect();

src/components/voice/call.test.ts

@@ -0,0 +1,179 @@
+import { describe, expect, test } from "bun:test";
+import { CallLog } from "./call";
+
+// Millisecond ticks, so media times read as written.
+const arrival = (capture: number, arrive: number) => ({
+	version: 2,
+	type: "arrival",
+	timescale: 1000,
+	media_ticks: capture,
+	arrival_ns: arrive * 1e6,
+});
+
+const publish = (track: string, start: number, end: number, at: number) => ({
+	version: 2,
+	type: "publish",
+	track,
+	timescale: 1000,
+	start_ticks: start,
+	end_ticks: end,
+	at_ns: at * 1e6,
+});
+
+// The agent's clock is 1000 ms ahead of ours, with 10 ms of transit each way.
+function synced(): CallLog {
+	const log = new CallLog();
+	log.agent(arrival(0, 1010));
+	log.agent(publish("bot-audio", 0, 5000, 1100));
+	log.received(0, 110, false);
+	return log;
+}
+
+describe("clock", () => {
+	test("places the agent's clock from the fastest transit each way", () => {
+		const log = synced();
+		expect(log.offset).toBe(1000);
+		expect(log.error).toBe(10);
+	});
+});
+
+describe("clock", () => {
+	// The first downlink sample is 90 ms slow, so the estimate starts 45 ms off (error 55).
+	function slow(): CallLog {
+		const log = new CallLog();
+		log.agent(arrival(0, 1010));
+		log.agent(publish("bot-audio", 0, 1000, 1100));
+		log.received(0, 200, false);
+		expect(log.offset).toBe(955);
+		return log;
+	}
+
+	test("a faster frame refines the estimate when its record comes first", () => {
+		const log = slow();
+		log.agent(publish("bot-audio", 1000, 2000, 1200));
+		log.received(1000, 210, false);
+		expect(log.offset).toBe(1000);
+		expect(log.error).toBe(10);
+	});
+
+	test("a faster frame refines the estimate when it beats its record", () => {
+		const log = slow();
+		log.received(1000, 210, false);
+		log.agent(publish("bot-audio", 1000, 2000, 1200));
+		expect(log.offset).toBe(1000);
+	});
+
+	test("a frame either of two overlapping publishes could have sent leaves the estimate alone", () => {
+		const log = slow();
+		// The barge-in's publish was sent at 215 local, after this frame arrived, but the loose estimate
+		// can't rule it out. Measuring transit from it would claim 1007.5 ± 2.5, excluding the truth.
+		log.agent(publish("bot-audio-1", 500, 1500, 1215));
+		log.received(600, 210, false);
+		expect(log.offset).toBe(955);
+		expect(log.error).toBe(55);
+	});
+});
+
+describe("publish", () => {
+	test("a barge-in's track claims the frames that beat its record", () => {
+		const log = synced();
+		// The new track resumes at media 2000. One frame of the old track arrived before it was sent,
+		// and one of the new track arrived before its record did.
+		log.received(2020, 2500, false);
+		log.received(2000, 3050, false);
+		log.agent(publish("bot-audio-1", 2000, 3000, 4000));
+
+		const [old, claimed] = [log.reply.find((f) => f.media === 2020), log.reply.find((f) => f.media === 2000)];
+		expect(old?.track).toBe("bot-audio");
+		expect(claimed?.track).toBe("bot-audio-1");
+		expect(claimed && log.downLateness(claimed)).toBe(40);
+	});
+
+	test("a claimed frame sheds the retired track's discard", () => {
+		const log = synced();
+		log.received(2020, 2500, false);
+		log.received(2000, 3050, false);
+		log.agent({
+			version: 2,
+			type: "retire",
+			track: "bot-audio",
+			timescale: 1000,
+			unplayed_ticks: 3000,
+			at_ns: 3900 * 1e6,
+		});
+		log.agent(publish("bot-audio-1", 2000, 3000, 4000));
+
+		const [old, claimed] = [log.reply.find((f) => f.media === 2020), log.reply.find((f) => f.media === 2000)];
+		expect(old?.discarded).toBe(true);
+		expect(claimed?.track).toBe("bot-audio-1");
+		expect(claimed?.discarded).toBeFalsy();
+		expect(claimed?.play).toBeUndefined();
+	});
+
+	test("a frame of a track retired before it arrived never plays", () => {
+		const log = synced();
+		log.agent({
+			version: 2,
+			type: "retire",
+			track: "bot-audio",
+			timescale: 1000,
+			unplayed_ticks: 3000,
+			at_ns: 3900 * 1e6,
+		});
+		log.received(3000, 3000, false);
+
+		const late = log.reply.find((f) => f.media === 3000);
+		expect(late?.track).toBe("bot-audio");
+		expect(late?.discarded).toBe(true);
+		expect(late?.play).toBeDefined();
+	});
+
+	test("a frame of a track retired before any reply arrived is placed at the retirement", () => {
+		const log = new CallLog();
+		log.agent(arrival(0, 1010));
+		log.agent(publish("bot-audio", 0, 5000, 1100));
+		log.agent({
+			version: 2,
+			type: "retire",
+			track: "bot-audio",
+			timescale: 1000,
+			unplayed_ticks: 5000,
+			at_ns: 1200 * 1e6,
+		});
+		log.received(0, 300, false);
+
+		const frame = log.reply.find((f) => f.media === 0);
+		expect(frame?.discarded).toBe(true);
+		expect(frame?.play).toBeDefined();
+	});
+
+	test("an older publish still holds media past a newer, shorter one", () => {
+		const log = synced();
+		log.agent(publish("bot-audio-1", 2000, 3000, 4000));
+		log.received(4500, 4100, false);
+
+		expect(log.reply.find((f) => f.media === 4500)?.track).toBe("bot-audio");
+	});
+});
+
+describe("exchange", () => {
+	test("starts with the earlier of your latest statement and the agent's latest reply", () => {
+		const log = synced();
+		const record = (type: string, at: number, fields: object) => ({
+			version: 2,
+			type,
+			timescale: 1000,
+			at_ns: at * 1e6,
+			...fields,
+		});
+		// The agent starts replying at 3000 local, then you interrupt: speech from 5000, confirmed at 5200.
+		log.agent(record("llm", 4000, { event: "request" }));
+		log.agent(record("vad", 6210, { speaking: true, media_ticks: 5200, confirm_ticks: 200 }));
+		expect(log.exchange()).toBe(3000);
+		log.agent(record("turn", 6250, { speaking: true, media_ticks: 5240 }));
+		expect(log.exchange()).toBe(3000);
+		// Its next reply is to your interruption, so the exchange now starts where you did.
+		log.agent(record("llm", 7000, { event: "request" }));
+		expect(log.exchange()).toBe(5000);
+	});
+});