diff --git a/DESIGN_REVIEW.md b/DESIGN_REVIEW.md
new file mode 100644
index 0000000..585dd70
--- /dev/null
+++ b/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`.
diff --git a/src/_index.yaml b/src/_index.yaml
index 06da8f4..68e2ec1 100644
--- a/src/_index.yaml
+++ b/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:
diff --git a/src/child_output.lua b/src/child_output.lua
new file mode 100644
index 0000000..5d6445d
--- /dev/null
+++ b/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
diff --git a/src/node.lua b/src/node.lua
index 855d96a..e721040 100644
--- a/src/node.lua
+++ b/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
 
diff --git a/src/node/agent/_index.yaml b/src/node/agent/_index.yaml
index fca0109..ef9b808 100644
--- a/src/node/agent/_index.yaml
+++ b/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
diff --git a/src/node/agent/delegation_handler.lua b/src/node/agent/delegation_handler.lua
index 9380476..6dea3e5 100644
--- a/src/node/agent/delegation_handler.lua
+++ b/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
diff --git a/src/node/agent/node.lua b/src/node/agent/node.lua
index bec054a..3ba7ba7 100644
--- a/src/node/agent/node.lua
+++ b/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
diff --git a/src/node/cycle/_index.yaml b/src/node/cycle/_index.yaml
index 5ad869d..b57b721 100644
--- a/src/node/cycle/_index.yaml
+++ b/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
diff --git a/src/node/cycle/cycle.lua b/src/node/cycle/cycle.lua
index 71c3a68..3f142cf 100644
--- a/src/node/cycle/cycle.lua
+++ b/src/node/cycle/cycle.lua
@@ -23,7 +23,8 @@ cycle._deps = {
     consts = require("consts"),
     template_graph = require("template_graph"),
     data_reader = require("data_reader"),
-    node_reader = require("node_reader")
+    node_reader = require("node_reader"),
+    child_output = require("child_output")
 }
 
 cycle.DEFAULTS = table.freeze({
@@ -325,23 +326,7 @@ local function parse_content(content: any, content_type: any)
 end
 
 query_node_outputs = function(n, node_ids: {any})
-    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(cycle._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
+    return cycle._deps.child_output.query_node_outputs(n, node_ids)
 end
 
 local function collect_node_ids(uuid_mapping)
@@ -383,10 +368,22 @@ parse_output_rows = function(output_data)
     return results
 end
 
-local function collect_template_outputs(n, uuid_mapping)
+local function collect_outputs(n, node_ids, yield_result)
+    return cycle._deps.child_output.collect_outputs(n, node_ids, yield_result)
+end
+
+local function collect_error_message(err)
+    if type(err) == "table" then
+        return tostring(err.message or err.status or err.code or "Child workflow failed")
+    end
+
+    return tostring(err)
+end
+
+local function collect_template_outputs(n, uuid_mapping, yield_result)
     local iteration_node_ids = collect_node_ids(uuid_mapping)
 
-    local output_data, query_err = query_node_outputs(n, iteration_node_ids)
+    local output_data, query_err = collect_outputs(n, iteration_node_ids, yield_result)
     if query_err then
         return nil, query_err
     end
@@ -465,12 +462,12 @@ local function execute_template_iteration(n, template_graph, current_state, last
 
     local yield_result, yield_err = n:yield({ run_nodes = all_nodes })
     if yield_err then
-        return nil, "Template execution failed: " .. (yield_err :: string)
+        return nil, "Template execution failed: " .. collect_error_message(yield_err)
     end
 
-    local outputs, collect_err = collect_template_outputs(n, uuid_mapping)
+    local outputs, collect_err = collect_template_outputs(n, uuid_mapping, yield_result)
     if collect_err then
-        return nil, "Template execution failed: " .. (collect_err :: string)
+        return nil, "Template execution failed: " .. collect_error_message(collect_err)
     end
 
     return outputs, nil
@@ -479,10 +476,10 @@ end
 -- Collects the (durable) outputs of the given child nodes into a single result
 -- (one content, or an array of contents). Used both on the normal path and when
 -- resuming an interrupted iteration on recovery.
-local function collect_children_result(n, node_ids: {any})
-    local output_data, collect_err = query_node_outputs(n, node_ids)
+local function collect_children_result(n, node_ids: {any}, yield_result)
+    local output_data, collect_err = collect_outputs(n, node_ids, yield_result)
     if collect_err then
-        return nil, "Failed to collect child outputs: " .. (collect_err :: string)
+        return nil, "Failed to collect child outputs: " .. collect_error_message(collect_err)
     end
 
     if output_data and #output_data > 0 then
@@ -580,34 +577,14 @@ local function process_control_commands(n, control_commands, iteration_number, c
             return nil, "Control command execution failed: " .. (yield_err :: string)
         end
 
-        return collect_children_result(n, created_node_ids)
+        return collect_children_result(n, created_node_ids, yield_result)
     end
 
     return nil, nil
 end
 
--- Resumes an iteration whose children were created before a crash: collects their
--- outputs if already complete, otherwise waits on the existing children (no re-run,
--- no duplicate node creation).
 local function resume_children(n, child_node_ids: {any}?)
-    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: " .. (yield_err :: string)
-    end
-
-    return collect_children_result(n, child_node_ids)
+    return cycle._deps.child_output.resume_children(n, child_node_ids, collect_children_result)
 end
 
 local function update_node_metadata(n, metadata_updates)
diff --git a/src/node/func/_index.yaml b/src/node/func/_index.yaml
index c8eaa3b..82e4b2d 100644
--- a/src/node/func/_index.yaml
+++ b/src/node/func/_index.yaml
@@ -56,6 +56,23 @@ entries:
       test: wippy.test:test
     method: run_tests
 
+  # userspace.dataflow.node.func:func_child_output_test
+  - name: func_child_output_test
+    kind: function.lua
+    meta:
+      name: Func Node Child Output Tests
+      type: test
+      comment: Verifies func node child output collection and recovery waits.
+      group: Workflow / Function Node
+    source: file://func_child_output_test.lua
+    modules:
+      - uuid
+    imports:
+      consts: userspace.dataflow:consts
+      func: userspace.dataflow.node.func:node
+      test: wippy.test:test
+    method: run_tests
+
   # userspace.dataflow.node.func:func_test
   - name: func_test
     kind: function.lua
@@ -106,6 +123,7 @@ entries:
       - json
       - time
     imports:
+      child_output: userspace.dataflow:child_output
       consts: userspace.dataflow:consts
       node: userspace.dataflow:node
     method: run
diff --git a/src/node/func/func_child_output_test.lua b/src/node/func/func_child_output_test.lua
new file mode 100644
index 0000000..7910440
--- /dev/null
+++ b/src/node/func/func_child_output_test.lua
@@ -0,0 +1,365 @@
+local test = require("test")
+local uuid = require("uuid")
+local consts = require("consts")
+local func = require("func")
+
+local function child_output_row()
+    return {
+        data_id = "child-output-1",
+        data_type = consts.DATA_TYPE.NODE_OUTPUT,
+        node_id = "child-1",
+        key = "result",
+        content = { child = true },
+        content_type = consts.CONTENT_TYPE.JSON
+    }
+end
+
+local function make_reader(state)
+    return {
+        with_data = function(self: any, data_ids: any)
+            state.queried_data_ids = data_ids
+            return self
+        end,
+        with_nodes = function(self: any, node_ids: any)
+            state.queried_nodes = node_ids
+            state.queried_data_ids = nil
+            return self
+        end,
+        with_data_types = function(self: any, data_type: any)
+            state.queried_data_type = data_type
+            return self
+        end,
+        fetch_options = function(self: any, options: any)
+            state.fetch_options = options
+            return self
+        end,
+        all = function()
+            state.output_query_count = state.output_query_count + 1
+            if state.queried_data_type == consts.DATA_TYPE.NODE_RESULT then
+                local rows = {}
+                for _, data_id in ipairs(state.queried_data_ids or {}) do
+                    if data_id == "child-result-1" then
+                        table.insert(rows, {
+                            data_id = "child-result-1",
+                            data_type = consts.DATA_TYPE.NODE_RESULT,
+                            node_id = "child-1",
+                            content = state.result_content or {
+                                success = true,
+                                data_ids = state.result_data_ids or { "child-output-1" }
+                            }
+                        })
+                    end
+                end
+                return rows
+            end
+            if state.queried_data_type == consts.DATA_TYPE.NODE_OUTPUT and state.queried_data_ids then
+                for _, data_id in ipairs(state.queried_data_ids) do
+                    if data_id == "child-output-1" then
+                        return { child_output_row() }
+                    end
+                end
+                return {}
+            end
+            if state.output_available and state.output_available(state) then
+                return { child_output_row() }
+            end
+            return {}
+        end
+    }
+end
+
+local function make_async_command(result)
+    local response_topic = "func-child-output-response-" .. uuid.v7()
+    local response_channel = process.listen(response_topic)
+
+    return {
+        topic = response_topic,
+        response_channel = response_channel,
+        command = {
+            response = function()
+                return response_channel
+            end,
+            is_canceled = function()
+                return false
+            end,
+            result = function()
+                return {
+                    data = function()
+                        return result
+                    end
+                }, nil
+            end,
+            cancel = function()
+            end
+        }
+    }
+end
+
+local function install_async_function(result)
+    local command_pair = make_async_command(result)
+
+    func._deps.funcs = {
+        new = function()
+            return {
+                with_context = function(self: any)
+                    return self
+                end,
+                async = function()
+                    return command_pair.command
+                end
+            }
+        end
+    }
+
+    process.send(process.pid(), command_pair.topic :: string, {})
+end
+
+local function make_mock_node(state, options)
+    options = options or {}
+
+    return {
+        dataflow_id = "dataflow-1",
+        node_id = "func-1",
+        path = {},
+        config = function()
+            return options.config or { func_id = "userspace.dataflow.node.func:test_func" }
+        end,
+        metadata = function()
+            return options.metadata or {}
+        end,
+        inputs = function()
+            return {
+                default = {
+                    content = options.input or { ok = true }
+                }
+            }
+        end,
+        update_metadata = function(_self: any, metadata: any)
+            table.insert(state.metadata_updates, metadata)
+            return true
+        end,
+        command = function(_self: any, cmd: any)
+            table.insert(state.commands, cmd)
+            return true
+        end,
+        yield = function(_self: any, yield_options: any)
+            state.yield_calls = state.yield_calls + 1
+            table.insert(state.yield_options, yield_options)
+            if options.yield_error then
+                return nil, options.yield_error
+            end
+            return options.yield_result or state.yield_result or {}, nil
+        end,
+        query = function()
+            state.query_calls = state.query_calls + 1
+            return make_reader(state)
+        end,
+        complete = function(_self: any, output: any, message: string?)
+            state.completed = { output = output, message = message }
+            return { success = true, result = output, message = message }
+        end,
+        fail = function(_self: any, err: any, message: string?)
+            state.failed = { err = err, message = message }
+            return { success = false, error = err, message = message }
+        end
+    }
+end
+
+local function make_state(output_available)
+    return {
+        commands = {},
+        metadata_updates = {},
+        yield_options = {},
+        yield_calls = 0,
+        query_calls = 0,
+        output_query_count = 0,
+        output_available = output_available
+    }
+end
+
+local function define_tests()
+    describe("Func Node Child Output Collection", function()
+        local original_deps: any
+
+        before_each(function()
+            original_deps = {
+                node = func._deps.node,
+                funcs = func._deps.funcs,
+                consts = func._deps.consts
+            }
+        end)
+
+        after_each(function()
+            func._deps.node = original_deps.node
+            func._deps.funcs = original_deps.funcs
+            func._deps.consts = original_deps.consts
+        end)
+
+        it("collects child output through the yield node-result ids", function()
+            local state = make_state(function()
+                return false
+            end)
+            state.yield_result = {
+                ["child-1"] = "child-result-1"
+            }
+            local function_result = {
+                _control = {
+                    commands = {
+                        {
+                            type = consts.COMMAND_TYPES.CREATE_NODE,
+                            payload = {
+                                node_id = "child-1",
+                                node_type = "userspace.dataflow.node.func:node",
+                                status = consts.STATUS.PENDING,
+                                config = {}
+                            }
+                        }
+                    }
+                }
+            }
+
+            install_async_function(function_result)
+            func._deps.node = {
+                new = function()
+                    return make_mock_node(state), nil
+                end
+            }
+
+            local result = func.run({})
+
+            test.is_true(result.success)
+            test.eq(state.yield_calls, 1, "func waits once for the child yield")
+            test.eq(result.result._dataflow_ref, "child-output-1", "func completes from child output")
+            test.eq(state.queried_data_ids[1], "child-output-1", "func resolves output from node-result data ids")
+            test.is_nil(state.queried_nodes, "func does not fall back to node-id output lookup after yield")
+        end)
+
+        it("does not fall back to child node output after a completed yield omits data ids", function()
+            local state = make_state(function()
+                return true
+            end)
+            state.result_data_ids = {}
+            state.yield_result = {
+                ["child-1"] = "child-result-1"
+            }
+            local function_result = {
+                _control = {
+                    commands = {
+                        {
+                            type = consts.COMMAND_TYPES.CREATE_NODE,
+                            payload = {
+                                node_id = "child-1",
+                                node_type = "userspace.dataflow.node.func:node",
+                                status = consts.STATUS.PENDING,
+                                config = {}
+                            }
+                        }
+                    }
+                }
+            }
+
+            install_async_function(function_result)
+            func._deps.node = {
+                new = function()
+                    return make_mock_node(state), nil
+                end
+            }
+
+            local result = func.run({})
+
+            test.is_true(result.success)
+            test.eq(state.yield_calls, 1, "func waits once for the child yield")
+            test.eq(result.result._dataflow_ref, nil, "live yield result without output ids does not guess child output")
+            test.is_nil(state.queried_nodes, "live yield result path never falls back to node-id lookup")
+        end)
+
+        it("uses the returned result row node id only for legacy non-envelope results", function()
+            local state = make_state(function()
+                return true
+            end)
+            state.result_content = "Completed"
+            state.yield_result = {
+                ["child-1"] = "child-result-1"
+            }
+            local function_result = {
+                _control = {
+                    commands = {
+                        {
+                            type = consts.COMMAND_TYPES.CREATE_NODE,
+                            payload = {
+                                node_id = "child-1",
+                                node_type = "userspace.dataflow.node.func:node",
+                                status = consts.STATUS.PENDING,
+                                config = {}
+                            }
+                        }
+                    }
+                }
+            }
+
+            install_async_function(function_result)
+            func._deps.node = {
+                new = function()
+                    return make_mock_node(state), nil
+                end
+            }
+
+            local result = func.run({})
+
+            test.is_true(result.success)
+            test.eq(state.yield_calls, 1, "func waits once for the child yield")
+            test.eq(result.result._dataflow_ref, "child-output-1", "legacy result rows still resolve child output")
+            test.eq(state.queried_nodes[1], "child-1", "legacy fallback is scoped to the returned result row node id")
+        end)
+
+        it("collects pending child output on recovery before yielding", function()
+            local state = make_state(function()
+                return true
+            end)
+
+            func._deps.node = {
+                new = function()
+                    return make_mock_node(state, {
+                        metadata = {
+                            func_pending = {
+                                child_node_ids = { "child-1" }
+                            }
+                        },
+                        config = {},
+                        yield_error = "recovery should collect existing child output"
+                    }), nil
+                end
+            }
+
+            local result = func.run({})
+
+            test.is_true(result.success)
+            test.eq(state.yield_calls, 0, "recovery uses durable child output without a fresh yield")
+            test.eq(result.result._dataflow_ref, "child-output-1", "recovery completes from child output")
+        end)
+
+        it("does not yield or query when the function returns no embedded children", function()
+            local state = make_state(function()
+                return false
+            end)
+
+            install_async_function({ ok = true })
+            func._deps.node = {
+                new = function()
+                    return make_mock_node(state, {
+                        yield_error = "plain function result should not yield"
+                    }), nil
+                end
+            }
+
+            local result = func.run({})
+
+            test.is_true(result.success)
+            test.eq(result.result.ok, true)
+            test.eq(state.yield_calls, 0, "plain function result does not yield")
+            test.eq(state.query_calls, 0, "plain function result does not query child output")
+        end)
+    end)
+end
+
+return test.run_cases(define_tests)
diff --git a/src/node/func/node.lua b/src/node/func/node.lua
index 3b58853..42bba36 100644
--- a/src/node/func/node.lua
+++ b/src/node/func/node.lua
@@ -1,5 +1,4 @@
 local func = {}
-local time = require("time")
 
 local ERROR_MISSING_FUNC_ID = "MISSING_FUNC_ID"
 local ERROR_NO_INPUT_DATA = "NO_INPUT_DATA"
@@ -9,7 +8,8 @@ local ERROR_FUNCTION_EXECUTION_FAILED = "FUNCTION_EXECUTION_FAILED"
 func._deps = {
     node = require("node"),
     funcs = require("funcs"),
-    consts = require("consts")
+    consts = require("consts"),
+    child_output = require("child_output")
 }
 
 local function build_execution_context_with_dataflow(base_context, dataflow_id, node_id, path)
@@ -104,51 +104,23 @@ local function safe_inputs(n)
     return inputs_or_err, nil
 end
 
--- Yields for the given child nodes and completes from their outputs. Checkpoints the
--- child ids before yielding so a crash mid-yield resumes these same children on recovery
--- instead of re-running the user function (which would create duplicate children).
-local function finish_with_children(n, created_node_ids, fallback_result)
-    n:update_metadata({ func_pending = { child_node_ids = created_node_ids } })
-
-    local _yield_result, yield_err = n:yield({ run_nodes = created_node_ids })
-    if yield_err then
-        return n:fail({
-            code = ERROR_FUNCTION_EXECUTION_FAILED,
-            message = "Control command execution failed: " .. yield_err
-        }, "Control command execution failed")
-    end
-
-    local reader = n:query()
-        :with_nodes(created_node_ids)
-        :with_data_types(func._deps.consts.DATA_TYPE.NODE_OUTPUT)
-        :fetch_options({ replace_references = true })
-
-    local output_data = nil
-    for _ = 1, 10 do
-        output_data = reader:all()
-        if output_data and #output_data > 0 then
-            break
-        end
-        time.sleep("10ms")
-    end
-
+local function build_child_output_result(output_data)
     if output_data and #output_data > 0 then
         for _, output in ipairs(output_data) do
             if output.discriminator == "error" then
                 local error_content = output.content
-                return n:fail({
+                return nil, {
                     code = "CHILD_WORKFLOW_FAILED",
-                    message = type(error_content) == "string" and error_content or "Child workflow failed"
-                }, "Child workflow failed")
+                    message = type(error_content) == "string" and error_content or "Child workflow failed",
+                    status = "Child workflow failed"
+                }
             end
         end
 
-        local final_output
-
         if #output_data == 1 then
-            final_output = {
+            return {
                 _dataflow_ref = output_data[1].data_id
-            }
+            }, nil
         else
             local ref_array = {}
             for _, output in ipairs(output_data) do
@@ -156,9 +128,70 @@ local function finish_with_children(n, created_node_ids, fallback_result)
                     _dataflow_ref = output.data_id
                 })
             end
-            final_output = ref_array
+            return ref_array, nil
+        end
+    end
+
+    return nil, nil
+end
+
+local function collect_child_result(n, child_node_ids, yield_result)
+    local output_data, collect_err = func._deps.child_output.collect_outputs(n, child_node_ids, yield_result)
+    if collect_err then
+        if type(collect_err) == "table" then
+            return nil, collect_err
         end
+        return nil, "Failed to collect child outputs: " .. tostring(collect_err)
+    end
+
+    return build_child_output_result(output_data)
+end
 
+local function fail_child_collect(n, collect_err)
+    if type(collect_err) == "table" then
+        local status = collect_err.status or collect_err.message
+        return n:fail({
+            code = collect_err.code,
+            message = collect_err.message
+        }, status)
+    end
+
+    return n:fail({
+        code = ERROR_FUNCTION_EXECUTION_FAILED,
+        message = tostring(collect_err)
+    }, tostring(collect_err))
+end
+
+local function complete_from_children(n, child_node_ids, fallback_result)
+    local final_output, collect_err = func._deps.child_output.resume_children(n, child_node_ids, collect_child_result)
+    if collect_err then
+        return fail_child_collect(n, collect_err)
+    end
+    if final_output ~= nil then
+        return n:complete(final_output, "Function executed successfully")
+    end
+
+    return n:complete(fallback_result, "Function executed successfully")
+end
+
+-- Checkpoints child ids before yielding so recovery resumes the same children
+-- instead of re-running the function and creating duplicate children.
+local function finish_with_children(n, created_node_ids, fallback_result)
+    n:update_metadata({ func_pending = { child_node_ids = created_node_ids } })
+
+    local yield_result, yield_err = n:yield({ run_nodes = created_node_ids })
+    if yield_err then
+        return n:fail({
+            code = ERROR_FUNCTION_EXECUTION_FAILED,
+            message = "Control command execution failed: " .. yield_err
+        }, "Control command execution failed")
+    end
+
+    local final_output, collect_err = collect_child_result(n, created_node_ids, yield_result)
+    if collect_err then
+        return fail_child_collect(n, collect_err)
+    end
+    if final_output ~= nil then
         return n:complete(final_output, "Function executed successfully")
     end
 
@@ -173,11 +206,9 @@ local function run(args)
 
     local config = n:config()
 
-    -- Recovery: a prior attempt already created child nodes and yielded; resume those
-    -- children instead of re-running the function (avoids duplicate child creation).
     local pending = (n:metadata() or {}).func_pending
     if type(pending) == "table" and type(pending.child_node_ids) == "table" and #pending.child_node_ids > 0 then
-        return finish_with_children(n, pending.child_node_ids, {})
+        return complete_from_children(n, pending.child_node_ids, {})
     end
     local func_id = config.func_id
     if not func_id or func_id == "" then
diff --git a/src/node_test.lua b/src/node_test.lua
index 995646c..3ec6bcf 100644
--- a/src/node_test.lua
+++ b/src/node_test.lua
@@ -1313,6 +1313,37 @@ local function define_tests()
                 test.eq(data_commands, 2)
             end)
 
+            it("should preserve target-provided output data ids", function()
+                local fixed_data_id = "fixed-output-data-id"
+                local fixed_node, _err = node.new({
+                    node_id = "test-node-123",
+                    dataflow_id = "test-dataflow-456",
+                    node = {
+                        config = {
+                            data_targets = {
+                                { data_type = "output.result", key = "main", data_id = fixed_data_id }
+                            }
+                        }
+                    }
+                }, mock_deps)
+                test.not_nil(fixed_node)
+
+                local result = fixed_node:complete({ message = "success" })
+
+                test.is_true(result.success)
+                test.eq(result.data_ids[1], fixed_data_id)
+
+                local first_submit = captured_calls.commit_submit[1]
+                local data_command: any = nil
+                for _, cmd in ipairs(first_submit.commands) do
+                    if cmd.type == consts.COMMAND_TYPES.CREATE_DATA then
+                        data_command = cmd
+                    end
+                end
+                test.not_nil(data_command)
+                test.eq(data_command.payload.data_id, fixed_data_id)
+            end)
+
             it("should route errors via error_targets from config on fail", function()
                 test.not_nil(test_node)
                 local result = test_node:fail("Something went wrong")
@@ -1330,6 +1361,37 @@ local function define_tests()
                 test.eq(data_commands, 2)
             end)
 
+            it("should preserve target-provided error data ids", function()
+                local fixed_data_id = "fixed-error-data-id"
+                local fixed_node, _err = node.new({
+                    node_id = "test-node-123",
+                    dataflow_id = "test-dataflow-456",
+                    node = {
+                        config = {
+                            error_targets = {
+                                { data_type = "error.details", key = "error", data_id = fixed_data_id }
+                            }
+                        }
+                    }
+                }, mock_deps)
+                test.not_nil(fixed_node)
+
+                local result = fixed_node:fail("Something went wrong")
+
+                test.is_false(result.success)
+                test.eq(result.data_ids[1], fixed_data_id)
+
+                local first_submit = captured_calls.commit_submit[1]
+                local data_command: any = nil
+                for _, cmd in ipairs(first_submit.commands) do
+                    if cmd.type == consts.COMMAND_TYPES.CREATE_DATA then
+                        data_command = cmd
+                    end
+                end
+                test.not_nil(data_command)
+                test.eq(data_command.payload.data_id, fixed_data_id)
+            end)
+
             it("should handle complete without output content", function()
                 local test_node_no_targets, _err = node.new({
                     node_id = "test-node-123",
diff --git a/src/persist/data_reader.lua b/src/persist/data_reader.lua
index d1a9c1c..0925ae9 100644
--- a/src/persist/data_reader.lua
+++ b/src/persist/data_reader.lua
@@ -224,8 +224,8 @@ local function parse_json_metadata(metadata_str)
         return {}
     end
 
-    local success, parsed = pcall(json.decode, metadata_str)
-    if success then
+    local parsed, parse_err = json.decode(metadata_str)
+    if not parse_err then
         return parsed
     else
         return {}
diff --git a/src/runner/orchestrator.lua b/src/runner/orchestrator.lua
index c7a5a98..f1b128b 100644
--- a/src/runner/orchestrator.lua
+++ b/src/runner/orchestrator.lua
@@ -626,6 +626,12 @@ local function handle_process_event(state: any, event: any)
     local persist_result, persist_err = state.workflow_state:persist()
 
     if exit_info and exit_info.yield_complete then
+        if not load_startup_pending_commits(state) then
+            return false
+        end
+        if not process_pending_commits(state) then
+            return false
+        end
         return handle_satisfy_yield(state, {
             parent_id = exit_info.yield_complete.parent_id,
             yield_id = exit_info.yield_complete.yield_info.yield_id,
diff --git a/src/runner/workflow_state.lua b/src/runner/workflow_state.lua
index 5933181..ca73cfb 100644
--- a/src/runner/workflow_state.lua
+++ b/src/runner/workflow_state.lua
@@ -55,8 +55,8 @@ end
 
 local function decode_json_content(content)
     if type(content) == "string" then
-        local success, parsed = pcall(json.decode, content)
-        if success then
+        local parsed, parse_err = json.decode(content)
+        if not parse_err then
             return parsed
         end
         return nil
@@ -850,8 +850,8 @@ function methods:get_failed_node_errors()
                 local content = result.content or "Unknown error"
 
                 if result.content_type == "application/json" or result.content_type == consts.CONTENT_TYPE.JSON then
-                    local success, parsed = pcall(json.decode, content)
-                    if success and type(parsed) == "table" then
+                    local parsed, parse_err = json.decode(content)
+                    if not parse_err and type(parsed) == "table" then
                         if parsed.error and type(parsed.error) == "table" and parsed.error.message then
                             error_message = tostring(parsed.error.message)
                         elseif parsed.message then
@@ -1047,8 +1047,8 @@ function methods:track_yield(node_id, yield_info)
                     -- and handle_satisfy_yield see a table instead of coercing to {}
                     local content = sig.content
                     if type(content) == "string" then
-                        local ok, parsed = pcall(json.decode, content)
-                        if ok then content = parsed end
+                        local parsed, parse_err = json.decode(content)
+                        if not parse_err then content = parsed end
                     end
                     yield_info.signal_data = content
                 end