knit (knitvoger) · Nov 13, 2026 · Nov 10, 2026 · Nov 10, 2026 · Nov 11, 2026 · Nov 11, 2026
@@ -0,0 +1,27 @@
+# CE needs a dedicated WebSocket/audio client. The shared HTTP deployment runner
+# skips this sample; planning as invocations validates raw assertions only.
+version: 1
+sample:
+  owner: wujin
+  experiences: [azd]
+tests:
+  - name: foreground-direct-response
+    turns:
+      - input: Say exactly "Ready for field operations." Do not start background work.
+        assertions:
+          - source: raw
+            type: contains
+            value: Ready for field operations.
+          - source: raw
+            type: regex
+            value: '"type"\s*:\s*"response\.done"'
+  - name: background-tool-result
+    turns:
+      - input: Find the fiber panel specifications for Quincy North using search_site_specs.
+        assertions:
+          - source: raw
+            type: regex
+            value: '(?i)OS2|OM4|Row C'
+          - source: raw
+            type: regex
+            value: '"type"\s*:\s*"response\.done"'
@@ -24,6 +24,7 @@ samples/csharp/hosted-agents/bring-your-own/conversation_engine/adapter_mode
 samples/csharp/hosted-agents/bring-your-own/conversation_engine/basic
 samples/python/hosted-agents/bring-your-own/conversation_engine/adapter_mode
 samples/python/hosted-agents/bring-your-own/conversation_engine/basic
+samples/python/hosted-agents/bring-your-own/conversation_engine/foreground_background
 samples/python/hosted-agents/bring-your-own/conversation_engine/langchain
 samples/python/hosted-agents/bring-your-own/conversation_engine/langgraph
 

@@ -288,7 +288,9 @@ path. After deploying, verify readiness and a spoken echo through the declared V
 - The story and its word sequence are directly in `OnUserMessageAsync`; no model history is added.
 - Echo is scheduled as cancellable work so the receive loop remains available for control events.
 - Barge-in and cancellation synchronize caller-heard text and stop pending output.
-- Playback state is bounded to 64 known responses and cleared on session termination.
+- Playback state is bounded to 64 known responses and cleared on session termination. Each
+  heard prefix retains at most the 512 KiB UTF-8 output budget, without splitting Unicode
+  characters; the 8,000-character user-input limit does not truncate valid reply prefixes.
 - Custom spans contain identifiers, durations, and status, but no prompt or response content.
 - A deployed two-turn audio smoke passed with public SDK packages; see the
   [validation status](../README.md#validation-status). Perform deployment-specific validation

@@ -39,6 +39,7 @@ internal sealed class VoiceConnectionState
     internal const int MaximumTerminalSends = 8;
     internal const int MaximumTrackedInputItems = 4096;
     internal const int MaximumTrackedPlaybackResponses = 64;
+    internal const int MaximumOutputBytes = 512 * 1024;
 
     private readonly object _sync = new();
     private readonly HashSet<string> _seenInputItemDigests = new(StringComparer.Ordinal);
@@ -287,13 +288,14 @@ internal void RecordHeardText(string responseId, string heardText)
             }
             TrackResponse(responseId);
             var length = 0;
-            var characters = 0;
+            var bytes = 0;
             foreach (var rune in heardText.EnumerateRunes())
             {
-                if (characters++ == 8_000)
+                if (bytes + rune.Utf8SequenceLength > MaximumOutputBytes)
                 {
                     break;
                 }
+                bytes += rune.Utf8SequenceLength;
                 length += rune.Utf16SequenceLength;
             }
             _heardTextByResponse[responseId] = heardText[..length];

