[agentserver][responses] fix replay stuck (#45930)

* use handler docorator and refining orchestration

* refining orchestration

* refined orchestration

* fixed conflict and handler registration

* fix replay

Author: lusu-msft

sdk/agentserver/azure-ai-agentserver-responses/azure/ai/agentserver/responses/hosting/_orchestrator.py

@@ -655,11 +655,55 @@ async def _finalize() -> None:
 
         # --- Fast path: no keep-alive ---
         if not self._runtime_options.sse_keep_alive_enabled:
+            if not (ctx.background and ctx.store):
+                # Simple fast path for non-background streaming.
+                try:
+                    async for event in self._process_handler_events(ctx, state, handler_iterator):
+                        yield encode_sse_payload(event["type"], event["payload"])
+                finally:
+                    await _finalize()
+                return
+
+            # Background+stream without keep-alive: run the handler as an independent
+            # asyncio.Task so that finalization (including subject.complete()) is
+            # guaranteed to run even when the original SSE connection is dropped before
+            # all events are delivered.  Without this, _live_stream can be abandoned
+            # mid-iteration by Starlette (the async-generator finalizer may not fire
+            # promptly), leaving GET-replay subscribers blocked on await q.get() forever.
+            _SENTINEL_BG = object()
+            bg_queue: asyncio.Queue[object] = asyncio.Queue()
+
+            async def _bg_producer() -> None:
+                try:
+                    async for event in self._process_handler_events(ctx, state, handler_iterator):
+                        await bg_queue.put(encode_sse_payload(event["type"], event["payload"]))
+                except Exception as exc:  # pylint: disable=broad-exception-caught
+                    state.captured_error = exc
+                finally:
+                    # Always finalize (includes subject.complete()) — this runs even if
+                    # the original POST SSE connection was dropped and _live_stream is
+                    # never properly closed by Starlette.
+                    await _finalize()
+                    await bg_queue.put(_SENTINEL_BG)
+
+            bg_task = asyncio.create_task(_bg_producer())
             try:
-                async for event in self._process_handler_events(ctx, state, handler_iterator):
-                    yield encode_sse_payload(event["type"], event["payload"])
+                while True:
+                    item = await bg_queue.get()
+                    if item is _SENTINEL_BG:
+                        break
+                    yield item  # type: ignore[misc]
+            except Exception:  # pylint: disable=broad-exception-caught
+                pass  # SSE connection dropped; bg_task continues independently
             finally:
-                await _finalize()
+                # Wait for the handler task so _finalize() has run before we exit.
+                # Do NOT cancel it — background+stream must reach a terminal state
+                # regardless of client connectivity.
+                if not bg_task.done():
+                    try:
+                        await bg_task
+                    except Exception:  # pylint: disable=broad-exception-caught
+                        pass
             return
 
         # --- Keep-alive path: merge handler events with periodic keep-alive comments ---

sdk/agentserver/azure-ai-agentserver-responses/samples/ConversationHistory/test.py

@@ -31,13 +31,6 @@ def _assert_ok(response: requests.Response) -> None:
         raise RuntimeError(f"HTTP request failed: {response.status_code} {response.text}") from exc
 
 
-def _ready() -> None:
-    _print_header("Ready")
-    response = requests.get(f"{BASE_URL}/ready", timeout=10)
-    _assert_ok(response)
-    _pretty_print(response.json())
-
-
 def _create(payload: dict[str, Any]) -> dict[str, Any]:
     response = requests.post(f"{BASE_URL}/responses", json=payload, timeout=10)
     _assert_ok(response)
@@ -104,7 +97,6 @@ def _turn_4_stream(previous_response_id: str) -> None:
 
 
 def main() -> None:
-    _ready()
     response_1_id = _turn_1()
     response_2_id = _turn_2(response_1_id)
     response_3_id = _turn_3(response_2_id)

sdk/agentserver/azure-ai-agentserver-responses/samples/FunctionCalling/test.py

@@ -32,13 +32,6 @@ def _assert_ok(response: requests.Response) -> None:
         raise RuntimeError(f"HTTP request failed: {response.status_code} {response.text}") from exc
 
 
-def _ready() -> None:
-    _print_header("Ready")
-    response = requests.get(f"{BASE_URL}/ready", timeout=10)
-    _assert_ok(response)
-    _pretty_print(response.json())
-
-
 def _turn_1_request_function_call() -> str:
     _print_header("Turn 1: Request function call")
     payload = {
@@ -103,7 +96,6 @@ def _turn_2_submit_function_output_streaming(call_id: str) -> None:
 
 
 def main() -> None:
-    _ready()
     call_id = _turn_1_request_function_call()
     _turn_2_submit_function_output(call_id)
     _turn_2_submit_function_output_streaming(call_id)

sdk/agentserver/azure-ai-agentserver-responses/samples/GetStarted/app.py

@@ -37,7 +37,7 @@ async def _events() -> AsyncIterable[dict[str, Any]]:
         text_content = message_item.add_text_content()
         yield text_content.emit_added()
         yield text_content.emit_delta("Hello from the Python GettingStarted sample!")
-        yield text_content.emit_done()
+        yield text_content.emit_done("Hello from the Python GettingStarted sample!")
         yield message_item.emit_content_done(text_content)
 
         yield message_item.emit_done()

sdk/agentserver/azure-ai-agentserver-responses/samples/GetStarted/test.py

@@ -32,13 +32,6 @@ def _assert_ok(response: requests.Response) -> None:
         raise RuntimeError(f"HTTP request failed: {response.status_code} {response.text}") from exc
 
 
-def _ready() -> None:
-    _print_header("Ready")
-    response = requests.get(f"{BASE_URL}/ready", timeout=10)
-    _assert_ok(response)
-    _pretty_print(response.json())
-
-
 def _default_mode() -> None:
     _print_header("Default mode (JSON)")
     payload = {"model": "gpt-4o-mini", "input": "hello"}
@@ -167,7 +160,6 @@ def _get_replay_mode() -> None:
 
 
 def main() -> None:
-    _ready()
     _default_mode()
     _stream_mode()
     _background_mode()

sdk/agentserver/azure-ai-agentserver-responses/samples/MultiOutput/test.py

@@ -31,13 +31,6 @@ def _assert_ok(response: requests.Response) -> None:
         raise RuntimeError(f"HTTP request failed: {response.status_code} {response.text}") from exc
 
 
-def _ready() -> None:
-    _print_header("Ready")
-    response = requests.get(f"{BASE_URL}/ready", timeout=10)
-    _assert_ok(response)
-    _pretty_print(response.json())
-
-
 def _default_mode() -> None:
     _print_header("Default mode (JSON) - reasoning + message")
     payload = {"model": "test"}
@@ -62,7 +55,6 @@ def _stream_mode() -> None:
 
 
 def main() -> None:
-    _ready()
     _default_mode()
     _stream_mode()