Refa: follow-up expose agent structured outputs in non-stream completions (#13524)

### What problem does this PR solve?

Follow-up expose agent structured outputs in non-stream completions
#13389.

### Type of change

- [x] Documentation Update
- [x] Refactoring

---------

Co-authored-by: writinwaters <cai.keith@gmail.com>

api/apps/sdk/session.py

@@ -597,11 +597,12 @@ async def agent_completions(tenant_id, agent_id):
                 reference.update(ans["data"]["reference"])
 
             if ans.get("event") == "node_finished":
-                node_out = ans.get("data", {}).get("outputs", {})
-                if node_out.get("structured"):
-                    structured_output = node_out["structured"]
+                data = ans.get("data", {})
+                node_out = data.get("outputs", {})
+                component_id = data.get("component_id")
+                if component_id is not None and "structured" in node_out:
+                    structured_output[component_id] = copy.deepcopy(node_out["structured"])
                 if return_trace:
-                    data = ans.get("data", {})
                     trace_items.append(
                         {
                             "component_id": data.get("component_id"),

docs/references/http_api_reference.md

@@ -3914,16 +3914,16 @@ Asks a specified agent a question to start an AI-powered conversation.
   - `"session_id"`: `string` (optional)
   - `"inputs"`: `object` (optional)
   - `"user_id"`: `string` (optional)
-  - `"return_trace"`: `boolean` (optional, default `false`) — include execution trace logs.
+  - `"return_trace"`: `boolean` (optional, default `false`) — whether to include execution trace logs. See the `node_finished` event.
   - `"release"`: `boolean` (optional, default `false`) - whether to visit the latest published canvas.
 
 #### Streaming events to handle
 
-When `stream=true`, the server sends Server-Sent Events (SSE). Clients should handle these `event` types:
+When `stream=true`, the server sends Server-Sent Events (SSE). A client should handle these events:
 
-- `message`: streaming content from Message components.
-- `message_end`: end of a Message component; may include `reference`/`attachment`.
-- `node_finished`: a component finishes; `data.inputs/outputs/error/elapsed_time` describe the node result. If `return_trace=true`, the trace is attached inside the same `node_finished` event (`data.trace`).
+- `message`: Streaming content from the **Message** components.
+- `message_end`: End of a **Message** component, which may include `reference`/`attachment`.
+- `node_finished`: A component finishes; `data.inputs/outputs/error/elapsed_time` describes the node result. If a component produces structured output, read it from that component's `data.outputs.structured`. If `return_trace=true`, the trace is attached inside the same `node_finished` event (`data.trace`).
 
 The stream terminates with `[DONE]`.
 
@@ -4203,6 +4203,8 @@ When `extra_body.reference_metadata.include` is `true`, each reference chunk may
 
 Non-stream:
 
+If one or more components produce structured output, ensure you set `return_trace=true` and check each component's structured output via `trace`. The top-level `data.structured` field is a shortcut aggregated by `component_id`.
+
 ```json
 {
     "code": 0,

test/testcases/test_http_api/test_session_management/test_session_sdk_routes_unit.py

@@ -862,7 +862,18 @@ def test_agent_completions_stream_and_nonstream_unit(monkeypatch):
 
     async def _agent_stream(*_args, **_kwargs):
         yield "data:not-json"
-        yield "data:" + json.dumps({"event": "node_finished", "data": {"component_id": "c1"}})
+        yield "data:" + json.dumps(
+            {
+                "event": "node_finished",
+                "data": {"component_id": "c1", "outputs": {"structured": {"alpha": 1}}},
+            }
+        )
+        yield "data:" + json.dumps(
+            {
+                "event": "node_finished",
+                "data": {"component_id": "c2", "outputs": {"structured": {}}},
+            }
+        )
         yield "data:" + json.dumps({"event": "other", "data": {}})
         yield "data:" + json.dumps({"event": "message", "data": {"content": "hello"}})
 
@@ -878,14 +889,36 @@ def test_agent_completions_stream_and_nonstream_unit(monkeypatch):
 
     async def _agent_nonstream(*_args, **_kwargs):
         yield "data:" + json.dumps({"event": "message", "data": {"content": "A", "reference": {"doc": "r"}}})
-        yield "data:" + json.dumps({"event": "node_finished", "data": {"component_id": "c2"}})
+        yield "data:" + json.dumps(
+            {
+                "event": "node_finished",
+                "data": {"component_id": "c2", "outputs": {"structured": {"foo": "bar"}}},
+            }
+        )
+        yield "data:" + json.dumps(
+            {
+                "event": "node_finished",
+                "data": {"component_id": "c3", "outputs": {"structured": {"baz": 1}}},
+            }
+        )
+        yield "data:" + json.dumps(
+            {
+                "event": "node_finished",
+                "data": {"component_id": "c4", "outputs": {"structured": {}}},
+            }
+        )
 
     monkeypatch.setattr(module, "agent_completion", _agent_nonstream)
     monkeypatch.setattr(module, "get_request_json", lambda: _AwaitableValue({"stream": False, "return_trace": True}))
     res = _run(inspect.unwrap(module.agent_completions)("tenant-1", "agent-1"))
     assert res["data"]["data"]["content"] == "A"
     assert res["data"]["data"]["reference"] == {"doc": "r"}
-    assert res["data"]["data"]["trace"][0]["component_id"] == "c2"
+    assert res["data"]["data"]["structured"] == {
+        "c2": {"foo": "bar"},
+        "c3": {"baz": 1},
+        "c4": {},
+    }
+    assert [item["component_id"] for item in res["data"]["data"]["trace"]] == ["c2", "c3", "c4"]
 
     async def _agent_nonstream_broken(*_args, **_kwargs):
         yield "data:{"