refactor(voice): unify runtime input and output

[nvddr]

Summary

• add one bidirectional VoiceAgent that owns VoiceSession, publishes accepted UserQuery plus voice lifecycle events to injected typed topics, and subscribes to voice.output
• keep media readiness, VAD/STT/TTS, wake gating, interruption, transport, and cleanup private to the owned session
• support finite and correlated incremental voice output without exposing Pipecat or a general runtime stream primitive
• migrate simple-vlm-example so SimpleVlmAgent owns its vision tool, streamed turns, cancellation, and participant frame cleanup while app.py only composes agents
• delete the superseded NAT voice adapter and response-returning handler API

Runtime boundary

The runtime provides typed publish/subscribe routing only. VoiceAgent.run(runtime) owns the voice session lifecycle, and application agents own their own resources and tasks. VoiceAgent owns the UserQuery, VoiceParticipantLeft, and VoiceInterrupted schemas while each sample chooses their topic names.

Participant departure is published with participant scope. Interruption can be participant-scoped or global. Application agents subscribe directly and clean up their own tasks, state, and tools; the composition root installs no transport callbacks and contains no resource logic.

Voice producers publish either a complete voice.output message or correlated incremental chunks. Incremental identity is scoped by participant, producer, and response ID, so independent agents cannot merge output accidentally. VoiceAgent owns the lock protecting response aggregation and FIFO state. Runtime publication waits for subscriber delivery, so no duplicate runtime RPC or SPEAK API is needed.

For simple VLM, SimpleVlmAgent owns the injected StreamingVisionTool, participant-scoped background tasks, cancellation, frame release, and nested stream cleanup. app.py constructs and registers the agents and supplies the sample-specific topic names only.

Review feedback addressed

• partial chunks survive iterator failures and remain in the final data response
• response stream keys include producer identity
• cancelled response streams release blocked publishers and close their source iterator
• originating query timestamps propagate to TTS output
• readiness-failure cleanup is idempotent
• global cancellation is modeled without a synthetic participant
• external response FIFO is preserved while queued work exists
• orphan and ambiguous empty stream terminators are rejected
• removed duplicate task error reporting and stale process-model documentation
• removed the compatibility adapter, handler-return path, observer hooks, query queue mode, and separate simple-VLM voice agent

Validation

• runtime, voice runtime, simple VLM, voice pipeline/session, and Pipecat bridge regressions: 192 passed
• stream-cleanup regressions rerun after conflict resolution: 2 passed
• Ruff, SPDX, and DCO pre-commit hooks passed

Rebased directly onto main after #350 merged and squashed to one commit, ef05580.

[wenxind-nvidia]

Blockers

1. agent-sdk/xr-ai-voice/xr_ai_voice/_processors/handler.py:348-359 keeps accumulated chunks inside _consume_response(). If the iterator raises after yielding, callers retain response = "" and finalize an empty agent turn. I reproduced both query and external paths emitting a TextFrame("partial output " ) followed by AssistantResponseEndFrame(text=""). The user hears partial speech, but observation and data echo lose it. This also regresses the existing query path, whose accumulator previously survived iterator errors. Please preserve accumulated text across failures and add coverage for an iterator that raises after yielding.

2. agent-sdk/xr-ai-nat/xr_ai_nat/events/voice.py:160 keys active streams only by (participant_id, response_id). Different agents can reuse local identifiers, causing their output to merge. I reproduced two producers using "answer-1" producing the stream A1, B1, A2, with B2 treated as a separate finite response. Please include the producer or another globally scoped identity in the key and test concurrent producers.

All 103 focused tests pass locally, and the current GitHub checks are green. PR 351 remains stacked on the unchanged PR 350 head, so PR 350's posted blockers also remain transitively relevant.

[yanziz-nvidia]

Reviewed by yanziz-reviewer-bot

Summary

Adds VoiceSession.enqueue_response(...), a typed VoiceOutput mailbox payload, and a VoiceOutputWorker subscriber that routes complete and streaming responses through the existing participant-aware TTS/data-echo path without coupling producers to Pipecat. CI is all green (pytest 3.11 + 3.12, ruff, CodeQL, lock-check, SPDX, DCO); no dependency changes.

Legend: 🚫 Blocker · 💡 Suggestion · 🔍 Nit

	Finding
