From c1ac687dcfc343f8641e8dc90d0c2921ae0786ea Mon Sep 17 00:00:00 2001 From: Yahya Alhinai Date: Sun, 28 Jun 2026 06:45:37 +0000 Subject: [PATCH] feat(voice): on-demand PodMan voice test + Hermes notify + TTS tuning Snapshot of in-progress voice work, committed to unblock concurrent edits: speakInRoom + POST voice-test endpoint, Test PodMan voice button (PodView/api), Hermes notify action + scripts/hermes-notify.mjs, agent exits on LiveKit disconnect for auto-restart, TTS playback tuning (subscriber-ready delay, preroll/tail silence, fallback line, microphone source), verify script updates. Co-Authored-By: Claude Opus 4.8 --- backend/src/action/hermes.ts | 53 ++++++++++++ backend/src/agent.ts | 6 ++ backend/src/agent/podman.ts | 20 ++--- backend/src/env.ts | 1 + backend/src/server.ts | 126 ++++++++++++++++++++++++++-- backend/src/voice/live.ts | 65 ++++++++++++-- frontend/src/components/PodView.tsx | 65 ++++++++++++-- frontend/src/lib/api.ts | 11 +++ package.json | 1 + scripts/hermes-notify.mjs | 54 ++++++++++++ scripts/verify-backend.mjs | 20 +++++ scripts/verify-frontend.mjs | 47 ++++++++++- 12 files changed, 438 insertions(+), 31 deletions(-) create mode 100644 scripts/hermes-notify.mjs diff --git a/backend/src/action/hermes.ts b/backend/src/action/hermes.ts index 3f654e4..076a71b 100644 --- a/backend/src/action/hermes.ts +++ b/backend/src/action/hermes.ts @@ -1,6 +1,10 @@ import type { Room } from '@livekit/rtc-node'; +import { Room as LiveKitRoom } from '@livekit/rtc-node'; +import { AccessToken } from 'livekit-server-sdk'; import type { Collision, DataMessage, HermesMessage, Intervention } from '@podman/shared'; import { DATA_TOPIC } from '@podman/shared'; +import { env } from '../env.js'; +import { speak } from '../voice/live.js'; const encoder = new TextEncoder(); @@ -37,3 +41,52 @@ export async function publishHermesMessage( topic: DATA_TOPIC, }); } + +export async function publishHermesIntervention( + room: Room, + collision: Collision, + intervention: Intervention, + voiceLine?: string, +): Promise { + const data: DataMessage = { type: 'COLLISION', collision, intervention }; + await room.localParticipant?.publishData(encoder.encode(JSON.stringify(data)), { + reliable: true, + topic: DATA_TOPIC, + }); + await publishHermesMessage(room, collision, intervention); + if (voiceLine) await speak(room, voiceLine); +} + +async function hermesToken(roomName: string): Promise { + const at = new AccessToken(env.LIVEKIT_API_KEY, env.LIVEKIT_API_SECRET, { + identity: `podman-hermes-${Date.now()}`, + name: 'PodMan Hermes', + ttl: '10m', + }); + at.addGrant({ + roomJoin: true, + room: roomName, + canPublish: true, + canSubscribe: true, + canPublishData: true, + }); + return at.toJwt(); +} + +export async function notifyHermesInterventionInRoom( + roomName: string, + collision: Collision, + intervention: Intervention, + voiceLine?: string, +): Promise { + const room = new LiveKitRoom(); + try { + await room.connect(env.LIVEKIT_URL, await hermesToken(roomName), { + autoSubscribe: false, + dynacast: false, + }); + await publishHermesIntervention(room, collision, intervention, voiceLine); + } finally { + await room.disconnect().catch(() => {}); + } +} diff --git a/backend/src/agent.ts b/backend/src/agent.ts index 29d2296..06487a8 100644 --- a/backend/src/agent.ts +++ b/backend/src/agent.ts @@ -173,6 +173,12 @@ async function main() { await withTimeout(dispose(), SHUTDOWN_GRACE_MS); process.exit(0); }; + room.on(RoomEvent.Disconnected, () => { + if (!shuttingDown) { + console.error('[agent] LiveKit disconnected; exiting so systemd restarts the worker'); + process.exit(1); + } + }); process.on('SIGINT', shutdown); process.on('SIGTERM', shutdown); } diff --git a/backend/src/agent/podman.ts b/backend/src/agent/podman.ts index ad55b6e..7bab328 100644 --- a/backend/src/agent/podman.ts +++ b/backend/src/agent/podman.ts @@ -1,6 +1,5 @@ import { RoomEvent, type Room } from '@livekit/rtc-node'; import type { EngineerContext, Collision, Intervention, DataMessage } from '@podman/shared'; -import { DATA_TOPIC } from '@podman/shared'; import { analyzeFrame } from '../vision/gemini.js'; import { detectCollisions } from '../collision/detector.js'; import { getGithubState } from '../github/client.js'; @@ -13,12 +12,10 @@ import { import { getGitStates } from '../memory/db.js'; import { recallSimilar } from '../memory/vectors.js'; import { shouldIntervene, preferredAction } from '../memory/policy.js'; -import { speak } from '../voice/live.js'; -import { publishHermesMessage } from '../action/hermes.js'; +import { publishHermesIntervention } from '../action/hermes.js'; export class PodMan { private contexts = new Map(); - private encoder = new TextEncoder(); constructor( private room: Room, @@ -82,7 +79,7 @@ export class PodMan { (prior ? ' Seen before.' : ''); // Spoken line stays short, but uses natural phrasing for Gemini TTS prosody. - const voiceLine = `Heads up. ${names} are both editing ${shortFile}. Please sync before pushing.`; + const voiceLine = `${names} are both editing ${shortFile}. Please sync before pushing.`; const intervention: Intervention = { id: `int_${Date.now()}`, @@ -103,12 +100,11 @@ export class PodMan { }; await recordIntervention(intervention); - const data: DataMessage = { type: 'COLLISION', collision, intervention }; - await this.room.localParticipant?.publishData(this.encoder.encode(JSON.stringify(data)), { - reliable: true, - topic: DATA_TOPIC, - }); - await publishHermesMessage(this.room, collision, intervention); - if (collision.severity === 'critical') await speak(this.room, voiceLine); + await publishHermesIntervention( + this.room, + collision, + intervention, + collision.severity === 'critical' ? voiceLine : undefined, + ); } } diff --git a/backend/src/env.ts b/backend/src/env.ts index 976d026..b8197c6 100644 --- a/backend/src/env.ts +++ b/backend/src/env.ts @@ -21,6 +21,7 @@ export const env = { LIVEKIT_URL: req('LIVEKIT_URL'), LIVEKIT_API_KEY: req('LIVEKIT_API_KEY'), LIVEKIT_API_SECRET: req('LIVEKIT_API_SECRET'), + LIVEKIT_AGENT_NAME: opt('LIVEKIT_AGENT_NAME'), // Gemini GEMINI_API_KEY: reqAny('GEMINI_API_KEY', ['GOOGLE_API_KEY', 'GOOGLE_GENERATIVE_AI_API_KEY']), GEMINI_VISION_MODEL: opt('GEMINI_VISION_MODEL', 'gemini-2.0-flash'), diff --git a/backend/src/server.ts b/backend/src/server.ts index c1e1212..6448eb4 100644 --- a/backend/src/server.ts +++ b/backend/src/server.ts @@ -1,11 +1,12 @@ import express from 'express'; import cors from 'cors'; import { createServer } from 'node:http'; +import type { Socket } from 'node:net'; import { WebSocketServer } from 'ws'; -import { AccessToken, RoomConfiguration } from 'livekit-server-sdk'; +import { AccessToken, RoomAgentDispatch, RoomConfiguration } from 'livekit-server-sdk'; import { env } from './env.js'; import { createSyncPr } from './github/client.js'; -import { recordOutcome, memoryStats } from './memory/store.js'; +import { recordCollision, recordIntervention, recordOutcome, memoryStats } from './memory/store.js'; import { closeMemory, initMemory } from './memory/db.js'; import { listPods, @@ -20,13 +21,27 @@ import { import { getPresence, closeRoom } from './livekit/rooms.js'; import { loadPodGraph, reachFrom } from './graph/store.js'; import { listPodActivity } from './activity/store.js'; -import type { InterventionOutcome } from '@podman/shared'; +import { speakInRoom } from './voice/live.js'; +import { notifyHermesInterventionInRoom } from './action/hermes.js'; +import type { Collision, Intervention, InterventionOutcome, SuggestedActionKind } from '@podman/shared'; const app = express(); app.use(cors()); app.use(express.json()); app.get('/health', (_req, res) => res.json({ ok: true })); +function stringArray(value: unknown): string[] { + return Array.isArray(value) + ? value.map((item) => String(item).trim()).filter(Boolean) + : []; +} + +function suggestedAction(value: unknown): SuggestedActionKind { + return value === 'open_sync_pr' || value === 'ping_teammate' || value === 'none' + ? value + : 'ping_teammate'; +} + // Mint a LiveKit token for an engineer joining a pod. app.post('/api/token', async (req, res) => { const { room, identity, name, githubLogin } = req.body ?? {}; @@ -38,9 +53,22 @@ app.post('/api/token', async (req, res) => { metadata: JSON.stringify({ githubLogin: githubLogin ?? name }), }); at.addGrant({ roomJoin: true, room, canPublish: true, canSubscribe: true, canPublishData: true }); + const agents = env.LIVEKIT_AGENT_NAME + ? [ + new RoomAgentDispatch({ + agentName: env.LIVEKIT_AGENT_NAME, + metadata: JSON.stringify({ podId: room }), + }), + ] + : undefined; // Auto-clean the room: close 60s after it empties, drop a participant 20s // after they disconnect. Applied when LiveKit auto-creates the room. - at.roomConfig = new RoomConfiguration({ name: room, emptyTimeout: 60, departureTimeout: 20 }); + at.roomConfig = new RoomConfiguration({ + name: room, + emptyTimeout: 60, + departureTimeout: 20, + agents, + }); res.json({ token: await at.toJwt(), url: env.LIVEKIT_URL }); }); @@ -137,6 +165,84 @@ app.post('/api/pods/:id/members', async (req, res) => { } }); +app.post('/api/pods/:id/voice-test', async (req, res) => { + const podId = req.params.id; + const message = + typeof req.body?.message === 'string' && req.body.message.trim() + ? req.body.message.trim() + : 'PodMan voice test. Gemini TTS is playing through LiveKit.'; + try { + await speakInRoom(podId, message); + res.json({ ok: true }); + } catch (e) { + res.status(500).json({ error: (e as Error).message }); + } +}); + +app.post('/api/pods/:id/hermes/notify', async (req, res) => { + const podId = req.params.id; + const pod = await getPod(podId); + if (!pod) return res.status(404).json({ error: 'pod not found' }); + + const body = req.body ?? {}; + const message = typeof body.message === 'string' ? body.message.trim() : ''; + if (!message) return res.status(400).json({ error: 'message is required' }); + + const now = new Date().toISOString(); + const engineers = stringArray(body.engineers); + const recipients = engineers.length ? engineers : pod.members.slice(0, 2); + const file = typeof body.file === 'string' && body.file.trim() ? body.file.trim() : 'Hermes signal'; + const urgent = body.urgency === 'urgent' || body.severity === 'critical'; + const collision: Collision = { + id: typeof body.collisionId === 'string' && body.collisionId ? body.collisionId : `col_${Date.now()}`, + podId, + file, + symbol: typeof body.symbol === 'string' && body.symbol ? body.symbol : undefined, + engineers: recipients, + severity: urgent ? 'critical' : 'warn', + githubState: { unpushed: body.unpushed !== false }, + detectedAt: now, + }; + const intervention: Intervention = { + id: + typeof body.interventionId === 'string' && body.interventionId + ? body.interventionId + : `int_${Date.now()}`, + collisionId: collision.id, + podId, + kind: urgent ? 'voice' : 'card', + message, + suggestedAction: { + kind: suggestedAction(body.suggestedAction), + params: { + file, + engineers: recipients, + source: 'local-hermes', + }, + }, + status: 'pending', + createdAt: now, + }; + const voiceLine = + urgent && body.speak !== false + ? typeof body.voiceLine === 'string' && body.voiceLine.trim() + ? body.voiceLine.trim() + : message + : undefined; + + try { + await recordCollision(collision); + await recordIntervention(intervention); + if (body.dryRun === true) { + return res.status(202).json({ ok: true, collision, intervention, livekit: 'dry-run' }); + } + await notifyHermesInterventionInRoom(podId, collision, intervention, voiceLine); + res.status(202).json({ ok: true, collision, intervention, livekit: 'notified' }); + } catch (e) { + res.status(500).json({ error: (e as Error).message }); + } +}); + app.delete('/api/pods/:id/members/:name', async (req, res) => { const pod = await removeMember(req.params.id, req.params.name); if (!pod) return res.status(404).json({ error: 'pod not found' }); @@ -206,6 +312,12 @@ app.get('/api/pods/:id/activity/stream', async (req, res) => { }); const http = createServer(app); +const sockets = new Set(); + +http.on('connection', (socket) => { + sockets.add(socket); + socket.on('close', () => sockets.delete(socket)); +}); // ws relay: the agent pushes collision/intervention JSON here; PWAs subscribed by pod receive it. const wss = new WebSocketServer({ server: http, path: '/api/events' }); @@ -237,7 +349,11 @@ async function shutdown(signal: NodeJS.Signals): Promise { console.log(`[server] ${signal} received; shutting down`); for (const client of clients) client.close(); wss.close(); - await new Promise((resolve) => http.close(() => resolve())); + for (const socket of sockets) socket.destroy(); + await Promise.race([ + new Promise((resolve) => http.close(() => resolve())), + new Promise((resolve) => setTimeout(resolve, 5000)), + ]); await closeMemory().catch((e) => console.warn(`[memory] close failed: ${(e as Error).message}`)); process.exit(0); } diff --git a/backend/src/voice/live.ts b/backend/src/voice/live.ts index 5536010..4494563 100644 --- a/backend/src/voice/live.ts +++ b/backend/src/voice/live.ts @@ -3,24 +3,32 @@ import { AudioFrame, AudioSource, LocalAudioTrack, + Room, TrackPublishOptions, TrackSource, type LocalParticipant, - type Room, } from '@livekit/rtc-node'; import { GoogleGenAI, Modality, type LiveServerMessage, type Session } from '@google/genai'; +import { AccessToken } from 'livekit-server-sdk'; import { DATA_TOPIC, type DataMessage } from '@podman/shared'; import { env } from '../env.js'; const SAMPLE_RATE = 24_000; const CHANNELS = 1; const FRAME_SAMPLES = SAMPLE_RATE / 10; +const SUBSCRIBER_READY_MS = 750; +const AUDIO_PREROLL_MS = 300; +const AUDIO_TAIL_MS = 300; const VOICE_QUEUE_MS = 30_000; const VOICE_TRACK_PREFIX = 'podman-hermes-voice'; const encoder = new TextEncoder(); const ai = new GoogleGenAI({ apiKey: env.GEMINI_API_KEY }); let voiceQueue: Promise = Promise.resolve(); +function delay(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + function ttsPrompt(message: string): string { return [ 'Speak this PodMan coordination alert as a calm, natural engineering teammate.', @@ -55,7 +63,8 @@ function audioFrameFromBase64(data: string, mimeType?: string): AudioFrame | nul const buf = Buffer.from(data, 'base64'); if (buf.byteLength < 2) return null; const bytes = buf.byteLength % 2 === 0 ? buf : buf.subarray(0, buf.byteLength - 1); - const samples = new Int16Array(bytes.buffer, bytes.byteOffset, bytes.byteLength / 2); + const samples = new Int16Array(bytes.byteLength / 2); + for (let i = 0; i < samples.length; i += 1) samples[i] = bytes.readInt16LE(i * 2); return new AudioFrame(samples, SAMPLE_RATE, CHANNELS, samples.length / CHANNELS); } @@ -78,7 +87,7 @@ function framesFromPcmBase64(data: string, mimeType?: string): AudioFrame[] { const samples = frame.data; const frames: AudioFrame[] = []; for (let offset = 0; offset < samples.length; offset += FRAME_SAMPLES) { - const chunk = samples.subarray(offset, Math.min(offset + FRAME_SAMPLES, samples.length)); + const chunk = samples.slice(offset, Math.min(offset + FRAME_SAMPLES, samples.length)); frames.push(new AudioFrame(chunk, SAMPLE_RATE, CHANNELS, chunk.length / CHANNELS)); } return frames; @@ -100,8 +109,24 @@ async function generateTtsFrames(message: string): Promise { ); } +function fallbackVoiceLine(message: string): string { + const clean = message.replace(/^heads up[.!]?\s*/i, '').trim(); + if (clean && clean !== message) return clean; + return 'PodMan noticed a critical conflict. Please sync with the team before pushing.'; +} + async function speakWithTts(source: AudioSource, message: string): Promise { - for (const frame of await generateTtsFrames(message)) { + let frames: AudioFrame[]; + try { + frames = await generateTtsFrames(message); + } catch (err) { + const fallback = fallbackVoiceLine(message); + console.warn(`[voice] Gemini TTS retrying with fallback line: ${(err as Error).message}`); + frames = await generateTtsFrames(fallback); + } + if (frames.length === 0) throw new Error('Gemini TTS returned no audio frames'); + console.log(`[voice] publishing Gemini TTS audio frames=${frames.length}`); + for (const frame of frames) { await source.captureFrame(frame); } } @@ -150,6 +175,11 @@ async function waitForVoicePlayout(source: AudioSource): Promise { ]); } +async function captureSilence(source: AudioSource, durationMs: number): Promise { + const samples = Math.max(1, Math.round((SAMPLE_RATE * durationMs) / 1000)); + await source.captureFrame(new AudioFrame(new Int16Array(samples), SAMPLE_RATE, CHANNELS, samples)); +} + async function speakAudio(room: Room, message: string): Promise { const localParticipant = room.localParticipant; if (!localParticipant) return; @@ -157,14 +187,17 @@ async function speakAudio(room: Room, message: string): Promise { const source = new AudioSource(SAMPLE_RATE, CHANNELS, VOICE_QUEUE_MS); const track = LocalAudioTrack.createAudioTrack(`${VOICE_TRACK_PREFIX}-${Date.now()}`, source); const options = new TrackPublishOptions(); - options.source = TrackSource.SOURCE_UNKNOWN; + options.source = TrackSource.SOURCE_MICROPHONE; let publicationSid: string | undefined; try { const publication = await localParticipant.publishTrack(track, options); publicationSid = publication.sid; + await delay(SUBSCRIBER_READY_MS); + await captureSilence(source, AUDIO_PREROLL_MS); if (env.GEMINI_LIVE_MODEL.includes('tts')) await speakWithTts(source, message); else await speakWithLive(source, message); + await captureSilence(source, AUDIO_TAIL_MS); await waitForVoicePlayout(source); } catch (err) { console.warn(`[voice] Gemini voice publish failed: ${(err as Error).message}`); @@ -188,3 +221,25 @@ export async function speak(room: Room, message: string): Promise { voiceQueue = voiceQueue.catch(() => {}).then(() => speakAudio(room, message)); await voiceQueue; } + +export async function speakInRoom(roomName: string, message: string): Promise { + const room = new Room(); + try { + const at = new AccessToken(env.LIVEKIT_API_KEY, env.LIVEKIT_API_SECRET, { + identity: `podman-voice-${Date.now()}`, + name: 'PodMan voice', + ttl: '5m', + }); + at.addGrant({ + roomJoin: true, + room: roomName, + canPublish: true, + canSubscribe: true, + canPublishData: true, + }); + await room.connect(env.LIVEKIT_URL, await at.toJwt()); + await speak(room, message); + } finally { + await room.disconnect().catch(() => {}); + } +} diff --git a/frontend/src/components/PodView.tsx b/frontend/src/components/PodView.tsx index 6287176..73c5efe 100644 --- a/frontend/src/components/PodView.tsx +++ b/frontend/src/components/PodView.tsx @@ -20,9 +20,10 @@ import { WorkflowIcon, XIcon, } from 'lucide-react'; -import type { Room, RemoteTrack, RemoteTrackPublication, RemoteParticipant } from 'livekit-client'; +import type { Room, RemoteTrack, RemoteTrackPublication } from 'livekit-client'; import type { Pod, PodActivityEvent, PodActivityKind, PodActivitySource } from '@podman/shared'; import { useBeat } from '../livekit/useBeat.js'; +import { testPodVoice } from '../lib/api.js'; import { useInterventions, primeSpeech } from '../livekit/useInterventions.js'; import { usePodActivity } from '../hooks/use-pod-activity.js'; import LiveWaveform from '@/components/ruixen/live-waveform'; @@ -109,6 +110,7 @@ export function PodView({ }) { const [participants, setParticipants] = useState([]); const [sharing, setSharing] = useState(false); + const [testingVoice, setTestingVoice] = useState(false); const [note, setNote] = useState(null); const [audioBlocked, setAudioBlocked] = useState(false); const [leftStreamOpen, setLeftStreamOpen] = useState(() => @@ -122,6 +124,7 @@ export function PodView({ const activity = usePodActivity(team.id, me); const audioRef = useRef(null); + const audioElementsRef = useRef(new Map()); const screenTrackRef = useRef(null); const onLeaveRef = useRef(onLeave); onLeaveRef.current = onLeave; @@ -131,12 +134,34 @@ export function PodView({ const refresh = () => setParticipants(snapshot(room, me)); refresh(); - const onAudio = (track: RemoteTrack, _pub: RemoteTrackPublication, _p: RemoteParticipant) => { - if (track.kind === Track.Kind.Audio && audioRef.current) { - audioRef.current.appendChild(track.attach()); - } + const attachAudio = (track: RemoteTrack, pub: RemoteTrackPublication) => { + if (track.kind !== Track.Kind.Audio || !audioRef.current) return; + const key = pub.trackSid || track.sid || track.mediaStreamTrack.id; + if (audioElementsRef.current.has(key)) return; + const element = track.attach(); + element.autoplay = true; + audioElementsRef.current.set(key, element); + audioRef.current.appendChild(element); }; - const onAudioGone = (track: RemoteTrack) => track.detach().forEach((el) => el.remove()); + const removeAudio = (track: RemoteTrack, pub?: RemoteTrackPublication) => { + const key = pub?.trackSid || track.sid || track.mediaStreamTrack.id; + const attached = audioElementsRef.current.get(key); + if (attached) { + attached.remove(); + audioElementsRef.current.delete(key); + } + track.detach().forEach((el) => el.remove()); + }; + const attachExistingAudio = () => { + room.remoteParticipants.forEach((participant) => { + participant.audioTrackPublications.forEach((publication) => { + const track = publication.track; + if (track) attachAudio(track, publication); + }); + }); + }; + const onAudio = (track: RemoteTrack, pub: RemoteTrackPublication) => attachAudio(track, pub); + const onAudioGone = (track: RemoteTrack, pub: RemoteTrackPublication) => removeAudio(track, pub); const onDisconnected = () => onLeaveRef.current(); // Browsers block autoplay of incoming audio until a user gesture unlocks it. // Surface a button whenever the room can't play sound so PodMan's voice cues @@ -152,6 +177,7 @@ export function PodView({ .on(RoomEvent.TrackUnsubscribed, onAudioGone) .on(RoomEvent.AudioPlaybackStatusChanged, onPlaybackChanged) .on(RoomEvent.Disconnected, onDisconnected); + attachExistingAudio(); return () => { room @@ -162,6 +188,8 @@ export function PodView({ .off(RoomEvent.TrackUnsubscribed, onAudioGone) .off(RoomEvent.AudioPlaybackStatusChanged, onPlaybackChanged) .off(RoomEvent.Disconnected, onDisconnected); + audioElementsRef.current.forEach((el) => el.remove()); + audioElementsRef.current.clear(); }; }, [room, me]); @@ -200,6 +228,22 @@ export function PodView({ } } + async function playPodManVoiceTest() { + if (!room) return; + setNote(null); + setTestingVoice(true); + try { + await room.startAudio().catch(() => {}); + setAudioBlocked(!room.canPlaybackAudio); + await testPodVoice(team.id); + setNote('PodMan voice test sent.'); + } catch (e) { + setNote(`PodMan voice test failed: ${(e as Error).message}`); + } finally { + setTestingVoice(false); + } + } + async function toggleScreen() { primeSpeech(); // unlock browser voice from this gesture too if (!room) return; @@ -317,6 +361,14 @@ export function PodView({ {beat.on ? (beat.mine ? 'Stop audio' : `Stop (${beat.by})`) : 'Test audio'} +