Harden child output collection

DESIGN_REVIEW.md

@@ -0,0 +1,50 @@
+# Child Output Collection Design Review
+
+## Contract
+
+Embedded-child output collection now follows the durable yield result path:
+
+1. A parent submits child nodes and calls `n:yield({ run_nodes = child_node_ids })`.
+2. The orchestrator tracks each yielded child and, when it exits, records the exact `node.result` data row ID in `yield_info.results[child_node_id]`.
+3. Before satisfying the parent yield on child exit, the orchestrator flushes pending commits so the result rows and routed output rows are visible to the parent.
+4. Parent node implementations pass the returned yield result map into `child_output.collect_outputs(...)`.
+5. `child_output` resolves those exact `node.result` data IDs, reads their `data_ids` envelope, and then fetches the routed `node.output` rows by data ID.
+
+The invariants are:
+
+- Live yield collection must not guess output by child `node_id`.
+- A structured successful result with an empty or missing `data_ids` array means "no child output"; it is not a license to scan child outputs.
+- A missing returned result row is an error.
+- Legacy non-envelope result rows may fall back to output lookup, but only by the `node_id` on the exact returned `node.result` row.
+- Routed output/error rows preserve any target-provided `data_id`, so the `data_ids` envelope and the data table agree.
+- Recovery paths may collect already durable child outputs before yielding again, but must not poll-and-complete-empty while children are pending.
+
+## Changes By File
+
+- `src/child_output.lua`: Added strict yield-result collection, exact result-row resolution, routed output lookup by `data_id`, structured child failure propagation, and recovery resume plumbing that reuses returned yield result IDs.
+- `src/_index.yaml`: Declares the `json` module for `child_output`; without this, runtime binding fails before node execution.
+- `src/node.lua`: Preserves target-provided output and error `data_id` values during routing.
+- `src/node/func/node.lua`: Uses `child_output.collect_outputs(...)` after live yields and via recovery resume; the plain no-children path remains a direct completion path with no yield or child-output query.
+- `src/node/func/func_child_output_test.lua`: Pins yield-result-based collection, no guessing when a result envelope has no output IDs, scoped legacy fallback, recovery collection before re-yield, and unchanged no-children behavior.
+- `src/node/cycle/cycle.lua`: Uses strict collection for template and control-command children; wraps structured child collection failures with cycle/template context.
+- `src/node/agent/node.lua` and `src/node/agent/_index.yaml`: Agent control-command children use shared strict output collection and bind the helper.
+- `src/node/agent/delegation_handler.lua`: Treats invalid delegated JSON as an error instead of returning raw text as a successful delegated result.
+- `src/runner/orchestrator.lua`: Flushes startup and pending commits before satisfying a yield after child exit.
+- `src/runner/workflow_state.lua` and `src/persist/data_reader.lua`: Replace bad `pcall(json.decode)` patterns with the local JSON API's explicit error return.
+- `src/node_test.lua`: Pins preservation of target-provided output and error data IDs.
+
+## Risks
+
+- Legacy fallback is intentionally narrow; older non-envelope child result rows still work only when the exact returned `node.result` row provides a `node_id`.
+- Strict result-row lookup turns missing or malformed result envelopes into visible failures. This is expected and prevents silent `{}` completion.
+- Callers that create embedded children but bypass `n:yield` still use the older `node_id` query path only when no yield results exist.
+
+## Design Gaps
+
+No strict-contract design gaps were found during this pass. The failures uncovered in verification were implementation issues: the missing `json` module binding for `child_output`, and cycle's attempt to treat a structured child failure as a string instead of wrapping it with cycle/template context.
+
+## Verification
+
+- Initial full suite from the owner WIP: `727 passed, 121 failed`.
+- After binding `json`: `899 passed, 1 failed`.
+- Final full suite after cycle error wrapping: `900 tests` passed in `208.0s`.

src/_index.yaml

@@ -78,6 +78,18 @@ entries:
       description: "Defines constants for dataflow operations including command types, data types, \ncontent types, status values, and message topics.\n"
     source: file://consts.lua
 
+  # userspace.dataflow:child_output
+  - name: child_output
+    kind: library.lua
+    meta:
+      name: Child Output Helpers
+      comment: Shared helpers for collecting and resuming child node outputs.
+    source: file://child_output.lua
+    modules:
+      - json
+    imports:
+      consts: userspace.dataflow:consts
+
   - name: agent_ref
     kind: library.lua
     meta:

src/child_output.lua

@@ -0,0 +1,175 @@
+local json = require("json")
+
+local child_output = {}
+
+child_output._deps = {
+    consts = require("consts")
+}
+
+function child_output.query_node_outputs(n, node_ids)
+    if type(node_ids) ~= "table" or #node_ids == 0 then
+        return {}, nil
+    end
+
+    local ok, output_data_or_err = pcall(function()
+        return (n:query() :: any)
+            :with_nodes(node_ids)
+            :with_data_types(child_output._deps.consts.DATA_TYPE.NODE_OUTPUT)
+            :fetch_options({ replace_references = true })
+            :all()
+    end)
+
+    if not ok then
+        return nil, "Failed to query output data: " .. tostring(output_data_or_err)
+    end
+
+    return output_data_or_err, nil
+end
+
+local function decode_content(content)
+    if type(content) ~= "string" then
+        return content
+    end
+
+    local decoded, decode_err = json.decode(content)
+    if not decode_err then
+        return decoded
+    end
+    return content
+end
+
+local function child_failure_message(content)
+    if type(content.error) == "table" then
+        return tostring(content.error.message or content.error.status or content.message or "Child workflow failed")
+    end
+
+    return tostring(content.error or content.message or "Child workflow failed")
+end
+
+function child_output.outputs_from_yield_results(n, yield_results)
+    if type(yield_results) ~= "table" or next(yield_results) == nil then
+        return nil, nil
+    end
+
+    local result_ids = {}
+    local expected_result_ids = {}
+    for _, result_id in pairs(yield_results) do
+        if type(result_id) == "string" and result_id ~= "" then
+            table.insert(result_ids, result_id)
+            expected_result_ids[result_id] = true
+        end
+    end
+    if #result_ids == 0 then
+        return nil, nil
+    end
+
+    local ok, result_rows_or_err = pcall(function()
+        return (n:query() :: any)
+            :with_data(result_ids)
+            :with_data_types(child_output._deps.consts.DATA_TYPE.NODE_RESULT)
+            :fetch_options({ replace_references = true })
+            :all()
+    end)
+    if not ok then
+        return nil, "Failed to query child result data: " .. tostring(result_rows_or_err)
+    end
+
+    local output_ids = {}
+    local fallback_node_ids = {}
+    local seen_output_ids = {}
+    local seen_fallback_node_ids = {}
+    local seen_result_ids = {}
+    for _, row in ipairs(result_rows_or_err or {}) do
+        if row.data_id then
+            seen_result_ids[row.data_id] = true
+        end
+        local content = decode_content(row.content)
+        if type(content) == "table" then
+            if content.success == false then
+                return nil, {
+                    code = "CHILD_WORKFLOW_FAILED",
+                    message = child_failure_message(content),
+                    status = "Child workflow failed"
+                }
+            end
+
+            if content.data_ids ~= nil and type(content.data_ids) ~= "table" then
+                return nil, "Child result data_ids must be an array"
+            end
+
+            for _, data_id in ipairs(content.data_ids or {}) do
+                if type(data_id) == "string" and data_id ~= "" and not seen_output_ids[data_id] then
+                    table.insert(output_ids, data_id)
+                    seen_output_ids[data_id] = true
+                end
+            end
+        elseif row.node_id and not seen_fallback_node_ids[row.node_id] then
+            table.insert(fallback_node_ids, row.node_id)
+            seen_fallback_node_ids[row.node_id] = true
+        end
+    end
+
+    for _, result_id in ipairs(result_ids) do
+        if not seen_result_ids[result_id] and expected_result_ids[result_id] then
+            return nil, "Child result data not found: " .. result_id
+        end
+    end
+
+    if #output_ids == 0 then
+        return child_output.query_node_outputs(n, fallback_node_ids)
+    end
+
+    local output_ok, output_rows_or_err = pcall(function()
+        return (n:query() :: any)
+            :with_data(output_ids)
+            :with_data_types(child_output._deps.consts.DATA_TYPE.NODE_OUTPUT)
+            :fetch_options({ replace_references = true })
+            :all()
+    end)
+    if not output_ok then
+        return nil, "Failed to query child output data: " .. tostring(output_rows_or_err)
+    end
+
+    return output_rows_or_err, nil
+end
+
+function child_output.collect_outputs(n, node_ids, yield_results)
+    local has_yield_results = type(yield_results) == "table" and next(yield_results) ~= nil
+    local output_data, collect_err = child_output.outputs_from_yield_results(n, yield_results)
+    if collect_err then
+        return nil, collect_err
+    end
+
+    if has_yield_results then
+        return output_data or {}, nil
+    end
+
+    if output_data and #output_data > 0 then
+        return output_data, nil
+    end
+
+    return child_output.query_node_outputs(n, node_ids)
+end
+
+function child_output.resume_children(n, child_node_ids, collect_children_result)
+    if type(child_node_ids) ~= "table" or #child_node_ids == 0 then
+        return nil, nil
+    end
+
+    local existing, existing_err = collect_children_result(n, child_node_ids)
+    if existing_err then
+        return nil, existing_err
+    end
+    if existing ~= nil then
+        return existing, nil
+    end
+
+    local yield_result, yield_err = n:yield({ run_nodes = child_node_ids })
+    if yield_err then
+        return nil, "Failed to resume children: " .. tostring(yield_err)
+    end
+
+    return collect_children_result(n, child_node_ids, yield_result)
+end
+
+return child_output