🚫	None
💡	xr_ai_voice/_processors/handler.py:239 — _spawn_response can jump the per-participant queue when the inflight task is done but _start_next has not yet run. _finish_response awaits a frame push before popping _inflight; a caller landing in that window sees is_active=False and starts immediately, ahead of items already in _queued[pid].
🔍	xr_ai_nat/events/voice.py:40 — Empty-text final chunks with interrupt=True are silently dropped. The empty-text guard fires before the interrupt path, so the cancel is a no-op with no error.

Actionables (for bots — copy-paste-ready for AI)

Fix if it makes sense in context — these are agent-generated suggestions, not human-vetted obligations. Skip anything that's wrong, already addressed, or not worth the churn.

• agent-sdk/xr-ai-voice/xr_ai_voice/_processors/handler.py:239 — In _spawn_response, guard the direct _start_response call with if not self._queued.get(pid): and otherwise append to the queue, preserving FIFO across the _finish_response await gap.
• agent-sdk/xr-ai-nat/xr_ai_nat/events/voice.py:40 — In VoiceOutput.validate_boundary, reject final=True + non-None response_id + empty text, or document that interrupt has no effect on empty-text terminators.

[nvddr]

Updated in 92764ae on the rewritten #350. Wenxin's concerns are covered by retaining the accumulator outside _consume_response() (including iterator failures) and keying incremental output by participant, producer, and response ID. Yanzi's FIFO suggestion now queues whenever prior queued work exists, with a regression test; empty interrupting terminators are rejected. The former NAT event worker is gone: VoiceOutputAgent is an ordinary runtime call target and voice.output subscriber, while incremental aggregation stays private to voice/TTS. The focused combined suite passed (106 tests plus the final FIFO regression), targeted Pyright is clean, and both SDK wheels build. Please re-review the new head.

[nvddr]

Update: simple-vlm-example now uses the complete voice/runtime path in 2d277f7: typed and spoken input publish one query topic, a resident agent owns the streaming VLM task, and VoiceOutputAgent alone bridges voice.output to TTS. Participant/global interruption cancels runtime-owned work. The focused combined suite passes 113 tests. @wenxind-nvidia @blongs-nv @yanziz-nvidia, please re-review the updated head.

[blongs-nv]

Reviewed at ce095299c8989c472302b078c371766ff18c76e2 by a bot.

Blockers

1. A cancelled producer's stream terminator kills the entire runtime on the first barge-in or supersede.

   agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/agent.py:111-120
   On interruption or supersede the io processor cancels response playback first, which closes and evicts the stream (agent-sdk/xr-ai-voice/xr_ai_voice/_processors/io.py:114-118); the subsequent VoiceInterrupted/new-query delivery cancels _stream, whose finally still publishes the empty terminator, and the resulting ValueError (agent-sdk/xr-ai-voice/xr_ai_voice/_runtime.py:248) makes the cancelled task end with an exception, which _background_failed (agent-sdk/xr-ai-agent-runtime/xr_ai_runtime/runtime.py:196-201) escalates to whole-runtime teardown. I reproduced this end to end; with interrupt_on_supersede=True in the sample, "user speaks while the agent answers" is the primary interaction. A cancelled producer should not publish a terminator, and/or keys the voice side closed should be tombstoned so their terminators are swallowed.

2. VoiceAgent._output_lock is held across stream.send(), which blocks on queue capacity, so one queued or slow stream deadlocks all voice output.

   agent-sdk/xr-ai-voice/xr_ai_voice/_runtime.py:219, agent-sdk/xr-ai-voice/xr_ai_voice/_runtime.py:267
   With two producers on one participant, producer B's stream is FIFO-queued (unconsumed) behind in-flight A (agent-sdk/xr-ai-voice/xr_ai_voice/_processors/io.py:196-203); B fills its bounded queue and blocks holding the lock, A's next publish blocks acquiring it, A never terminates, and B's stream never starts. The same global lock also head-of-line-blocks every other participant. Hold the lock only for the _streams lookup/create and _enqueue_response, and call send() outside it.

3. A non-cancellation error in _run_response never closes the stream, and the regression test for cancel-time release passes with the fix removed.

   agent-sdk/xr-ai-voice/xr_ai_voice/_processors/io.py:233-238
