fix(runner): completion persists as its own batch; consumed child errors are not failure evidence

src/node/agent/_index.yaml

@@ -135,8 +135,10 @@ entries:
       client: userspace.dataflow:client
       node: userspace.dataflow:node
       node_reader: userspace.dataflow.persist:node_reader
+      data_reader: userspace.dataflow.persist:data_reader
       commit: userspace.dataflow.persist:commit
       consts: userspace.dataflow:consts
+      workflow_state: userspace.dataflow.runner:workflow_state
       test: wippy.test:test
     method: run_tests
 
@@ -303,6 +305,32 @@ entries:
       test: wippy.test:test
     method: run_tests
 
+  - name: agent_tool_failure_test
+    kind: function.lua
+    meta:
+      name: Agent Tool Failure Tests
+      type: test
+      comment: Aggregate dataflow status stays derived from the agent outcome when a tool child fails
+      group: Workflow / Agent Node
+      timeout: "120s"
+      tags:
+        - dataflow
+        - agent
+        - test
+    source: file://agent_tool_failure_test.lua
+    modules:
+      - time
+      - uuid
+    imports:
+      agent_consts: userspace.dataflow.node.agent:consts
+      client: userspace.dataflow:client
+      consts: userspace.dataflow:consts
+      data_reader: userspace.dataflow.persist:data_reader
+      dataflow_repo: userspace.dataflow.persist:dataflow_repo
+      node_reader: userspace.dataflow.persist:node_reader
+      test: wippy.test:test
+    method: run_tests
+
   - name: agent_checkpoint_test
     kind: function.lua
     meta:

src/node/agent/agent_tool_failure_test.lua

