From 5b62efa1eeeab68a1dbb2219ec41968abeda196c Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Sat, 1 Aug 2026 09:48:02 -0400 Subject: [PATCH 1/3] fix(persist): order terminal lifecycle locks --- src/persist/activation_repo.lua | 21 +++++++++++++-------- src/persist/ops.lua | 22 ++++++++++++++++++---- src/persist/ops_test.lua | 31 +++++++++++++++++++++++++++++++ 3 files changed, 62 insertions(+), 12 deletions(-) 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 acf4c70..407a5ad 100644 --- a/src/persist/ops.lua +++ b/src/persist/ops.lua @@ -794,6 +794,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 @@ -928,10 +946,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 6ba6364..176cdd2 100644 --- a/src/persist/ops_test.lua +++ b/src/persist/ops_test.lua @@ -992,6 +992,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() From a3aacd42cc9d7122c6d3cc7e5a6e29a5fe9acc2f Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Fri, 7 Aug 2026 23:03:50 -0400 Subject: [PATCH 2/3] fix(agent): keep aggregate failure evidence derived from unhandled outcomes A tool.call or delegated child whose error the parent agent consumes as an observation now carries an error_observed declaration, written in the same durable commit as the observation itself. get_failed_node_errors skips declared-consumed failures, so a handled child error is never reported as the dataflow's failure cause; workflow status continues to derive only from terminal outcomes. Tests: - agent_tool_failure_test: a failing tool consumed by the agent leaves the aggregate completed and the agent node driven to its own terminal status; an unhandled agent failure is attributed to the agent node, not the consumed tool child - delegation_handler_test: a consumed delegation failure is declared on the child and excluded from workflow failure evidence --- src/node/agent/_index.yaml | 28 +++ src/node/agent/agent_tool_failure_test.lua | 209 +++++++++++++++++++++ src/node/agent/delegation_handler.lua | 15 ++ src/node/agent/delegation_handler_test.lua | 64 +++++++ src/node/agent/node.lua | 30 ++- src/node/agent/stub/recovery_generate.lua | 37 +++- src/node/agent/stub/recovery_tool.lua | 4 + src/runner/workflow_state.lua | 8 +- test/.wippy.yaml | 2 + 9 files changed, 385 insertions(+), 12 deletions(-) create mode 100644 src/node/agent/agent_tool_failure_test.lua 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..e662821 --- /dev/null +++ b/src/node/agent/agent_tool_failure_test.lua @@ -0,0 +1,209 @@ +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) + :with_node_types("tool.call") + :all() or {} + test.eq(#tool_nodes, 1, "one tool.call child node created") + test.eq(tool_nodes[1].status, consts.STATUS.COMPLETED_FAILURE, "tool.call child records the failure") + test.is_true(tool_nodes[1].metadata.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) + :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.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) + :with_nodes(workflow.node_id) + :all() or {} + test.eq(#agent_nodes, 1, "agent node row present") + test.eq(agent_nodes[1].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) + :with_node_types("tool.call") + :all() or {} + test.eq(#tool_nodes, 1, "one tool.call child node created") + test.eq(tool_nodes[1].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 aggregate_error = tostring(row.metadata and 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_nodes[1].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..23f2159 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, nil, { + 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, nil, { + 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/runner/workflow_state.lua b/src/runner/workflow_state.lua index dd0c2ef..75d143c 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 diff --git a/test/.wippy.yaml b/test/.wippy.yaml index 3690c3d..bc89865 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 From 7a9b9853d2da6d737496239b2969897349701283 Mon Sep 17 00:00:00 2001 From: Wolfy-J Date: Sat, 8 Aug 2026 00:18:19 -0400 Subject: [PATCH 3/3] fix(runner): persist completion as a generation-fenced batch head A failed transaction leaves its command batch queued for retry; appending COMPLETE_WORKFLOW to that retained batch violated the persist layer's batch-order contract, the rejected completion cost the run its owner, and the overseer terminalized a healthy run as failed. workflow_state:queue_completion inserts the completion command at the batch head so retained commands persist behind the generation fence in one transaction, applied only when the fence wins. On fence loss the orchestrator rebuilds workflow state from durable rows before rescheduling, so re-evaluation never acts on outcomes that were never persisted. Exit-path persist failures are logged; the batch remains queued for retry. Tests: orchestrator_completion_flush_test drives the real orchestrator, workflow state, scheduler and persistence with a simulated transaction abort; covers completion retry of the retained batch and durable re-derivation after losing the completion fence. --- src/node/agent/agent_tool_failure_test.lua | 26 +- src/node/agent/stub/recovery_generate.lua | 4 +- src/runner/_index.yaml | 27 ++ src/runner/orchestrator.lua | 56 +++- .../orchestrator_completion_flush_test.lua | 239 ++++++++++++++++++ .../orchestrator_process_event_test.lua | 1 + src/runner/orchestrator_test.lua | 1 + src/runner/workflow_state.lua | 9 + test/.wippy.yaml | 2 + 9 files changed, 338 insertions(+), 27 deletions(-) create mode 100644 src/runner/orchestrator_completion_flush_test.lua diff --git a/src/node/agent/agent_tool_failure_test.lua b/src/node/agent/agent_tool_failure_test.lua index e662821..83cde43 100644 --- a/src/node/agent/agent_tool_failure_test.lua +++ b/src/node/agent/agent_tool_failure_test.lua @@ -143,29 +143,31 @@ local function define_tests() 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) + 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") - test.eq(tool_nodes[1].status, consts.STATUS.COMPLETED_FAILURE, "tool.call child records the failure") - test.is_true(tool_nodes[1].metadata.has_error == true, "tool.call child carries has_error metadata") + 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) + 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.discriminator, "result.success", "agent completed successfully after observing the error") + 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) + 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].status, consts.STATUS.COMPLETED_SUCCESS, + 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. @@ -188,19 +190,21 @@ local function define_tests() 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) + 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") - test.eq(tool_nodes[1].status, consts.STATUS.COMPLETED_FAILURE, "tool.call child records the failure") + 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 aggregate_error = tostring(row.metadata and row.metadata.error or "") + 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_nodes[1].node_id, 1, true) == nil, + 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) diff --git a/src/node/agent/stub/recovery_generate.lua b/src/node/agent/stub/recovery_generate.lua index 23f2159..ad9e4c2 100644 --- a/src/node/agent/stub/recovery_generate.lua +++ b/src/node/agent/stub/recovery_generate.lua @@ -70,7 +70,7 @@ local function handler(contract_args) 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, nil, { + 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 @@ -80,7 +80,7 @@ local function handler(contract_args) 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, nil, { + 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 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 75d143c..63042b1 100644 --- a/src/runner/workflow_state.lua +++ b/src/runner/workflow_state.lua @@ -1446,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 bc89865..9db49e7 100644 --- a/test/.wippy.yaml +++ b/test/.wippy.yaml @@ -99,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