_close_response runs only on CancelledError; if push_frame or the consume loop raises, the stream stays registered with no consumer and its producer eventually blocks while holding the output lock (blocker 2's failure mode). Separately, tests/test_voice_runtime.py:319 asserts _streams == {} only after async with runtime exits (lifespan teardown makes that tautological) and its capacity-1 queue is already drained before the cancel, so it was verified to still pass with the aclose-on-cancel branch deleted; make the test gate the consumer so a send() is genuinely blocked, and assert inside the runtime context.

4. A chunk arriving after the voice side closed a stream silently opens a new stream and speaks the tail of the answer the user just interrupted.

   agent-sdk/xr-ai-voice/xr_ai_voice/_runtime.py:256-266
   Barge-in evicts (participant, source, response_id) before the producer is cancelled via VoiceInterrupted, so any chunk published in that window recreates the key with no in-flight playback and plays immediately. The same tombstone mechanism as blocker 1 covers this.

5. Query fan-out occupies the participant's cancellable slot, so an interrupting first chunk cancels delivery of the query to every other subscriber.

   agent-sdk/xr-ai-voice/xr_ai_voice/_processors/io.py:152-160
   While _run_query awaits publish's gather over all query-topic subscribers it sits in _inflight[pid]; the first responder's interrupt=True chunk (as the sample sends at agent.py:105) triggers _cancel_pid, cancelling the gather and silently dropping delivery to the remaining subscribers. Multi-agent fan-out is this PR's stated purpose, so query delivery should not share the response playback slot.

6. The sample opens hub ZMQ sockets before readiness probes, contradicting the invariant this same diff writes in its docs.

   agent-samples/simple-vlm-example/worker/simple_vlm_example_worker/app.py:67
session.transport.endpoint at composition time lazily constructs HubVoiceTransport (_session.py:69-74), whose ProcessorEndpoint.__init__ connects sockets immediately (_processor.py:181-187), while docs/source/components/agent-sdk.md:339 and DEPENDENCIES.md:95 still say failed readiness never opens hub sockets. The base branch constructed the tool inside async with session:; pass an explicit transport or defer endpoint resolution into the agent's lifespan.

7. Deleting the NAT voice adapter left its packaging and the authoritative dependency docs behind.

   agent-sdk/xr-ai-nat/pyproject.toml:23
   The voice = ["xr-ai-voice"] extra and its [tool.uv.sources] entry (line 37) survive with nothing in xr_ai_nat importing xr_ai_voice, and DEPENDENCIES.md:150 plus its prose at DEPENDENCIES.md:159-162 still document the deleted as_voice_handler / record_voice_transcripts. AGENTS.md makes DEPENDENCIES.md authoritative and requires it to move with every pyproject.toml change.

8. Deleting tests/test_adapters_voice.py removed the only coverage of the still-shipping xr_conversation_memory function group.

   agent-sdk/xr-ai-nat/xr_ai_nat/functions/text_memory/functions.py:224
   Only the adapter half of that file's coverage died with its production code; recall_conversation time-window filtering and the generated request/result schema contract now have zero tests anywhere. Relocate the surviving cases.

Suggestions

1. Widening MessageMetadata.participant_id to str | None (runtime.py:62) changes a runtime-wide contract to serve one voice event: every subscriber now needs a null check and a mis-scoped publish fails late inside the subscriber rather than at the publish site. Consider keeping metadata scoping mandatory and modeling "global" on the event body instead.
2. Two behaviors lost their tests in the relocation: dropping typed text while the session is stopped (_runtime.py:320) and clearing _seen_output on participant-left (io.py:138, where a regression means a rejoining participant's first turn spuriously interrupts). Add both back.
3. Register the stream in _streams only after _enqueue_response succeeds (_runtime.py:256-266); a stopped-session RuntimeError currently strands the key. _streams is also unbounded: a producer that opens streams and never terminates them leaks a queue per response for the life of the process.
4. A mid-stream chunk with interrupt=True passes validation but is silently ignored (_runtime.py:267); reject it or apply it.
5. A text_transform returning empty or whitespace text fails UserQuery validation deep inside _run_query as a logged exception (_runtime.py:324-326); re-check emptiness after the transform. Note also that the default text_ignore_topics=() treats every inbound data-channel topic as a user query.
6. endpoint.on_data(self._on_data) is registered in lifespan but never unregistered in its finally (_runtime.py:178), so a re-entered lifespan or second agent over the same endpoint accumulates callbacks.
7. On token mismatch _consume_response continues draining the superseded iterator to exhaustion (io.py:270-274); return instead, or a long remote stream keeps burning tokens after supersession.
8. VoiceSession is public yet everything VoiceAgent consumes is private (_run, _enqueue_query, _enqueue_response) and the sample reaches through session.transport.endpoint; give the session a narrow intentional surface.
9. Minor hygiene: the topic is None guards in _publish_participant_left / _publish_interrupted are unreachable given lifespan's wiring (_runtime.py:283-285), _publish_input skips the _running_context() helper its siblings use, and the unbounded while waits at tests/test_voice_pipeline.py:1210 and tests/test_simple_vlm_example_worker.py:347 hang the suite instead of failing it on regression.

The direction here is right: one bidirectional VoiceAgent, typed topics for input and lifecycle, session internals kept private, and app.py reduced to pure composition, and most of the prior review round verifiably landed (chunk accumulation across iterator failures, producer-keyed streams, timestamp propagation, idempotent close, global scope without a synthetic participant, and FIFO order all have real regression assertions). But the incremental-output contract cannot be honored by the shipped producer under cancellation, and the output lock's scope turns backpressure into a global stall. The blockers above should be resolved before merge.

[nvddr]

Updated in 7e91705.

The active voice/runtime findings are addressed:

• Cancelled producers no longer publish orphan terminators; closed stream keys are tombstoned so late chunks cannot reopen interrupted playback.
• Stream sends run outside the global output lock, active and closed stream state is bounded, admission failures cannot strand keys, and mid-stream interrupts are rejected.
• Response iterators close on every exit, superseded iterators stop draining, and query-topic fan-out no longer shares the cancellable playback slot.
• Typed input is dropped while stopped, transformed text is revalidated, the output topic is ignored by default, participant output state is cleared on departure, and endpoint callbacks are unregistered.
• VoiceSession now exposes the narrow lifecycle/enqueue/endpoint surface used by VoiceAgent.
• The simple-VLM vision tool is constructed only after health readiness initializes the session endpoint, and sample cancellation closes its producer without sending a terminator.

I also fixed the current GitHub code-quality findings: direct transport type usage, explicit consumption of awaited results, idempotent callback removal without an empty except, and explicit async-iterator termination.

Per review scope, the legacy NAT packaging/conversation-memory compatibility items and the runtime-wide participant-metadata compatibility suggestion are unchanged.

Validation: 139 focused tests pass. The refreshed GitHub suite is green on Python 3.11 and 3.12, Ruff 0.15.16, both DCO checks, CodeQL, SPDX headers, dependency locks, and strict docs build.

@blongs-nv please re-review the updated head.

[blongs-nv]

Reviewed at 7e91705ebe0de6c759473276f7669611b7c9dc72 by a bot.

Blockers 1 through 6 from my previous review are resolved, and the fixes were verified in the code rather than by commit message: a cancelled producer no longer publishes its terminator and voice-closed keys are tombstoned, so barge-in no longer raises; send() now runs outside the output lock, with a real regression test that holds a genuinely blocked publisher at capacity 1 while unrelated output proceeds; failed and cancelled responses are closed on every exit path; post-eviction chunks are swallowed instead of re-opening a stream; query fan-out no longer occupies the cancellable response slot (with a delivery test); and the sample builds its vision tool lazily behind readiness, so hub sockets no longer open before probes. The suggestions on relocated coverage, mid-stream interrupt, on_data unregistration, the superseded-iterator drain, stream registration order, and the session's public surface were addressed as well.

On the deferred items: given xr-ai-nat's status as a migration-period compatibility surface and your scope call, I'm treating the [voice] packaging/DEPENDENCIES.md cleanup and the xr_conversation_memory test relocation as deferred to NAT's own cleanup rather than holding this PR for them. Please make sure the three recall_conversation cases aren't lost for good if that function group migrates rather than dies. The runtime-wide participant_id metadata suggestion is likewise noted as deferred.

Suggestions

1. await stream.aclose() is called twice in a row in the final-chunk path (_runtime.py:338-339); the second call is a no-op, drop it.

The fix round is thorough and well-tested, and the remaining items are deferred by agreement. This is ready to merge.