From 2337af01dd25fd31fcacec26dcc09b7c9ce10cd8 Mon Sep 17 00:00:00 2001
From: Yueqian Lin <linyueqian@outlook.com>
Date: Sun, 4 Oct 2026 16:03:19 -0700
Subject: [PATCH] fix(tests): wire output buffers into duplex model harnesses

Signed-off-by: Yueqian Lin <linyueqian@outlook.com>
---
 .../test_session_runner_nemotron_voicechat.py | 23 ++++++++++++++++---
 .../duplex/test_session_runner_personaplex.py | 23 ++++++++++++++++---
 2 files changed, 40 insertions(+), 6 deletions(-)

diff --git a/tests/engine/duplex/test_session_runner_nemotron_voicechat.py b/tests/engine/duplex/test_session_runner_nemotron_voicechat.py
index 254434fa5d2..24fc92e730f 100644
--- a/tests/engine/duplex/test_session_runner_nemotron_voicechat.py
+++ b/tests/engine/duplex/test_session_runner_nemotron_voicechat.py
@@ -21,6 +21,7 @@
 from vllm_omni.config.stage_config import DuplexSessionRuntimeConfig
 from vllm_omni.engine.duplex import commands
 from vllm_omni.engine.duplex.config import DuplexSessionConfig
+from vllm_omni.engine.duplex.delivery import DuplexOutputBuffer
 from vllm_omni.engine.duplex.messages import OpenDuplexSessionMessage
 from vllm_omni.engine.duplex.plugin import load_duplex_plugin
 from vllm_omni.engine.duplex.session.manager import DuplexSessionManager
@@ -57,20 +58,36 @@ async def harness(monkeypatch):
     port = RecordingStagePort(stage_count=3)
     output: asyncio.Queue = asyncio.Queue()
     results: asyncio.Queue = asyncio.Queue()
+    limits = DuplexSessionRuntimeConfig()
+    output_buffer = DuplexOutputBuffer(
+        max_bytes=limits.max_pending_output_bytes_per_session,
+        max_events=limits.max_pending_output_events_per_session,
+    )
     manager = DuplexSessionManager(
         plugin=plugin,
         stage_port=port,
         output_sink=output,
         result_sink=results,
-        runtime_config=DuplexSessionRuntimeConfig(),
+        runtime_config=limits,
         model_config=SimpleNamespace(hf_config=SimpleNamespace(stt_cfg={}), max_model_len=8192),
     )
     try:
         config = DuplexSessionConfig(model="nemotron", instructions="hi", extra_body={"auto_response": True})
-        await manager.handle(OpenDuplexSessionMessage(control_id="open", session_id=SESSION_ID, session_config=config))
+        await manager.handle(
+            OpenDuplexSessionMessage(
+                control_id="open", session_id=SESSION_ID, session_config=config, output_buffer=output_buffer
+            )
+        )
         result = await asyncio.wait_for(results.get(), timeout=2)
         assert result.ok, result
-        h = Harness(manager, port, output, results, manager.runners[SESSION_ID])
+        h = Harness(
+            manager=manager,
+            port=port,
+            output=output,
+            output_buffer=output_buffer,
+            results=results,
+            runner=manager.runners[SESSION_ID],
+        )
         await h.settle()
         yield h
     finally:
diff --git a/tests/engine/duplex/test_session_runner_personaplex.py b/tests/engine/duplex/test_session_runner_personaplex.py
index 6d835803c02..538737626da 100644
--- a/tests/engine/duplex/test_session_runner_personaplex.py
+++ b/tests/engine/duplex/test_session_runner_personaplex.py
@@ -34,6 +34,7 @@
 from vllm_omni.config.stage_config import DuplexSessionRuntimeConfig
 from vllm_omni.engine.duplex import commands
 from vllm_omni.engine.duplex.config import DuplexSessionConfig
+from vllm_omni.engine.duplex.delivery import DuplexOutputBuffer
 from vllm_omni.engine.duplex.messages import DuplexControlResultMessage, OpenDuplexSessionMessage
 from vllm_omni.engine.duplex.session.manager import DuplexSessionManager
 from vllm_omni.model_executor.models.personaplex.duplex import stage0
@@ -55,12 +56,17 @@ async def open_personaplex_harness(*, extra_body: dict[str, object] | None = Non
     port = RecordingStagePort(stage_count=2)
     output: asyncio.Queue[Any] = asyncio.Queue()
     results: asyncio.Queue[Any] = asyncio.Queue()
+    limits = DuplexSessionRuntimeConfig()
+    output_buffer = DuplexOutputBuffer(
+        max_bytes=limits.max_pending_output_bytes_per_session,
+        max_events=limits.max_pending_output_events_per_session,
+    )
     manager = DuplexSessionManager(
         plugin=plugin,
         stage_port=port,
         output_sink=output,
         result_sink=results,
-        runtime_config=DuplexSessionRuntimeConfig(),
+        runtime_config=limits,
         model_config=SimpleNamespace(model="/models/personaplex-7b-v1"),
     )
     config = DuplexSessionConfig(
@@ -70,10 +76,21 @@ async def open_personaplex_harness(*, extra_body: dict[str, object] | None = Non
         voice="NATF2.pt",
         extra_body=dict(extra_body or {}),
     )
-    await manager.handle(OpenDuplexSessionMessage(control_id="c-open", session_id=SESSION_ID, session_config=config))
+    await manager.handle(
+        OpenDuplexSessionMessage(
+            control_id="c-open", session_id=SESSION_ID, session_config=config, output_buffer=output_buffer
+        )
+    )
     result = await asyncio.wait_for(results.get(), timeout=2.0)
     assert isinstance(result, DuplexControlResultMessage) and result.ok, result
-    harness = Harness(manager=manager, port=port, output=output, results=results, runner=manager.runners[SESSION_ID])
+    harness = Harness(
+        manager=manager,
+        port=port,
+        output=output,
+        output_buffer=output_buffer,
+        results=results,
+        runner=manager.runners[SESSION_ID],
+    )
     await harness.settle()
     return harness