diff --git a/src/node/agent/_index.yaml b/src/node/agent/_index.yaml index e1420d2..b5121d1 100644 --- a/src/node/agent/_index.yaml +++ b/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: diff --git a/src/node/agent/agent_tool_failure_test.lua b/src/node/agent/agent_tool_failure_test.lua new file mode 100644 index 0000000..83cde43 --- /dev/null +++ b/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) diff --git a/src/node/agent/delegation_handler.lua b/src/node/agent/delegation_handler.lua index 6dea3e5..661738a 100644 --- a/src/node/agent/delegation_handler.lua +++ b/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 diff --git a/src/node/agent/delegation_handler_test.lua b/src/node/agent/delegation_handler_test.lua index f7d7331..1834d56 100644 --- a/src/node/agent/delegation_handler_test.lua +++ b/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 diff --git a/src/node/agent/node.lua b/src/node/agent/node.lua index 766133b..919b3da 100644 --- a/src/node/agent/node.lua +++ b/src/node/agent/node.lua @@ -1409,7 +1409,7 @@ local function configure_tool_wrappers(caller, agent_instance, n, agent_id, mode end local function process_tool_results(n, tool_results, iteration, exit_tool_name, agent_result: any, arena_config, - session_context) + session_context, tool_call_to_node_id) local control_responses = {} local control_delegations = {} local task_complete = false @@ -1505,6 +1505,22 @@ local function process_tool_results(n, tool_results, iteration, exit_tool_name, is_error = tool_error ~= nil } }) + + local viz_node_id = tool_error ~= nil and tool_call_to_node_id and + tool_call_to_node_id[call_id] or nil + if viz_node_id then + -- Declared alongside the observation just recorded; both + -- flush in one durable commit. A consumed child failure + -- is not workflow-terminal evidence; the consuming + -- parent's own outcome is. + n:command({ + type = consts.COMMAND_TYPES.UPDATE_NODE, + payload = { + node_id = viz_node_id, + metadata = { error_observed = true } + } + }) + end end end end @@ -1576,7 +1592,8 @@ local function check_completion(tool_calling, agent_result: any, iteration, min_ end local function finalize_iteration(n, agent_ctx, session_context, iteration, max_iterations, min_iterations, tool_calling, - exit_tool_name, agent_result: any, delegate_calls: any, tool_results, arena_config) + exit_tool_name, agent_result: any, delegate_calls: any, tool_results, arena_config, + tool_call_to_node_id) local control_responses, control_delegations, task_complete, final_result = process_tool_results( n, tool_results, @@ -1584,7 +1601,8 @@ local function finalize_iteration(n, agent_ctx, session_context, iteration, max_ exit_tool_name, agent_result, arena_config, - session_context + session_context, + tool_call_to_node_id ) append_control_delegations(delegate_calls, control_delegations) @@ -1718,7 +1736,8 @@ local function recover_persisted_action(n, agent_ctx, agent_instance, caller, se recovered_agent_result, delegate_calls, tool_results, - config.arena + config.arena, + tool_call_to_node_id ) if finalize_err then @@ -2259,7 +2278,8 @@ local function run(args) }, delegate_calls, tool_results, - config.arena + config.arena, + tool_call_to_node_id ) if finalize_err then if type(finalize_err) == "table" then diff --git a/src/node/agent/stub/recovery_generate.lua b/src/node/agent/stub/recovery_generate.lua index 7875b7d..ad9e4c2 100644 --- a/src/node/agent/stub/recovery_generate.lua +++ b/src/node/agent/stub/recovery_generate.lua @@ -9,7 +9,16 @@ local function response_tokens(prompt_tokens, completion_tokens) } end -local function tool_call_response(scenario_id, step, delay_ms, prompt_tokens, completion_tokens, tool_name) +local function tool_call_response(scenario_id, step, delay_ms, prompt_tokens, completion_tokens, tool_name, extra_args) + local arguments = { + scenario_id = scenario_id, + step = step, + delay_ms = delay_ms or 0 + } + for key, value in pairs(extra_args or {}) do + arguments[key] = value + end + return { success = true, result = { @@ -18,11 +27,7 @@ local function tool_call_response(scenario_id, step, delay_ms, prompt_tokens, co { id = helpers.call_id(scenario_id, step), name = tool_name or "recovery_tool", - arguments = { - scenario_id = scenario_id, - step = step, - delay_ms = delay_ms or 0 - } + arguments = arguments } } }, @@ -63,6 +68,26 @@ local function handler(contract_args) -- prompt token count above the checkpoint threshold deterministically local base_prompt = tonumber(scenario.prompt_tokens) or nil + if scenario.mode == "failing_tool_then_final" then + if result_count == 0 then + return tool_call_response(scenario.scenario_id, 1, scenario.tool_delay_ms, base_prompt or 13, 8, "recovery_tool", { + fail_message = scenario.fail_message or "Page returned status 403" + }) + end + + return final_response(scenario.scenario_id, scenario.mode, result_count, base_prompt or 9, 4) + end + + if scenario.mode == "failing_tool_then_llm_error" then + if result_count == 0 then + return tool_call_response(scenario.scenario_id, 1, scenario.tool_delay_ms, base_prompt or 13, 8, "recovery_tool", { + fail_message = scenario.fail_message or "Page returned status 403" + }) + end + + return nil, "recovery provider unavailable" + end + if scenario.mode == "single_tool_then_final" then if result_count == 0 then return tool_call_response(scenario.scenario_id, 1, scenario.tool_delay_ms, base_prompt or 13, 8) diff --git a/src/node/agent/stub/recovery_tool.lua b/src/node/agent/stub/recovery_tool.lua index 9c6fe19..0627935 100644 --- a/src/node/agent/stub/recovery_tool.lua +++ b/src/node/agent/stub/recovery_tool.lua @@ -26,6 +26,10 @@ local function handler(input) time.sleep(tostring(delay_ms) .. "ms") end + if type(input.fail_message) == "string" and input.fail_message ~= "" then + return nil, input.fail_message + end + return { ok = true, scenario_id = scenario_id, diff --git a/src/persist/activation_repo.lua b/src/persist/activation_repo.lua index 67669a4..626d822 100644 --- a/src/persist/activation_repo.lua +++ b/src/persist/activation_repo.lua @@ -156,7 +156,12 @@ local function normalize_row(row: any) }, nil end -local function lock_workflow_status_tx(tx, dataflow_id) +-- The workflow row is the first lock in every transaction that also mutates +-- activation or wake rows. PostgreSQL foreign-key checks can hold KEY SHARE on +-- this parent row, so acquiring a weaker UPDATE lock and upgrading it later can +-- deadlock with a concurrent commit. Callers that cross the workflow/lifecycle +-- boundary must establish this lock before either side is changed. +function activation_repo.lock_workflow_tx(tx, dataflow_id) local db_type, type_err = tx:db_type() if type_err then return nil, type_err end if db_type ~= sql.type.POSTGRES and db_type ~= "postgres" then @@ -271,7 +276,7 @@ function activation_repo.request_activation_tx(tx, dataflow_id, launch_args, now if not valid then return nil, id_err end valid, id_err = validate_timestamp(now_value, "requested_at") if not valid then return nil, id_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then return terminal, nil end @@ -290,7 +295,7 @@ function activation_repo.activate_for_signal_tx(tx, dataflow_id, wake_key, wake_ valid, validation_err = validate_timestamp(now_value, "requested_at") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -348,7 +353,7 @@ function activation_repo.activate_due_tx(tx, dataflow_id, wake_key, now_value) valid, validation_err = validate_timestamp(now_value, "now") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -423,7 +428,7 @@ function activation_repo.release_if_generation_tx(tx, dataflow_id, generation, n valid, validation_err = validate_timestamp(now_value, "updated_at") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -477,7 +482,7 @@ function activation_repo.claim_epoch_tx( valid, validation_err = validate_timestamp(now_value, "updated_at") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -515,7 +520,7 @@ function activation_repo.consume_wake_tx(tx, dataflow_id, wake_key, generation) if not valid then return nil, validation_err end if type(wake_key) ~= "string" or wake_key == "" then return nil, "wake_key is required" end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end local terminal = terminal_result_from_status(status) if terminal then @@ -568,7 +573,7 @@ function activation_repo.disable_terminal_tx(tx, dataflow_id, now_value) if not valid then return nil, validation_err end valid, validation_err = validate_timestamp(now_value, "updated_at") if not valid then return nil, validation_err end - local status, status_err = lock_workflow_status_tx(tx, dataflow_id) + local status, status_err = activation_repo.lock_workflow_tx(tx, dataflow_id) if status_err then return nil, status_err end if not TERMINAL_STATUS[status] then return nil, "dataflow is not terminal" end return cleanup_terminal_tx(tx, dataflow_id, status, now_value) diff --git a/src/persist/ops.lua b/src/persist/ops.lua index 409aa54..d049403 100644 --- a/src/persist/ops.lua +++ b/src/persist/ops.lua @@ -952,6 +952,24 @@ handlers[constants.COMMAND_TYPES.UPDATE_WORKFLOW] = function(tx, dataflow_id, op local payload = command.payload or {} local wf_id_to_update = payload.dataflow_id or dataflow_id + local terminal = payload.status == constants.STATUS.COMPLETED_SUCCESS or + payload.status == constants.STATUS.COMPLETED_FAILURE or + payload.status == constants.STATUS.CANCELLED or + payload.status == constants.STATUS.TERMINATED + + -- A terminal update crosses from the workflow row into activation and wake + -- rows. Establish the canonical parent-first lock order before UPDATE takes + -- PostgreSQL's weaker NO KEY UPDATE lock; upgrading that lock afterwards can + -- deadlock with a concurrent commit holding a foreign-key KEY SHARE lock. + if terminal then + local _, lock_err = activation_repo.lock_workflow_tx(tx, wf_id_to_update) + if lock_err then + if tostring(lock_err) == "dataflow not found" then + return nil, "Workflow not found or no changes applied" + end + return nil, "Failed to lock workflow lifecycle: " .. tostring(lock_err) + end + end -- Metadata merge configuration - default is merge=true for consistency local merge_metadata = payload.merge_metadata @@ -1086,10 +1104,6 @@ handlers[constants.COMMAND_TYPES.UPDATE_WORKFLOW] = function(tx, dataflow_id, op return nil, "Workflow not found or no changes applied" end - local terminal = payload.status == constants.STATUS.COMPLETED_SUCCESS or - payload.status == constants.STATUS.COMPLETED_FAILURE or - payload.status == constants.STATUS.CANCELLED or - payload.status == constants.STATUS.TERMINATED if terminal then local _, projection_err = activation_repo.disable_terminal_tx(tx, wf_id_to_update, now_ts) if projection_err then diff --git a/src/persist/ops_test.lua b/src/persist/ops_test.lua index 066b8c6..f0fcb50 100644 --- a/src/persist/ops_test.lua +++ b/src/persist/ops_test.lua @@ -1201,6 +1201,37 @@ local function define_tests() test.is_false(db_bool(activations[1].desired_active)) end) + it("locks the workflow before a terminal status update", function() + local resources = setup_test_resources() + local tx = get_test_transaction() + local observed_statuses = {} + local original_lock = activation_repo.lock_workflow_tx + activation_repo.lock_workflow_tx = function(lock_tx, dataflow_id) + local rows, query_err = txq(lock_tx, + "SELECT status FROM dataflows WHERE dataflow_id = ?", + { dataflow_id }) + if query_err then return nil, query_err end + observed_statuses[#observed_statuses + 1] = rows[1].status + return original_lock(lock_tx, dataflow_id) + end + + local execute_result + local execute_err + local called, call_err = pcall(function() + execute_result, execute_err = ops.execute(tx, resources.dataflow_id, nil, { + type = ops.COMMAND_TYPES.UPDATE_WORKFLOW, + payload = { status = ops.STATUS.CANCELLED }, + }) + end) + activation_repo.lock_workflow_tx = original_lock + if not called then error(call_err) end + + test.is_nil(execute_err) + test.not_nil(execute_result) + test.eq(observed_statuses[1], "active") + test.eq(observed_statuses[2], ops.STATUS.CANCELLED) + end) + it("rejects stale completion after a newer signal activation", function() local resources = setup_test_resources() local tx = get_test_transaction() diff --git a/src/runner/_index.yaml b/src/runner/_index.yaml index 4d62b6c..b1e1dd3 100644 --- a/src/runner/_index.yaml +++ b/src/runner/_index.yaml @@ -66,6 +66,33 @@ entries: test: wippy.test:test method: run_tests + - name: orchestrator_completion_flush_test + kind: function.lua + meta: + name: Orchestrator Completion Flush Tests + type: test + comment: Completion forms a valid fenced batch after a failed transaction leaves commands queued for retry + group: Workflow + timeout: "60s" + tags: + - dataflow + - orchestrator + - test + source: file://orchestrator_completion_flush_test.lua + modules: + - uuid + imports: + client: userspace.dataflow:client + commit: userspace.dataflow.persist:commit + consts: userspace.dataflow:consts + execution_frame: userspace.dataflow:execution_frame + node_reader: userspace.dataflow.persist:node_reader + orchestrator: userspace.dataflow.runner:orchestrator + scheduler: userspace.dataflow.runner:scheduler + test: wippy.test:test + workflow_state: userspace.dataflow.runner:workflow_state + method: run_tests + - name: orchestrator_process_event_test kind: function.lua meta: diff --git a/src/runner/orchestrator.lua b/src/runner/orchestrator.lua index 53fcff0..e2230ea 100644 --- a/src/runner/orchestrator.lua +++ b/src/runner/orchestrator.lua @@ -654,18 +654,18 @@ function handle_complete_workflow(state: OrchestratorState, payload: any) end end - local commands = { - { - type = consts.COMMAND_TYPES.COMPLETE_WORKFLOW, - payload = { - activation_generation = state.activation_generation, - status = final_status, - metadata = { error = not success and detailed_error or nil } - } + -- A failed transaction leaves its batch queued for retry. The completion + -- command is queued at the batch head so retained commands persist behind + -- the generation fence in one transaction, and are dropped with it when a + -- newer activation owns the workflow. + state.workflow_state:queue_completion({ + type = consts.COMMAND_TYPES.COMPLETE_WORKFLOW, + payload = { + activation_generation = state.activation_generation, + status = final_status, + metadata = { error = not success and detailed_error or nil } } - } - - state.workflow_state:queue_commands(commands) + }) local persist_result, persist_err = state.workflow_state:persist() if persist_err then @@ -693,8 +693,27 @@ function handle_complete_workflow(state: OrchestratorState, payload: any) state.running = false return false, false end - -- The completion decision was made against an older activation. Let the - -- scheduler reload and re-evaluate the newer durable work immediately. + -- The lost fence dropped this batch durably, including any commands a + -- failed transaction had left queued. Rebuild from durable state so the + -- re-evaluation cannot act on in-memory outcomes that never persisted; + -- completion is only decided without live node processes, so a reload + -- orphans nothing. + local fresh_state, fresh_err = state.runtime.workflow_state.new(state.dataflow_id) + local loaded = nil + if fresh_state then + loaded, fresh_err = fresh_state:load_state() + end + if not loaded then + state.exit_result = { + success = false, + dataflow_id = state.dataflow_id, + error = "Failed to reload workflow state after losing the completion fence: " .. + tostring(fresh_err), + } + state.running = false + return false, false + end + state.workflow_state = fresh_state state.reschedule_requested = true return true, true end @@ -1053,7 +1072,16 @@ local function handle_process_event(state: OrchestratorState, event: any) end local exit_info = state.workflow_state:handle_process_exit(from_pid, success, terminal_result) - local persist_result, persist_err = state.workflow_state:persist() + local _persist_result, persist_err = state.workflow_state:persist() + if persist_err then + -- The failed batch stays queued; the next persist on this state + -- retries it as part of its own transaction. + logger:warn("process exit persistence failed; batch retained for retry", { + dataflow_id = state.dataflow_id, + node_id = node_id, + error = tostring(persist_err), + }) + end if exit_info and exit_info.yield_complete then local completed_yield = exit_info.yield_complete diff --git a/src/runner/orchestrator_completion_flush_test.lua b/src/runner/orchestrator_completion_flush_test.lua new file mode 100644 index 0000000..28e0cb0 --- /dev/null +++ b/src/runner/orchestrator_completion_flush_test.lua @@ -0,0 +1,239 @@ +local test = require("test") +local uuid = require("uuid") +local orchestrator = require("orchestrator") +local workflow_state = require("workflow_state") +local scheduler = require("scheduler") +local commit = require("commit") +local client = require("client") +local consts = require("consts") +local node_reader = require("node_reader") +local execution_frame = require("execution_frame") + +-- Drives the real orchestrator loop with the real workflow state, scheduler and +-- persistence layer. Only the process/channel seams are scripted, and a single +-- transaction abort is simulated for the exit batch: per the workflow-state +-- contract a failed transaction leaves its batch queued for retry, and the +-- completion path must still form a valid generation-fenced batch. +local function define_tests() + describe("Orchestrator completion after a failed exit-batch transaction", function() + local function create_probe_workflow(c) + local node_id = uuid.v7() + local input_id = uuid.v7() + local dataflow_id, create_err = (c :: any):create_workflow({ + { + type = consts.COMMAND_TYPES.CREATE_NODE, + payload = { + node_id = node_id, + node_type = "test_node", + status = consts.STATUS.PENDING, + config = {}, + metadata = { title = "Completion flush probe" } + } + }, + { + type = consts.COMMAND_TYPES.CREATE_DATA, + payload = { + data_id = input_id, + data_type = consts.DATA_TYPE.NODE_INPUT, + node_id = node_id, + key = "default", + content = { probe = true }, + content_type = consts.CONTENT_TYPE.JSON + } + } + }) + test.is_nil(create_err, "workflow created") + return dataflow_id, node_id + end + + -- probes is a shared table: closure upvalue reassignment is not + -- observable across the runtime boundary here; table mutation is. + local function build_runtime(dataflow_id, node_id, probes, on_first_event: any) + local runtime: any = { + workflow_state = { + new = function(id: string): (any?, string?) + local ws, ws_err = workflow_state.new(id) + if not ws then return nil, ws_err end + local ws_any = (ws :: any) :: { [string]: any } + local real_persist = ws_any["persist"] + -- One transaction abort for the first batch carrying the + -- node's terminal update. The batch stays queued exactly + -- as a rolled-back transaction leaves it. + ws_any["persist"] = function(self: any): (any?, string?) + if not probes.aborted then + for _, cmd in ipairs(self.queued_commands) do + if cmd.type == consts.COMMAND_TYPES.UPDATE_NODE and + (cmd.payload or {}).node_id == node_id and + (cmd.payload or {}).status == consts.STATUS.COMPLETED_SUCCESS then + probes.aborted = true + return nil, "Failed to persist commands: simulated transaction abort" + end + end + end + return real_persist(self) + end + return ws, nil + end, + }, + scheduler = scheduler, + commit = commit, + activation_repo = { + get = function(): (any, nil) + return { generation = probes.generation, desired_active = true }, nil + end, + }, + execution_frame = execution_frame, + wake_repo = { remove = function(): (boolean, nil) return true, nil end }, + overseer = { notify = function(): (boolean, nil) return true, nil end }, + funcs = { + new = function(): any + local executor: any = {} + executor.with_actor = function(self: any): any return self end + executor.with_scope = function(self: any): any return self end + executor.call = function(): (any, nil) return {}, nil end + return executor + end, + }, + } + + local inbox = { case_receive = function(): any return { channel = "inbox" } end } + local events = { case_receive = function(): any return { channel = "events" } end } + runtime.process = { + registry = { + lookup = function(): (string?, any?) return nil, "not_found: name not registered" end, + register = function(): (boolean, nil) return true, nil end, + unregister = function() end, + }, + pid = function(): string return "orchestrator-pid" end, + set_options = function() end, + send = function(): (boolean, nil) return true, nil end, + terminate = function() end, + with_context = function(): any + local spawner: any = {} + spawner.with_actor = function(self: any): any return self end + spawner.with_scope = function(self: any): any return self end + spawner.spawn_linked_monitored = function(): (string, nil) return "child-pid", nil end + return spawner + end, + inbox = function(): any return inbox end, + events = function(): any return events end, + event = { EXIT = "pid.exit", LINK_DOWN = "pid.link.down", CANCEL = "pid.cancel" }, + } + + local function child_exit_event(): any + -- The node process routes its workflow output durably and then + -- exits, the production interleaving for a clean run. + local _, submit_err = commit.submit(dataflow_id, uuid.v7(), { + { + type = consts.COMMAND_TYPES.CREATE_DATA, + payload = { + data_id = uuid.v7(), + data_type = consts.DATA_TYPE.WORKFLOW_OUTPUT, + content = { done = true }, + content_type = consts.CONTENT_TYPE.JSON, + discriminator = "result", + node_id = node_id + } + } + }) + test.is_nil(submit_err, "node output submitted") + return { + ok = true, + channel = events, + value = { + kind = "pid.exit", + from = "child-pid", + result = { value = { success = true, message = "done", data_ids = {} } }, + }, + } + end + + runtime.channel = { + select = function(): any + probes.selects = (probes.selects or 0) + 1 + if probes.selects == 1 then + if on_first_event then on_first_event() end + return child_exit_event() + end + if probes.selects == 2 and probes.redeliver_exit then + return child_exit_event() + end + return { ok = false } + end, + } + + return runtime + end + + local function request_generation(dataflow_id) + local activation, activation_err = commit.request_activation(dataflow_id, {}, { notify = false }) + test.is_nil(activation_err, "activation requested") + local generation = tonumber((activation :: any).generation) + test.not_nil(generation, "activation generation available") + return generation + end + + local function assert_completed(c, dataflow_id, node_id) + local status, status_err = (c :: any):get_status(dataflow_id) + test.is_nil(status_err, "status readable") + test.eq(status, consts.STATUS.COMPLETED_SUCCESS, "workflow terminal status persisted") + + local rows = (node_reader.with_dataflow(dataflow_id) :: any) + :with_nodes(node_id) + :all() or {} + test.eq(#rows, 1, "node row present") + test.eq((rows[1] :: any).status, consts.STATUS.COMPLETED_SUCCESS, + "node terminal status persisted") + end + + it("retries the retained batch and persists completion as its own batch", function() + local c, client_err = client.new() + test.is_nil(client_err, "client created") + + local dataflow_id, node_id = create_probe_workflow(c) + local probes: any = { aborted = false } + probes.generation = request_generation(dataflow_id) + + local runtime = build_runtime(dataflow_id, node_id, probes, nil) + local result = orchestrator.run({ + dataflow_id = dataflow_id, + activation_generation = probes.generation, + }, runtime) :: any + + test.is_true(probes.aborted, "exit-batch transaction abort was exercised") + test.is_true(result.success, + "a retained batch does not break completion: " .. tostring(result.error)) + assert_completed(c, dataflow_id, node_id) + end) + + it("rebuilds from durable state when the retained batch loses the completion fence", function() + local c, client_err = client.new() + test.is_nil(client_err, "client created") + + local dataflow_id, node_id = create_probe_workflow(c) + local probes: any = { aborted = false, redeliver_exit = true } + probes.generation = request_generation(dataflow_id) + local stale_generation = probes.generation + + -- The durable generation advances between the exit and the + -- completion attempt, so the fenced batch is dropped and the run + -- must re-derive the node outcome from durable state. + local runtime = build_runtime(dataflow_id, node_id, probes, function() + probes.generation = request_generation(dataflow_id) + end) + + local result = orchestrator.run({ + dataflow_id = dataflow_id, + activation_generation = stale_generation, + }, runtime) :: any + + test.is_true(probes.aborted, "exit-batch transaction abort was exercised") + test.is_true(probes.generation > stale_generation, "durable generation advanced mid-run") + test.is_true(result.success, + "a dropped fenced batch does not terminalize unpersisted state: " .. tostring(result.error)) + assert_completed(c, dataflow_id, node_id) + end) + end) +end + +return test.run_cases(define_tests) diff --git a/src/runner/orchestrator_process_event_test.lua b/src/runner/orchestrator_process_event_test.lua index f936238..3841b50 100644 --- a/src/runner/orchestrator_process_event_test.lua +++ b/src/runner/orchestrator_process_event_test.lua @@ -34,6 +34,7 @@ local function define_tests() track_process = function(_self: any, _node_id: string, _pid: string) end, queue_commands = function(_self: any, _commands: any) end, + queue_completion = function(_self: any, _command: any) end, persist = function(_self: any): ({ changes_made: boolean }?, string?) if pending_node_result ~= nil then persisted_node_result = pending_node_result diff --git a/src/runner/orchestrator_test.lua b/src/runner/orchestrator_test.lua index 369a071..2470697 100644 --- a/src/runner/orchestrator_test.lua +++ b/src/runner/orchestrator_test.lua @@ -50,6 +50,7 @@ local function harness(options: HarnessOptions?): any workflow_state.get_failed_node_errors = function(): string? return cfg.failed_node_errors end workflow_state.track_process = function(self: any): any return self end workflow_state.queue_commands = function(self: any): any return self end + workflow_state.queue_completion = function(self: any): any return self end workflow_state.discard_queued_commands = function(self: any): any return self end workflow_state.get_node = function(): any return { type = "test_node", status = consts.STATUS.PENDING } end workflow_state.handle_process_exit = function(): string? return nil end diff --git a/src/runner/workflow_state.lua b/src/runner/workflow_state.lua index dd0c2ef..63042b1 100644 --- a/src/runner/workflow_state.lua +++ b/src/runner/workflow_state.lua @@ -973,7 +973,13 @@ function methods:get_failed_node_errors() local failed_nodes = {} for node_id, node_data in pairs(self.nodes) do if node_data.status == consts.STATUS.COMPLETED_FAILURE then - table.insert(failed_nodes, node_id) + local metadata = type(node_data.metadata) == "table" and node_data.metadata or {} + -- A failure whose error was consumed by its parent (declared via + -- error_observed) is handled; only unhandled failures are + -- workflow-terminal evidence. + if metadata.error_observed ~= true then + table.insert(failed_nodes, node_id) + end end end @@ -1440,6 +1446,15 @@ function methods:discard_queued_commands() return self end +-- COMPLETE_WORKFLOW is a generation-fenced batch precondition: the persist +-- layer applies it first and applies the rest of the batch only when the fence +-- wins. Queue the completion at the head so commands retained by an earlier +-- failed transaction ride behind the fence in the same transaction. +function methods:queue_completion(command) + table.insert(self.queued_commands, 1, command) + return self +end + function methods:persist() if #self.queued_commands == 0 then return { changes_made = false, message = "No commands to persist" }, nil diff --git a/test/.wippy.yaml b/test/.wippy.yaml index 3690c3d..9db49e7 100644 --- a/test/.wippy.yaml +++ b/test/.wippy.yaml @@ -45,6 +45,8 @@ override: "userspace.dataflow.node.agent.run_context:run_context_test:security.policies": [app:test_policy] "userspace.dataflow.node.agent:agent_checkpoint_test:security.actor.id": dataflow.test "userspace.dataflow.node.agent:agent_checkpoint_test:security.policies": [app:test_policy] + "userspace.dataflow.node.agent:agent_tool_failure_test:security.actor.id": dataflow.test + "userspace.dataflow.node.agent:agent_tool_failure_test:security.policies": [app:test_policy] "userspace.dataflow.node.agent:agent_truncation_real_test:security.actor.id": dataflow.test "userspace.dataflow.node.agent:agent_truncation_real_test:security.policies": [app:test_policy] "userspace.dataflow.node.agent:agent_truncation_test:security.actor.id": dataflow.test @@ -97,6 +99,8 @@ override: "userspace.dataflow.persist:ops_test:security.policies": [app:test_policy] "userspace.dataflow.persist:wake_repo_test:security.actor.id": dataflow.test "userspace.dataflow.persist:wake_repo_test:security.policies": [app:test_policy] + "userspace.dataflow.runner:orchestrator_completion_flush_test:security.actor.id": dataflow.test + "userspace.dataflow.runner:orchestrator_completion_flush_test:security.policies": [app:test_policy] "userspace.dataflow.runner:orchestrator_process_event_test:security.actor.id": dataflow.test "userspace.dataflow.runner:orchestrator_process_event_test:security.policies": [app:test_policy] "userspace.dataflow.runner:orchestrator_test:security.actor.id": dataflow.test