@@ -0,0 +1,213 @@
+local test = require("test")
+local uuid = require("uuid")
+local time = require("time")
+local client = require("client")
+local consts = require("consts")
+local agent_consts = require("agent_consts")
+local data_reader = require("data_reader")
+local node_reader = require("node_reader")
+local dataflow_repo = require("dataflow_repo")
+
+local function define_tests()
+    describe("Agent Tool Failure Aggregate Status", function()
+        local c
+
+        before_all(function()
+            c = client.new()
+            test.not_nil(c, "client created")
+        end)
+
+        local function wait_until(predicate, timeout_ms, interval_ms)
+            local timeout = timeout_ms or 25000
+            local interval = interval_ms or 100
+            local attempts = math.ceil(timeout / interval)
+
+            for _ = 1, attempts do
+                local ok, value = pcall(predicate)
+                if ok and value ~= nil then
+                    return value
+                end
+                time.sleep(tostring(interval) .. "ms")
+            end
+
+            return nil
+        end
+
+        local function wait_terminal(df_id, timeout_ms)
+            return wait_until(function()
+                local status = c:get_status(df_id)
+                if status == consts.STATUS.COMPLETED_SUCCESS or
+                    status == consts.STATUS.COMPLETED_FAILURE or
+                    status == consts.STATUS.CANCELLED or
+                    status == consts.STATUS.TERMINATED then
+                    return status
+                end
+                return nil
+            end, timeout_ms or 25000, 100)
+        end
+
+        local function create_failing_tool_workflow(fail_message, mode)
+            local node_id = uuid.v7()
+            local input_id = uuid.v7()
+            local node_input_id = uuid.v7()
+            local scenario_id = "agent-tool-failure-" .. uuid.v7()
+
+            local commands = {
+                {
+                    type = consts.COMMAND_TYPES.CREATE_NODE,
+                    payload = {
+                        node_id = node_id,
+                        node_type = "userspace.dataflow.node.agent:node",
+                        status = consts.STATUS.PENDING,
+                        config = {
+                            agent = "userspace.dataflow.node.agent.stub:recovery_test_agent",
+                            arena = {
+                                prompt = "Execute the failing tool scenario.",
+                                max_iterations = 4,
+                                tool_calling = "auto",
+                                tools = {
+                                    "userspace.dataflow.node.agent.stub:recovery_tool"
+                                }
+                            },
+                            data_targets = {
+                                {
+                                    data_type = consts.DATA_TYPE.WORKFLOW_OUTPUT,
+                                    key = "result",
+                                    content_type = consts.CONTENT_TYPE.TEXT
+                                }
+                            }
+                        },
+                        metadata = {
+                            title = "Agent Tool Failure Test"
+                        }
+                    }
+                },
+                {
+                    type = consts.COMMAND_TYPES.CREATE_DATA,
+                    payload = {
+                        data_id = input_id,
+                        data_type = consts.DATA_TYPE.WORKFLOW_INPUT,
+                        content = {
+                            scenario_id = scenario_id,
+                            mode = mode or "failing_tool_then_final",
+                            fail_message = fail_message
+                        },
+                        content_type = consts.CONTENT_TYPE.JSON
+                    }
+                },
+                {
+                    type = consts.COMMAND_TYPES.CREATE_DATA,
+                    payload = {
+                        data_id = node_input_id,
+                        data_type = consts.DATA_TYPE.NODE_INPUT,
+                        node_id = node_id,
+                        key = input_id,
+                        content = "",
+                        content_type = consts.CONTENT_TYPE.REFERENCE
+                    }
+                }
+            }
+
+            local dataflow_id, err = c:create_workflow(commands, {
+                metadata = { title = "Agent Tool Failure Test Workflow" }
+            })
+            test.is_nil(err, "workflow created")
+
+            return {
+                dataflow_id = dataflow_id,
+                node_id = node_id,
+                scenario_id = scenario_id
+            }
+        end
+
+        it("keeps the aggregate status derived from the agent outcome when a tool call fails", function()
+            local fail_message = "Page returned status 403"
+            local workflow = create_failing_tool_workflow(fail_message)
+
+            c:start(workflow.dataflow_id)
+
+            local final_status = wait_terminal(workflow.dataflow_id)
+            test.not_nil(final_status, "workflow reached a terminal status")
+
+            -- The tool error is delivered to the agent as an observation.
+            local observations = data_reader.with_dataflow(workflow.dataflow_id)
+                :with_nodes(workflow.node_id)
+                :with_data_types(agent_consts.DATA_TYPE.AGENT_OBSERVATION)
+                :all() or {}
+            local error_observation = nil
+            for _, row in ipairs(observations) do
+                if row.metadata and row.metadata.is_error == true then
+                    error_observation = row
+                end
+            end
+            test.not_nil(error_observation, "tool error observation recorded for the agent")
+
+            -- The tool.call child keeps per-node error visibility.
+            local tool_nodes = (node_reader.with_dataflow(workflow.dataflow_id) :: any)
+                :with_node_types("tool.call")
+                :all() or {}
+            test.eq(#tool_nodes, 1, "one tool.call child node created")
+            local tool_node = tool_nodes[1] :: any
+            test.eq(tool_node.status, consts.STATUS.COMPLETED_FAILURE, "tool.call child records the failure")
+            test.is_true((tool_node.metadata or {}).has_error == true, "tool.call child carries has_error metadata")
+
+            -- The agent consumed the error and finished its run.
+            local agent_result = (data_reader.with_dataflow(workflow.dataflow_id) :: any)
+                :with_nodes(workflow.node_id)
+                :with_data_types(consts.DATA_TYPE.NODE_RESULT)
+                :one()
+            test.not_nil(agent_result, "agent node produced a result")
+            test.eq((agent_result :: any).discriminator, "result.success",
+                "agent completed successfully after observing the error")
+
+            -- The engine keeps driving the agent past the tool-child error: the
+            -- agent node reaches its own terminal status instead of staying a
+            -- zombie 'running' row.
+            local agent_nodes = (node_reader.with_dataflow(workflow.dataflow_id) :: any)
+                :with_nodes(workflow.node_id)
+                :all() or {}
+            test.eq(#agent_nodes, 1, "agent node row present")
+            test.eq((agent_nodes[1] :: any).status, consts.STATUS.COMPLETED_SUCCESS,
+                "agent node is driven to completion after the tool-child error")
+
+            -- The terminal aggregate is backed by a true terminal outcome.
+            local output = data_reader.with_dataflow(workflow.dataflow_id)
+                :with_data_types(consts.DATA_TYPE.WORKFLOW_OUTPUT)
+                :one()
+            test.not_nil(output, "workflow output produced by the agent terminal outcome")
+
+            -- A handled tool failure must not flip the dataflow aggregate.
+            test.eq(final_status, consts.STATUS.COMPLETED_SUCCESS,
+                "aggregate status derives from the agent terminal outcome, not the failed tool child")
+        end)
+
+        it("attributes an unhandled agent failure to the agent, not the consumed tool child", function()
+            local workflow = create_failing_tool_workflow("Page returned status 403", "failing_tool_then_llm_error")
+
+            c:start(workflow.dataflow_id)
+
+            local final_status = wait_terminal(workflow.dataflow_id)
+            test.eq(final_status, consts.STATUS.COMPLETED_FAILURE,
+                "unhandled agent failure terminates the workflow as failed")
+
+            local tool_nodes = (node_reader.with_dataflow(workflow.dataflow_id) :: any)
+                :with_node_types("tool.call")
+                :all() or {}
+            test.eq(#tool_nodes, 1, "one tool.call child node created")
+            local tool_node = tool_nodes[1] :: any
+            test.eq(tool_node.status, consts.STATUS.COMPLETED_FAILURE, "tool.call child records the failure")
+
+            local row, row_err = dataflow_repo.get(workflow.dataflow_id)
+            test.is_nil(row_err, "dataflow row loaded")
+            local row_metadata = (row :: any).metadata or {}
+            local aggregate_error = tostring(row_metadata.error or "")
+            test.is_true(aggregate_error ~= "", "aggregate failure carries error details")
+            test.is_true(string.find(aggregate_error, workflow.node_id, 1, true) ~= nil,
+                "aggregate failure names the agent node")
+            test.is_true(string.find(aggregate_error, tool_node.node_id, 1, true) == nil,
+                "aggregate failure does not blame the tool child whose error the agent consumed")
+        end)
+    end)
+end
+
+return test.run_cases(define_tests)

src/node/agent/delegation_handler.lua

@@ -399,6 +399,21 @@ function delegation_handler.map_delegation_results_to_conversation(delegation_re
                     is_error = true
                 }
             })
+
+            if info.child_id then
+                -- This update and the error observation above flush in one
+                -- durable commit; the declaration exists only together with the
+                -- delivered observation. A consumed child failure is not
+                -- workflow-terminal evidence; the consuming parent's own
+                -- outcome is.
+                parent_node_sdk:command({
+                    type = "UPDATE_NODE",
+                    payload = {
+                        node_id = info.child_id,
+                        metadata = { error_observed = true }
+                    }
+                })
+            end
         end
     end
 end

src/node/agent/delegation_handler_test.lua

@@ -4,8 +4,10 @@ local client = require("client")
 local node = require("node")
 local consts = require("consts")
 local node_reader = require("node_reader")
+local data_reader = require("data_reader")
 local commit = require("commit")
 local delegation_handler = require("delegation_handler")
+local workflow_state = require("workflow_state")
 
 -- Applies the node's queued commands synchronously (the orchestrator is not running
 -- in this test), mirroring the control_handler integration tests.
