feat(openai): emit AG-UI text and tool events

packages/client/src/launchdarkly_ai_server/responses.py

@@ -0,0 +1,57 @@
+"""Map OpenAI Responses stream events onto the AG-UI text and tool subset.
+
+Both OpenAI handler packages call :func:`map_responses_event`. They only
+offer function tools, so hosted and MCP item types are ignored until those
+tools are wired.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from .agui import TurnEmitter
+
+
+def _call_id(event: Any, item_to_call: dict[str, str]) -> str:
+    direct = getattr(event, "call_id", None)
+    if isinstance(direct, str) and direct:
+        return direct
+    item_id = getattr(event, "item_id", None)
+    if isinstance(item_id, str) and item_id in item_to_call:
+        return item_to_call[item_id]
+    return item_id if isinstance(item_id, str) else ""
+
+
+def map_responses_event(
+    event: Any, emitter: TurnEmitter, item_to_call: dict[str, str]
+) -> list[dict[str, Any]]:
+    """Map one Responses event. Unknown events, including hosted tools, are ignored."""
+    kind = getattr(event, "type", None)
+    if kind == "response.created":
+        return emitter.end_text()
+    if kind == "response.output_text.delta":
+        delta = getattr(event, "delta", "")
+        if not isinstance(delta, str) or not delta:
+            return []
+        return emitter.text(delta)
+    if kind == "response.output_item.added":
+        item = getattr(event, "item", None)
+        if getattr(item, "type", None) != "function_call":
+            return []
+        call_id = getattr(item, "call_id", None) or getattr(item, "id", None) or ""
+        name = getattr(item, "name", None) or ""
+        item_id = getattr(item, "id", None)
+        if isinstance(item_id, str) and isinstance(call_id, str) and call_id:
+            item_to_call[item_id] = call_id
+        return emitter.tool_start(str(call_id), str(name))
+    if kind == "response.function_call_arguments.delta":
+        delta = getattr(event, "delta", "")
+        if not isinstance(delta, str):
+            return []
+        return emitter.tool_args(_call_id(event, item_to_call), delta)
+    if kind == "response.function_call_arguments.done":
+        return emitter.tool_end(
+            _call_id(event, item_to_call),
+            arguments=getattr(event, "arguments", None),
+        )
+    return []

packages/client/tests/test_agui.py

@@ -149,6 +149,16 @@ def test_fail_open_masks_a_call_that_never_finished(self) -> None:
         assert events[-1]["content"] == ""
         assert events[-1]["metadata"]["launchdarkly"]["toolFailed"] is True
 
+    def test_tool_end_extends_a_streamed_argument_prefix(self) -> None:
+        emitter = TurnEmitter(capture_content=True)
+        events = emitter.tool_start("a", "search")
+        events.extend(emitter.tool_args("a", '{"q":'))
+        events.extend(emitter.tool_end("a", arguments={"q": "x"}))
+        deltas = [
+            event["delta"] for event in events if event["type"] == "TOOL_CALL_ARGS"
+        ]
+        assert "".join(deltas) == '{"q":"x"}'
+
     def test_a_second_start_for_the_same_id_is_dropped(self) -> None:
         emitter = TurnEmitter(capture_content=True)
         assert emitter.tool_start("a", "search")

packages/openai-agents/src/launchdarkly_ai_openai_agents/handler.py

@@ -45,7 +45,9 @@
     set_tool_call_content_attributes,
     text_message,
 )
+from launchdarkly_ai_server.agui import TurnEmitter
 from launchdarkly_ai_server.lifecycle import effective_capture_content
+from launchdarkly_ai_server.responses import map_responses_event
 
 from ._version import PACKAGE_NAME, __version__
 from .spans import (
@@ -250,10 +252,16 @@ def __init__(
         self.run_usage = run_usage
         self.open_model_span: Any = None
         self.open_tool_spans: dict[str, Any] = {}
+        self.emitter = TurnEmitter(capture_content=capture_content)
+        self.pending: list[dict[str, Any]] = []
+        self.item_to_call: dict[str, str] = {}
 
     async def on_llm_start(
         self, context: Any, agent: Any, system_prompt: str | None, input_items: Any
     ) -> None:
+        # Text is closed from response.created, in the order the consumer
+        # reads raw events. Closing here runs ahead of unread deltas and
+        # splits one turn across two messages.
         span = start_model_span(self.config, self.parent)
         self.open_model_span = span
         if self.capture_content:
@@ -320,6 +328,16 @@ async def on_tool_start(self, context: Any, agent: Any, tool: Any) -> None:
         # abandon_open_spans both walk this dict, so a span missing from it never ends and never
         # exports.
         self.open_tool_spans[str(call_id)] = span
+        self.pending.extend(self.emitter.tool_start(str(call_id), str(name)))
+        # The Runner can invoke this hook from its background task while
+        # argument deltas are still unread. Closing now would drop the rest
+        # and skip the full argument string, because a prefix is already
+        # recorded. arguments.done closes a call that is still streaming.
+        if not self.emitter.has_argument_deltas(str(call_id)):
+            tool_args = getattr(context, "tool_arguments", None)
+            self.pending.extend(
+                self.emitter.tool_end(str(call_id), arguments=tool_args)
+            )
         if self.capture_content:
             # `tool_arguments` is the raw JSON args string the model produced. Parsed here rather
             # than passed through, per TELEMETRY-CONTRACT.md section 12: arguments hold the object
@@ -336,7 +354,19 @@ async def on_tool_start(self, context: Any, agent: Any, tool: Any) -> None:
     async def on_tool_end(
         self, context: Any, agent: Any, tool: Any, result: Any
     ) -> None:
-        span = self.open_tool_spans.pop(self._tool_span_key(context, tool), None)
+        call_id = self._tool_span_key(context, tool)
+        # The hook can run before unread argument deltas. Close with the full
+        # argument string first, so tool_result does not seal a prefix.
+        # Text stays open: this runs before unread text deltas, and closing
+        # here would start a second message for the rest of the turn.
+        self.pending.extend(
+            self.emitter.tool_end(
+                str(call_id),
+                arguments=getattr(context, "tool_arguments", None),
+            )
+        )
+        self.pending.extend(self.emitter.tool_result(str(call_id), str(result)))
+        span = self.open_tool_spans.pop(call_id, None)
         if span is None:
             return
         # Once popped, ending it is this method's job alone: nothing else knows the span exists. A
@@ -383,6 +413,26 @@ def abandon_open_spans(self, ended: set[int], cancelled: bool = False) -> None:
             self.open_model_span = None
 
 
+def _take_finished_tools(hooks: _SpanningHooks) -> list[dict[str, Any]]:
+    """Return tool events recorded by the hook, text closed before each result.
+
+    The hook runs before unread deltas from the same model call. The caller
+    applies those deltas first. Tool results then follow ``TEXT_MESSAGE_END``
+    so the client does not attach them to the next message.
+    """
+    if not hooks.pending:
+        return []
+    earlier: list[dict[str, Any]] = []
+    results: list[dict[str, Any]] = []
+    for event in hooks.pending:
+        if event.get("type") == "TOOL_CALL_RESULT":
+            results.append(event)
+        else:
+            earlier.append(event)
+    hooks.pending.clear()
+    return [*earlier, *hooks.emitter.end_text(), *results]
+
+
 def _write_failed_run_usage(
     span: Any,
     config: AiConfigRep,
@@ -654,16 +704,28 @@ async def _stream_gen(
         full_output = ""
 
         async for event in streamed.stream_events():
-            if event.type == "raw_response_event":
-                raw = getattr(event, "data", None)
-                if (
-                    raw is not None
-                    and getattr(raw, "type", None) == "response.output_text.delta"
-                ):
-                    delta = getattr(raw, "delta", "")
-                    if isinstance(delta, str) and delta:
-                        yield {"type": "chunk", "text": delta}
-                        full_output += delta
+            raw = (
+                getattr(event, "data", None)
+                if event.type == "raw_response_event"
+                else None
+            )
+            # The next model call is queued only after this turn's tools have
+            # run. Unread deltas from this turn are still ahead of that event,
+            # so the tool result waits until they have been applied.
+            if getattr(raw, "type", None) == "response.created":
+                for rich in _take_finished_tools(hooks):
+                    yield rich
+            if raw is not None:
+                for rich in map_responses_event(raw, hooks.emitter, hooks.item_to_call):
+                    if rich.get("type") == "TEXT_MESSAGE_CONTENT":
+                        full_output += rich.get("delta") or ""
+                    yield rich
+        for rich in _take_finished_tools(hooks):
+            yield rich
+        for rich in hooks.emitter.end_text():
+            yield rich
+        for rich in hooks.emitter.fail_open():
+            yield rich
 
         stream_completed = True
         final_output = streamed.final_output
@@ -697,6 +759,12 @@ async def _stream_gen(
         hooks.close_open_spans(exc)
         _write_failed_run_usage(span, config, exc, run_usage)
         fail_span(span, exc, ended)
+        for rich in _take_finished_tools(hooks):
+            yield rich
+        for rich in hooks.emitter.end_text():
+            yield rich
+        for rich in hooks.emitter.fail_open():
+            yield rich
         raise
     finally:
         # A no-op on the success and failure paths: both already ended their spans through