@@ -337,7 +337,7 @@ await session.SendAsync(new VoiceResponseCreatedMessage(
                 ThrowIfStopped(state, operation, cancellationToken);
                 var chunk = iterator.Current;
                 if (++outputChunks > 4096 ||
-                    (outputBytes += Encoding.UTF8.GetByteCount(chunk)) > 512 * 1024)
+                    (outputBytes += Encoding.UTF8.GetByteCount(chunk)) > VoiceConnectionState.MaximumOutputBytes)
                 {
                     throw new InvalidOperationException("Response exceeds its output budget.");
                 }
@@ -418,7 +418,7 @@ await session.SendAsync(new VoiceResponseCreatedMessage(
             await Task.Yield();
             await created;
             ThrowIfStopped(state, operation, cancellationToken);
-            if (Encoding.UTF8.GetByteCount(text) > 512 * 1024)
+            if (Encoding.UTF8.GetByteCount(text) > VoiceConnectionState.MaximumOutputBytes)
             {
                 throw new InvalidOperationException("Response text exceeds its output budget.");
             }

@@ -1305,7 +1305,7 @@ await agent.SessionEndAsync(session,
 
     [TestCase(8_000)]
     [TestCase(8_001)]
-    public void HeardTextBudget_MatchesAcceptedUnicodeScalars(int count)
+    public void HeardTextBudget_DoesNotApplyInputLimit(int count)
     {
         var state = new VoiceConnectionState();
         state.TryBeginStart();
@@ -1314,10 +1314,37 @@ public void HeardTextBudget_MatchesAcceptedUnicodeScalars(int count)
         Assert.That(state.TryReserveOperation(operation), Is.True);
         state.RecordHeardText("r_unicode", string.Concat(Enumerable.Repeat("👋", count)));
         Assert.That(state.TryGetHeardText("r_unicode", out var heard), Is.True);
-        Assert.That(heard, Is.EqualTo(string.Concat(Enumerable.Repeat("👋", 8_000))));
+        Assert.That(heard, Is.EqualTo(string.Concat(Enumerable.Repeat("👋", count))));
         Assert.That(state.TryRemoveOperation("r_unicode", out _), Is.True);
     }
 
+    [TestCase("x", 524_288)]
+    [TestCase("👋", 131_072)]
+    [TestCase("界", 174_762)]
+    public void HeardTextBudget_PreservesWholeRunesWithinOutputBytes(string unit, int count)
+    {
+        var state = new VoiceConnectionState();
+        state.TryBeginStart();
+        state.Activate();
+        using var operation = new ResponseOperation("r_budget", "in_budget");
+        Assert.That(state.TryReserveOperation(operation), Is.True);
+        var prefix = string.Concat(Enumerable.Repeat(unit, count));
+        state.RecordHeardText("r_budget", prefix + unit);
+        Assert.That(state.TryGetHeardText("r_budget", out var heard), Is.True);
+        Assert.That(heard.Length, Is.EqualTo(prefix.Length));
+        Assert.That(heard, Is.EqualTo(prefix));
+        Assert.That(System.Text.Encoding.UTF8.GetByteCount(heard), Is.LessThanOrEqualTo(512 * 1024));
+        state.RecordHeardText("r_unknown", prefix);
+        Assert.That(state.TryGetHeardText("r_unknown", out _), Is.False);
+        state.RecordHeardText("r_budget", string.Empty);
+        Assert.That(state.TryGetHeardText("r_budget", out heard), Is.True);
+        Assert.That(heard, Is.Empty);
+        state.RecordHeardText("r_budget", "retry");
+        Assert.That(state.TryGetHeardText("r_budget", out heard), Is.True);
+        Assert.That(heard, Is.EqualTo("retry"));
+        Assert.That(state.TryRemoveOperation("r_budget", out _), Is.True);
+    }
+
     private static TestableVoiceAgent CreateAgent() =>
         new(NullLogger<VoiceAgent>.Instance);
 

@@ -30,6 +30,7 @@ require no model call.
 | --- | --- | --- |
 | Python | [Voice Live — Adapter Mode](adapter_mode/) | Recommended for reusing an invocation handler: a CE-only Agent calls a streaming HTTP-shaped Echo handler in process. |
 | Python | [Voice Live Bridge — Basic Python](basic/) | A framework-free Python hosted agent using `azure-ai-agentserver-invocations`. |
+| Python | [Foreground + Background Conversation Engine](foreground_background/) | Model-backed foreground Router and background tool Worker; CE counterpart to the retained Responses VoiceLive sample. |
 | Python | [Voice Live Bridge — LangChain Python](langchain/) | Model-free async story stream; legacy directory/Agent names retained. |
 | Python | [Voice Live Bridge — LangGraph Python](langgraph/) | Model-free async story stream; legacy directory/Agent names retained. |
 | C# | [Voice Live — Adapter Mode](../../../../csharp/hosted-agents/bring-your-own/conversation_engine/adapter_mode/) | The equivalent in-process adapter and streaming Echo handler for .NET. |
@@ -38,6 +39,11 @@ require no model call.
 All implementations demonstrate the same Bridge Protocol behavior so you can start with the
 language that matches your application.
 
+The foreground/background sample differs from the model-free story/Echo samples: it retains
+its Foundry model deployment, Router meta-tools, and mock domain tools. Worker tasks and bounded
+conversation history are isolated per physical connection. See its README for interruption,
+delivery, shutdown, and deployment prerequisites.
+
 ## Choose an integration style
 
 - **Adapter mode:** start here to reuse HTTP invocation logic without implementing Bridge
@@ -49,6 +55,7 @@ language that matches your application.
 - **Native CE mode:** use `basic`, `langchain`, or `langgraph` when you want to implement typed
   Voice callbacks directly and manage the application-owned protocol state. These existing
   samples retain their fixed story behavior; the legacy variant names do not add framework clients.
+  Use `foreground_background` for the model-backed Router/Worker version of native CE.
 
 Adapter mode does not take over a replacement handler's business history or roll back side effects.
 Its Echo is stateless; persistence and any reconciliation of unheard text remain business concerns.
@@ -61,7 +68,7 @@ Caller audio
     → Voice Live (STT, VAD, TTS, and barge-in)
     → Bridge Protocol 1.0 over voice_conversation_engine
     → hosted text agent
-    → Echo text (adapter mode) or a thirty-word story (native CE) streamed back to Voice Live
+    → Echo/story output, or model-backed Router/Worker output, streamed back to Voice Live
 ```
 
 The caller connects to a Foundry **voice agent**, not directly to the hosted text agent. The voice
@@ -86,6 +93,8 @@ protocol and is versioned independently from the Hosted Agent WebSocket ingress.
 Adapter mode echoes each valid input verbatim in streaming chunks, preserving whitespace and
 Unicode. Native CE samples stream the same thirty-word story for every valid input.
 Slash-prefixed text such as `/help` or `/end` is ordinary text, not a command, in both styles.
+The foreground/background variant instead routes the caller's text to its model-backed business
+workflow; it does not produce the fixed story or Echo reply.
 
 The samples demonstrate the Protocol `1.0` customer surface:
 

@@ -386,6 +386,8 @@ async def wait_for_detached_tasks(self) -> None:
                 *(asyncio.shield(task) for task in tuple(self._detached_tasks)),
                 return_exceptions=True,
             )
+            # gather() can finish synchronously; let ownership-release callbacks run.
+            await asyncio.sleep(0)
 
     def schedule_none(
         self, session: Session, state: SessionState, input_item_id: str,

@@ -0,0 +1,17 @@
+"""Drain must progress when all retained tasks finished before the join."""
+
+import asyncio
+
+import pytest
+
+from adapter.response_coordinator import ResponseCoordinator
+
+
+@pytest.mark.asyncio
+async def test_drain_yields_to_completed_task_release_callback() -> None:
+    coordinator = ResponseCoordinator()
+    task = asyncio.create_task(asyncio.sleep(0))
+    await task
+    coordinator._retain_task(task)
+    await asyncio.wait_for(coordinator.wait_for_detached_tasks(), 1)
+    assert not coordinator._detached_tasks
@@ -260,7 +260,9 @@ After deployment, verify readiness and a spoken echo through the Voice wrapper.
   `state.py` owns connection state and response/input-ID correlation.
 - `user.barge_in` and `response.cancelled` synchronize caller-heard text and cancel pending output.
   `response.timeout` cancels work selected by response ID or input IDs.
-- Playback state is bounded to 64 known responses with up to 8,000 characters of heard text each.
+- Playback state is bounded to 64 known responses, each with at most 512 KiB of UTF-8 heard text,
+  matching the response output budget rather than the 8,000-character input limit. Larger values
+  are cut only at complete Unicode characters; string/object overhead is additional to this budget.
   Session termination clears it together with pending work.
 - Response operations contain protocol bookkeeping, not `prompt` or `use_model` flags.
 - Runtime dependencies are locked in `uv.lock`; OpenAI, LangChain, and LangGraph clients

@@ -31,6 +31,7 @@
 
 from response_telemetry import ResponseTelemetry
 from state import (
+    MAX_OUTPUT_BYTES,
     MAX_PENDING_PROACTIVE,
     MAX_TERMINAL_SENDS,
     ApplicationTurn,
@@ -42,7 +43,6 @@
 )
 
 logger = logging.getLogger("voice_live_bridge")
-MAX_OUTPUT_BYTES = 512 * 1024
 MAX_OUTPUT_CHUNKS = 4096
 
 OutputChunk = str | Mapping[str, Any]
@@ -386,6 +386,8 @@ async def wait_for_detached_tasks(self) -> None:
                 *(asyncio.shield(task) for task in tuple(self._detached_tasks)),
                 return_exceptions=True,
             )
+            # gather() can finish synchronously; let ownership-release callbacks run.
+            await asyncio.sleep(0)
 
     def schedule_none(
         self, session: Session, state: SessionState, input_item_id: str,

@@ -18,6 +18,7 @@
 
 MAX_TRACKED_PLAYBACK_RESPONSES = 64
 MAX_MESSAGE_CHARACTERS = 8_000
+MAX_OUTPUT_BYTES = 512 * 1024
 MAX_ACTIVE_RESPONSES = 8
 MAX_TERMINAL_SENDS = 8
 MAX_PENDING_PROACTIVE = 8
@@ -157,8 +158,12 @@ def record_heard_text(self, response_id: str, heard_text: str) -> None:
         """Synchronize only responses this connection has actually tracked."""
         if response_id not in self.heard_text_by_response and response_id not in self.responses:
             return
+        # Playback describes response output, not user input. Bound both the UTF-8
+        # payload and temporary allocation; malformed escapes must not block cancellation.
+        encoded = heard_text[:MAX_OUTPUT_BYTES].encode("utf-8", errors="replace")
+        bounded = encoded[:MAX_OUTPUT_BYTES].decode("utf-8", errors="ignore")
         self.track_response(response_id)
-        self.heard_text_by_response[response_id] = heard_text[:MAX_MESSAGE_CHARACTERS]
+        self.heard_text_by_response[response_id] = bounded
 
     @property
     def terminating(self) -> bool:

@@ -0,0 +1,79 @@
+"""Heard prefixes use the response byte budget, not the user-input limit."""
+
+import pytest
+
+from response_coordinator import MAX_OUTPUT_BYTES
+from state import MAX_MESSAGE_CHARACTERS, MAX_TRACKED_PLAYBACK_RESPONSES, SessionState
+
+
+@pytest.mark.parametrize(
+    "unit,count,tail,expected_count",
+    [
+        ("x", 7999, "", 7999),
+        ("x", 8000, "", 8000),
+        ("x", 8001, "", 8001),
+        ("x", 8011, "", 8011),
+        ("👋", 8001, "", 8001),
+        ("x", 524288, "", 524288),
+        ("x", 524289, "", 524288),
+        ("👋", 131072, "", 131072),
+        ("👋", 131073, "", 131072),
+        ("界", 174763, "", 174762),
+        ("x", 524287, "👋", 524287),
+    ],
+    ids=[
+        "below-input-limit",
+        "at-input-limit",
+        "above-input-limit",
+        "result-plus-ack",
+        "long-unicode",
+        "exact-byte-budget",
+        "over-byte-budget",
+        "unicode-exact-budget",
+        "unicode-over-budget",
+        "three-byte-boundary",
+        "partial-rune-must-not-be-retained",
+    ],
+)
+def test_heard_prefix_is_complete_within_output_budget_and_bounded_above_it(
+    unit: str,
+    count: int,
+    tail: str,
+    expected_count: int,
+) -> None:
+    state = SessionState()
+    state.track_response("r_known")
+    heard = unit * count + tail
+    state.record_heard_text("r_known", heard)
+    stored = state.heard_text_by_response["r_known"]
+    assert len(stored) == len(unit * expected_count)
+    assert stored == unit * expected_count
+    assert heard.startswith(stored)
+    assert len(stored.encode("utf-8")) <= 512 * 1024
+    if len(heard.encode("utf-8")) <= 512 * 1024:
+        assert stored == heard
+    state.record_heard_text("r_known", "")
+    assert state.heard_text_by_response["r_known"] == ""
+    state.record_heard_text("r_known", "retry prefix")
+    assert state.heard_text_by_response == {"r_known": "retry prefix"}
+
+
+def test_unknown_and_evicted_responses_cannot_repopulate_playback_cache() -> None:
+    state = SessionState()
+    state.record_heard_text("r_unknown", "x" * 8011)
+    assert not state.heard_text_by_response
+    for index in range(MAX_TRACKED_PLAYBACK_RESPONSES + 1):
+        state.track_response(f"r_{index}")
+    state.record_heard_text("r_0", "evicted prefix")
+    assert "r_0" not in state.heard_text_by_response
+    assert len(state.heard_text_by_response) == 64
+    state.record_heard_text("r_64", "x" * 8011)
+    assert len(state.heard_text_by_response["r_64"]) == 8011
+    assert sum(len(text.encode("utf-8")) for text in state.heard_text_by_response.values()) <= (
+        64 * 512 * 1024
+    )
+
+
+def test_response_budget_does_not_expand_user_input_or_output_limits() -> None:
+    assert MAX_MESSAGE_CHARACTERS == 8000
+    assert MAX_OUTPUT_BYTES == 512 * 1024
@@ -0,0 +1,17 @@
+"""Drain must progress when all retained tasks finished before the join."""
+
+import asyncio
+
+import pytest
+
+from response_coordinator import ResponseCoordinator
+
+
+@pytest.mark.asyncio
+async def test_drain_yields_to_completed_task_release_callback() -> None:
+    coordinator = ResponseCoordinator()
+    task = asyncio.create_task(asyncio.sleep(0))
+    await task
+    coordinator._retain_task(task)
+    await asyncio.wait_for(coordinator.wait_for_detached_tasks(), 1)
+    assert not coordinator._detached_tasks