@@ -131,6 +133,68 @@ local function define_tests()
             test.eq(#rows, 1)
             test.eq((rows[1].metadata or {}).title, "Researcher")
         end)
+
+        it("declares a consumed delegation failure on the child and excludes it from failure evidence", function()
+            local n, dataflow_id, parent_node_id = setup_parent()
+            local session_context = { dataflow_id = dataflow_id, node_id = parent_node_id }
+            local delegation = {
+                agent_id = "researcher",
+                tool_call_id = "call-failed",
+                input_data = { task = "investigate" },
+                delegate_tool_name = "to_researcher"
+            }
+
+            local info = delegation_handler.create_child_node(n, delegation, 1, session_context)
+            apply(n, dataflow_id)
+
+            -- Child run ends in failure.
+            local _, update_err = commit.execute(dataflow_id, uuid.v7(), {
+                {
+                    type = consts.COMMAND_TYPES.UPDATE_NODE,
+                    payload = {
+                        node_id = info.child_id,
+                        status = consts.STATUS.COMPLETED_FAILURE
+                    }
+                }
+            }, { publish = false })
+            test.is_nil(update_err, "child failure persisted")
+
+            -- Parent consumes the failure as an observation.
+            delegation_handler.map_delegation_results_to_conversation({
+                {
+                    success = false,
+                    error = "delegated agent failed",
+                    delegation_info = info
+                }
+            }, n, 1)
+            apply(n, dataflow_id)
+
+            local observations = (data_reader.with_dataflow(dataflow_id) :: any)
+                :with_nodes(parent_node_id)
+                :with_data_types("agent.observation")
+                :all() or {}
+            local error_observation = nil
+            for _, row in ipairs(observations) do
+                if (row.metadata or {}).is_error == true then
+                    error_observation = row
+                end
+            end
+            test.not_nil(error_observation, "delegation error recorded as parent observation")
+
+            local child_rows = (node_reader.with_dataflow(dataflow_id) :: any)
+                :with_nodes(info.child_id)
+                :all() or {}
+            test.eq(#child_rows, 1)
+            test.eq(child_rows[1].status, consts.STATUS.COMPLETED_FAILURE, "child keeps its failure status")
+            test.is_true((child_rows[1].metadata or {}).error_observed == true,
+                "consumed failure declared on the child")
+
+            local ws = workflow_state.new(dataflow_id) :: any
+            local _, load_err = ws:load_state()
+            test.is_nil(load_err, "workflow state loaded")
+            test.is_nil(ws:get_failed_node_errors(),
+                "consumed delegation failure is not workflow failure evidence")
+        end)
     end)
 end