Enhance streaming and parsing functionality: add normalization for tool fragments, improve response text collection, and implement comprehensive unit tests for nested response handling.

Author: btfranklin

src/compendiumscribe/research/execution/stream_events.py

@@ -25,6 +25,93 @@
 ]
 
 
+def _coerce_action_payload(payload: Any) -> dict[str, Any]:
+    """Normalise tool arguments into a mapping suitable for summaries."""
+
+    if isinstance(payload, dict):
+        return payload
+
+    if isinstance(payload, str):
+        try:
+            parsed = json.loads(payload)
+        except json.JSONDecodeError:
+            return {"input": payload}
+        if isinstance(parsed, dict):
+            return parsed
+        return {"value": parsed}
+
+    if payload is None:
+        return {}
+
+    if isinstance(payload, (list, tuple, set)):
+        return {"items": [item for item in payload]}
+
+    return {"value": payload}
+
+
+def normalize_tool_fragment(fragment: Any) -> dict[str, Any] | None:
+    """Coerce streaming tool fragments into the tool-call snapshot format."""
+
+    if not isinstance(fragment, dict):
+        return None
+
+    candidate = dict(fragment)
+
+    embedded = candidate.pop("tool_call", None)
+    if isinstance(embedded, dict):
+        candidate = {**embedded, **candidate}
+
+    # Promote common identifier fields for streaming events.
+    identifier = first_non_empty(
+        coerce_optional_string(candidate.get(key))
+        for key in ("id", "tool_call_id", "call_id")
+    )
+    if identifier:
+        candidate["id"] = identifier
+
+    raw_type = coerce_optional_string(candidate.get("type"))
+    tool_name = coerce_optional_string(candidate.get("tool_name"))
+    tool_type = coerce_optional_string(candidate.get("tool_type"))
+
+    normalized_type: str | None = None
+    if raw_type and raw_type.endswith("_call"):
+        normalized_type = raw_type
+    elif tool_name:
+        normalized_type = f"{tool_name.replace('.', '_')}_call"
+    elif tool_type:
+        normalized_type = f"{tool_type.replace('.', '_')}_call"
+    elif raw_type and raw_type.endswith("_delta"):
+        normalized_type = f"{raw_type[: -len('_delta')]}_call"
+
+    if normalized_type:
+        candidate["type"] = normalized_type
+
+    status = first_non_empty(
+        coerce_optional_string(candidate.get(key))
+        for key in ("status", "state")
+    )
+    if status:
+        candidate["status"] = status
+
+    if "arguments" in candidate and "action" not in candidate:
+        candidate["action"] = _coerce_action_payload(
+            candidate.pop("arguments")
+        )
+
+    if "call_arguments" in candidate and "action" not in candidate:
+        candidate["action"] = _coerce_action_payload(
+            candidate.pop("call_arguments")
+        )
+
+    if "input" in candidate and "action" not in candidate:
+        candidate["action"] = _coerce_action_payload(candidate.pop("input"))
+
+    if not coerce_optional_string(candidate.get("type")):
+        return None
+
+    return candidate
+
+
 def emit_trace_updates_from_item(
     item: Any,
     *,
@@ -58,8 +145,12 @@ def accumulate_stream_tool_event(
     """Merge incremental fragments for a single tool call while streaming."""
     fragments = collect_stream_fragments(item)
     event_id = first_non_empty(
-        coerce_optional_string(fragment.get("id"))
+        first_non_empty(
+            coerce_optional_string(fragment.get(key))
+            for key in ("id", "tool_call_id", "call_id")
+        )
         for fragment in fragments
+        if isinstance(fragment, dict)
     )
 
     tool_events: dict[str, dict[str, Any]] = stream_state.setdefault(
@@ -97,13 +188,16 @@ def extract_stream_tool_fragment(
     existing: dict[str, Any],
 ) -> dict[str, Any] | None:
     for fragment in fragments:
-        item_type = coerce_optional_string(fragment.get("type"))
-        if item_type and item_type.endswith("_call"):
-            return fragment
+        normalized_fragment = normalize_tool_fragment(fragment)
+        if normalized_fragment is not None:
+            return normalized_fragment
 
         if (
-            fragment.get("response") is not None
-            or fragment.get("result") is not None
+            isinstance(fragment, dict)
+            and (
+                fragment.get("response") is not None
+                or fragment.get("result") is not None
+            )
         ):
             snapshot = dict(existing)
             if fragment.get("response") is not None:
@@ -114,7 +208,8 @@ def extract_stream_tool_fragment(
                 snapshot.setdefault("result", {}).update(
                     fragment.get("result") or {}
                 )
-            return snapshot
+            normalized_snapshot = normalize_tool_fragment(snapshot)
+            return normalized_snapshot or snapshot
 
     return None
 
@@ -123,31 +218,34 @@ def merge_tool_fragment(
     existing: dict[str, Any],
     fragment: dict[str, Any],
 ) -> dict[str, Any]:
-    merged = dict(existing)
+    normalized_existing = normalize_tool_fragment(existing) or existing
+    normalized_fragment = normalize_tool_fragment(fragment) or fragment
+
+    merged = dict(normalized_existing)
 
-    if fragment.get("type"):
-        merged["type"] = fragment["type"]
-    if fragment.get("id"):
-        merged["id"] = fragment["id"]
-    if fragment.get("status"):
-        merged["status"] = fragment["status"]
+    if normalized_fragment.get("type"):
+        merged["type"] = normalized_fragment["type"]
+    if normalized_fragment.get("id"):
+        merged["id"] = normalized_fragment["id"]
+    if normalized_fragment.get("status"):
+        merged["status"] = normalized_fragment["status"]
 
-    action_fragment = fragment.get("action")
+    action_fragment = normalized_fragment.get("action")
     if action_fragment is not None:
         merged_action = merged.get("action", {})
         merged_action = merge_action_payload(merged_action, action_fragment)
         merged["action"] = merged_action
 
-    if fragment.get("response") is not None:
+    if normalized_fragment.get("response") is not None:
         merged["response"] = merge_response_payload(
             merged.get("response"),
-            fragment["response"],
+            normalized_fragment["response"],
         )
 
-    if fragment.get("result") is not None:
+    if normalized_fragment.get("result") is not None:
         merged["result"] = merge_response_payload(
             merged.get("result"),
-            fragment["result"],
+            normalized_fragment["result"],
         )
 
     return merged

src/compendiumscribe/research/execution/streaming.py

@@ -135,6 +135,22 @@ def handle_stream_event(
             f"Deep research stream reported an error: {message}"
         )
 
+    if "tool_call" in normalized:
+        fragment = (
+            get_field(event, "item")
+            or get_field(event, "delta")
+            or get_field(event, "partial")
+            or get_field(event, "data")
+        )
+        if fragment is not None:
+            emit_trace_updates_from_item(
+                fragment,
+                config=config,
+                seen_tokens=seen_trace_tokens,
+                stream_state=stream_state,
+            )
+        return None
+
     if normalized == "response.output_item.added":
         item = get_field(event, "item")
         if item is not None:

src/compendiumscribe/research/parsing.py

@@ -8,6 +8,40 @@
 from .utils import coerce_optional_string, get_field
 
 
+def _iter_text_fragments(value: Any) -> list[str]:
+    """Recursively extract textual fragments from nested response payloads."""
+
+    fragments: list[str] = []
+
+    def visit(candidate: Any) -> None:
+        if candidate is None:
+            return
+
+        if isinstance(candidate, str):
+            if candidate:
+                fragments.append(candidate)
+            return
+
+        if isinstance(candidate, (list, tuple, set)):
+            for item in candidate:
+                visit(item)
+            return
+
+        if isinstance(candidate, dict):
+            # Many response payloads nest text inside these keys.
+            for key in ("text", "value", "content"):
+                if key in candidate:
+                    visit(candidate[key])
+            return
+
+        # Safety fallback: stringify scalars only (avoid object reprs).
+        if isinstance(candidate, (int, float, bool)):
+            fragments.append(str(candidate))
+
+    visit(value)
+    return fragments
+
+
 def parse_deep_research_response(response: Any) -> dict[str, Any]:
     text_payload = collect_response_text(response)
     return decode_json_payload(text_payload)
@@ -16,7 +50,9 @@ def parse_deep_research_response(response: Any) -> dict[str, Any]:
 def collect_response_text(response: Any) -> str:
     output_text = get_field(response, "output_text")
     if output_text:
-        return str(output_text).strip()
+        fragments = _iter_text_fragments(output_text)
+        if fragments:
+            return "".join(fragments).strip()
 
     output_items = get_field(response, "output")
     text_parts: list[str] = []
@@ -26,19 +62,13 @@ def collect_response_text(response: Any) -> str:
             item_type = coerce_optional_string(get_field(item, "type"))
             if item_type == "message":
                 for content in get_field(item, "content") or []:
-                    text_value = coerce_optional_string(
-                        get_field(content, "text")
-                    )
-                    if not text_value:
-                        text_value = coerce_optional_string(
-                            get_field(content, "value")
-                        )
-                    if text_value:
-                        text_parts.append(text_value)
+                    fragments = _iter_text_fragments(content)
+                    if fragments:
+                        text_parts.append("".join(fragments))
             elif item_type == "output_text":
-                text = coerce_optional_string(get_field(item, "text"))
-                if text:
-                    text_parts.append(text)
+                fragments = _iter_text_fragments(item)
+                if fragments:
+                    text_parts.append("".join(fragments))
 
     if text_parts:
         return "".join(text_parts).strip()

tests/research/execution/test_core.py

@@ -51,7 +51,19 @@ def callback(update):
     events = [
         SimpleNamespace(type="response.created"),
         SimpleNamespace(
-            type="response.output_item.added",
+            type="response.tool_call.delta",
+            delta={
+                "type": "web_search_call",
+                "id": "ws_1",
+                "status": "in_progress",
+                "action": {
+                    "type": "search",
+                    "query_delta": "double ",
+                },
+            },
+        ),
+        SimpleNamespace(
+            type="response.tool_call.completed",
             item={
                 "type": "web_search_call",
                 "id": "ws_1",
@@ -91,6 +103,62 @@ def create(self, **kwargs):
     )
 
 
+def test_execute_deep_research_streaming_handles_tool_name_payload():
+    progress_updates = []
+
+    def callback(update):
+        if update.phase == "trace":
+            progress_updates.append(update.message)
+
+    final_response = SimpleNamespace(
+        id="resp_456",
+        status="completed",
+        output_text='{"result": "ok"}',
+        output=[],
+    )
+
+    events = [
+        SimpleNamespace(type="response.created"),
+        SimpleNamespace(
+            type="response.tool_call.delta",
+            delta={
+                "type": "tool_call_delta",
+                "tool_name": "web.search",
+                "tool_call_id": "ws_2",
+                "arguments": {"query_delta": "history of"},
+            },
+        ),
+        SimpleNamespace(
+            type="response.tool_call.completed",
+            item={
+                "tool_name": "web.search",
+                "tool_call_id": "ws_2",
+                "status": "completed",
+                "arguments": {"query": "history of flutes"},
+            },
+        ),
+        SimpleNamespace(type="response.completed", response=final_response),
+    ]
+
+    class StubResponses:
+        def __init__(self):
+            self.calls: list[dict[str, object]] = []
+
+        def create(self, **kwargs):
+            self.calls.append(kwargs)
+            return StubStream(events, final_response)
+
+    responses = StubResponses()
+    client = SimpleNamespace(responses=responses)
+
+    config = ResearchConfig(stream_progress=True, progress_callback=callback)
+
+    execute_deep_research(client, "prompt", config)
+
+    assert progress_updates
+    assert any("history of flutes" in message for message in progress_updates)
+
+
 def test_execute_deep_research_streaming_raises_on_error():
     events = [
         SimpleNamespace(type="response.created"),