src/node.lua

@@ -601,7 +601,7 @@ function methods:_route_outputs(content)
                        " (transform: " .. tostring(target.transform or "none") .. ")"
         end
 
-        local data_id = uuid.v7()
+        local data_id = target.data_id or uuid.v7()
         data_id_count = data_id_count + 1
         routed_data_ids[data_id_count] = data_id
 
@@ -659,7 +659,7 @@ function methods:_route_errors(error_content)
             end
         end
 
-        local data_id = uuid.v7()
+        local data_id = target.data_id or uuid.v7()
         data_id_count = data_id_count + 1
         routed_data_ids[data_id_count] = data_id
 

src/node/agent/_index.yaml

@@ -169,6 +169,7 @@ entries:
       tools: wippy.agent.discovery:tools
       agent_consts: userspace.dataflow.node.agent:consts
       agent_context: wippy.agent:context
+      child_output: userspace.dataflow:child_output
       consts: userspace.dataflow:consts
       control_handler: userspace.dataflow.node.agent:control_handler
       delegation_handler: userspace.dataflow.node.agent:delegation_handler

src/node/agent/delegation_handler.lua

@@ -329,7 +329,7 @@ function delegation_handler.collect_delegation_result(delegation_info, parent_no
     if actual_result_data.content_type == "application/json" and type(result_content) == "string" then
         local decoded, decode_err = json.decode(result_content)
         if decode_err then
-            return result_content, nil
+            return nil, "Failed to parse delegated result content: " .. tostring(decode_err)
         end
         result_content = decoded
     end

src/node/agent/node.lua

@@ -11,6 +11,7 @@ local checkpoint_runtime = require("checkpoint_runtime")
 local prompt_builder = require("prompt_builder")
 local control_handler = require("control_handler")
 local delegation_handler = require("delegation_handler")
+local child_output = require("child_output")
 local agent_consts = require("agent_consts")
 local consts = require("consts")
 local tools = require("tools")
@@ -1141,12 +1142,13 @@ local function run_control_response_commands(control_responses, agent_ctx, n, it
 
     local yield_result, yield_err = n:yield({ run_nodes = created_node_ids })
     if yield_result then
-        local reader = n:query()
-            :with_nodes(created_node_ids)
-            :with_data_types(consts.DATA_TYPE.NODE_OUTPUT)
-            :fetch_options({ replace_references = true })
-
-        local output_data = reader:all()
+        local output_data, output_err = child_output.collect_outputs(n, created_node_ids, yield_result)
+        if output_err then
+            if type(output_err) == "table" then
+                return tostring(output_err.message or output_err.status or "Child workflow failed")
+            end
+            return tostring(output_err)
+        end
 
         if output_data and #output_data > 0 then
             local output_content = output_data[1].content

src/node/cycle/_index.yaml

@@ -30,6 +30,7 @@ entries:
       - time
       - uuid
     imports:
+      child_output: userspace.dataflow:child_output
       consts: userspace.dataflow:consts
       data_reader: userspace.dataflow.persist:data_reader
       node_reader: userspace.dataflow.persist:node_reader