mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-11-15 07:41:20 +00:00
9cb2bde557
Bundled examples keep their behavior tests in backend/tests/, so the core backend workflow ran optional extension behavior on every pull request. Mark those suites with pytest.mark.bundled_extension, exclude them from the core shards and from the duration baseline, and run them in a dedicated workflow through make test-bundled-extensions. Tests of host mechanisms stay in the core shards even when an example package is the payload. The marked suites import deerflow.agents, config, runtime, knowledge and app.gateway, so the workflow triggers for backend/packages/**, backend/app/gateway/** and the marked test modules; a backend test pins those paths, the make target and both marker selections, and fails when a marked module is missing from the path list. Host/extension compatibility checks — a host API gate and authorization of contributed routes — live in test_extension_host_contracts.py so they still fail core CI. The committed duration baseline drops the 482 entries for the excluded suites. Refs #8173 Co-authored-by: REI-330 <173389910+REI-330@users.noreply.github.com> Co-authored-by: willem-bd <143703838+willem-bd@users.noreply.github.com> Co-authored-by: Eilen6316 <136898293+Eilen6316@users.noreply.github.com>
1165 lines
50 KiB
Python
1165 lines
50 KiB
Python
"""Script workflows through the public extension loader and an isolated JS worker."""
|
|
|
|
import asyncio
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import pytest_asyncio
|
|
from deerflow_extension_api import AgentRun, AgentRunError, ExtensionPrincipal
|
|
from deerflow_extension_api.plugins import ActionContext
|
|
from test_extension_agent_runs import host as _host_fixture
|
|
from test_extension_agent_runs import runtime as _runtime_fixture
|
|
|
|
from deerflow.extensions.loader import ExtensionSpec, load_extensions
|
|
|
|
pytestmark = [
|
|
pytest.mark.bundled_extension,
|
|
pytest.mark.skipif(os.name != "posix", reason="Workflow worker execution requires POSIX"),
|
|
]
|
|
|
|
ROOT = Path(__file__).resolve().parents[2]
|
|
|
|
|
|
class Runs:
|
|
def __init__(self):
|
|
self.threads = {}
|
|
self.runs = {}
|
|
self.starts = []
|
|
self.active = 0
|
|
self.peak = 0
|
|
self.entered = asyncio.Event()
|
|
self.release = asyncio.Event()
|
|
self.admission = asyncio.Event()
|
|
self.return_admission = asyncio.Event()
|
|
self.hold_admission = False
|
|
self.jobs = []
|
|
self.denied = False
|
|
|
|
revision = "initial"
|
|
|
|
async def configuration_evidence(self, *, thread_id, run_id):
|
|
return {"run_id": run_id, "thread_id": thread_id, "source": "host_assembly", "version": "synthetic-assembly", "model": "test-model"}
|
|
|
|
async def read_artifact(self, *, thread_id, path):
|
|
return self.file_content
|
|
|
|
file_content = b"original file"
|
|
|
|
async def configuration_revision(self, *, assistant_ids, durable=False):
|
|
if self.denied:
|
|
raise AgentRunError(403, "revoked")
|
|
return self.revision
|
|
|
|
async def create_thread(self, *, assistant_id, thread_id, metadata):
|
|
if self.denied:
|
|
raise AgentRunError(403, "revoked")
|
|
self.threads.setdefault(
|
|
thread_id,
|
|
{
|
|
"values": {"messages": []},
|
|
"next": [],
|
|
"interrupts": [],
|
|
"assistant": assistant_id,
|
|
},
|
|
)
|
|
return thread_id
|
|
|
|
async def start(self, *, thread_id, input, idempotency_key):
|
|
if self.denied:
|
|
raise AgentRunError(403, "revoked")
|
|
if idempotency_key in self.runs:
|
|
return self.runs[idempotency_key]
|
|
run = AgentRun(thread_id, idempotency_key, "running", self.threads[thread_id]["assistant"])
|
|
self.runs[run.run_id] = run
|
|
self.starts.append((thread_id, input, idempotency_key))
|
|
self.threads[thread_id]["values"]["messages"].extend(input["messages"])
|
|
self.jobs.append(asyncio.create_task(self.answer(run, input)))
|
|
self.admission.set()
|
|
if self.hold_admission:
|
|
await self.return_admission.wait()
|
|
return run
|
|
|
|
async def answer(self, run, input):
|
|
task = json.loads(input["messages"][0]["content"])
|
|
self.active += 1
|
|
self.peak = max(self.peak, self.active)
|
|
self.entered.set()
|
|
try:
|
|
if task == "blocked":
|
|
await self.release.wait()
|
|
else:
|
|
await asyncio.sleep(0.02)
|
|
if self.runs[run.run_id].status != "running":
|
|
return
|
|
state = self.threads[run.thread_id]
|
|
if task == "interrupt":
|
|
state["next"] = ["tools"]
|
|
state["interrupts"] = [{"value": "Approve?"}]
|
|
elif task == "clarification":
|
|
state["values"]["messages"].append(
|
|
{
|
|
"type": "tool",
|
|
"artifact": {
|
|
"human_input": {
|
|
"kind": "human_input_request",
|
|
"version": 2,
|
|
"request_id": "req",
|
|
"source": "ask_clarification",
|
|
"question": "Confirm?",
|
|
}
|
|
},
|
|
}
|
|
)
|
|
else:
|
|
state["values"]["messages"].append({"type": "ai", "content": json.dumps({"echo": task})})
|
|
self.runs[run.run_id] = AgentRun(run.thread_id, run.run_id, "error" if task == "fail" else "success")
|
|
finally:
|
|
self.active -= 1
|
|
|
|
async def get(self, *, thread_id, run_id):
|
|
if self.denied:
|
|
raise AgentRunError(403, "revoked")
|
|
return self.runs[run_id]
|
|
|
|
async def wait(self, *, thread_id, run_id, timeout=60):
|
|
async with asyncio.timeout(timeout):
|
|
while (await self.get(thread_id=thread_id, run_id=run_id)).status == "running":
|
|
await asyncio.sleep(0.002)
|
|
return await self.get(thread_id=thread_id, run_id=run_id)
|
|
|
|
async def get_state(self, *, thread_id):
|
|
return self.threads[thread_id]
|
|
|
|
async def cancel(self, *, thread_id, run_id):
|
|
self.runs[run_id] = AgentRun(thread_id, run_id, "interrupted")
|
|
self.release.set()
|
|
|
|
async def close(self):
|
|
self.release.set()
|
|
self.return_admission.set()
|
|
await asyncio.gather(*self.jobs, return_exceptions=True)
|
|
|
|
|
|
@pytest_asyncio.fixture
|
|
async def plugin(tmp_path, monkeypatch):
|
|
monkeypatch.syspath_prepend(str(ROOT / "examples/deerflow-extension-workflows"))
|
|
loaded, diagnostics = await asyncio.to_thread(
|
|
load_extensions,
|
|
[
|
|
ExtensionSpec(
|
|
use="deerflow_extension_workflows:install",
|
|
config={
|
|
"enabled": True,
|
|
"storage_path": str(tmp_path / "workflows.sqlite"),
|
|
"agents": ["lead_agent", "reviewer"],
|
|
"max_calls": 8,
|
|
"concurrency": 2,
|
|
"timeout_seconds": 5,
|
|
},
|
|
)
|
|
],
|
|
)
|
|
assert not diagnostics
|
|
((_, declaration),) = loaded.plugins
|
|
actions = {a.name: a.handler for a in declaration.backend}
|
|
service = actions["launch"].__self__
|
|
await service.start(None)
|
|
runs = Runs()
|
|
yield loaded, actions, service, runs
|
|
await service.stop()
|
|
await runs.close()
|
|
|
|
|
|
def context(runs, owner="alice"):
|
|
return ActionContext(ExtensionPrincipal(owner), {"enabled": True}, agent_runs=runs)
|
|
|
|
|
|
async def launch(plugin, script, **extra):
|
|
_, actions, _, runs = plugin
|
|
return await actions["launch"]({"request_id": "test", "script": script, "args": {}, **extra}, context(runs))
|
|
|
|
|
|
async def done(plugin, run_id):
|
|
_, actions, service, runs = plugin
|
|
await asyncio.wait_for(service.wait_idle(run_id), 10)
|
|
return await actions["get"]({"run_id": run_id}, context(runs))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_sequential_parallel_branch_loop_structured_results(plugin):
|
|
result = await launch(
|
|
plugin,
|
|
"""
|
|
const a = await agent({stepId:'a',agentRef:'reviewer',task:args.product});
|
|
const b = await parallel([0,1,2,3].map(i => () => agent({stepId:'p'+i,agentRef:'lead_agent',task:{i,a:a.output}})));
|
|
const c=[]; if(b.length===4) for(let i=0;i<2;i++) c.push(await agent({stepId:'loop'+i,agentRef:'lead_agent',task:b[i].output}));
|
|
log('Checks finished'); return {a:a.output,b:b.map(x=>x.output),c};
|
|
""",
|
|
args={"product": "synthetic"},
|
|
)
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "completed", final
|
|
assert final["result"]["a"] == {"echo": "synthetic"}
|
|
assert [x["echo"]["i"] for x in final["result"]["b"]] == [0, 1, 2, 3]
|
|
assert plugin[3].peak == 2
|
|
assert len(final["steps"]) == 7
|
|
assert all(s["attempt"] == 1 and s["thread_id"] and s["run_id"] for s in final["steps"])
|
|
assert final["script_hash"] and final["args_hash"]
|
|
assert final["logs"] == ["Checks finished"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("material", ["x" * 9000, "材料" * 750, "x" * 11998], ids=["ascii-report", "chinese-report", "maximum-output"])
|
|
async def test_step_output_can_be_handed_to_next_agent_without_truncation(plugin, material):
|
|
runs = plugin[3]
|
|
original_answer = runs.answer
|
|
|
|
async def answer(run, input):
|
|
await original_answer(run, input)
|
|
task = json.loads(input["messages"][0]["content"])
|
|
content = material if task == "prepare" else {"received": True}
|
|
runs.threads[run.thread_id]["values"]["messages"][-1]["content"] = json.dumps(content)
|
|
|
|
runs.answer = answer
|
|
launched = await launch(
|
|
plugin,
|
|
"""
|
|
const first = await agent({stepId:'prepare',agentRef:'lead_agent',task:'prepare'});
|
|
const second = await agent({stepId:'produce',agentRef:'lead_agent',task:{material:first.output,instruction:'Use the supplied material'}});
|
|
return second.output;
|
|
""",
|
|
)
|
|
final = await done(plugin, launched["run_id"])
|
|
assert final["status"] == "completed", final
|
|
assert final["result"] == {"received": True}
|
|
assert len(runs.starts) == 2
|
|
assert json.loads(runs.starts[1][1]["messages"][0]["content"]) == {"material": material, "instruction": "Use the supplied material"}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_oversized_agent_task_is_rejected_before_child_admission(plugin):
|
|
launched = await launch(plugin, "return await agent({stepId:'large',agentRef:'lead_agent',task:'x'.repeat(16001)});")
|
|
final = await done(plugin, launched["run_id"])
|
|
assert final["status"] == "failed"
|
|
assert not plugin[3].starts
|
|
assert not plugin[2].processes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_background_return_idempotency_and_owner_scope(plugin):
|
|
script = "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});"
|
|
first = await launch(plugin, script)
|
|
await asyncio.wait_for(plugin[3].entered.wait(), 5)
|
|
retry = await launch(plugin, script)
|
|
assert retry["run_id"] == first["run_id"] and len(plugin[3].starts) == 1
|
|
with pytest.raises(ValueError):
|
|
await launch(plugin, "return 2;")
|
|
for action in ("get", "cancel", "delete"):
|
|
with pytest.raises(ValueError):
|
|
await plugin[1][action]({"run_id": first["run_id"]}, context(plugin[3], "bob"))
|
|
assert (await plugin[1]["list"]({}, context(plugin[3], "bob")))["runs"] == []
|
|
await plugin[1]["cancel"]({"run_id": first["run_id"]}, context(plugin[3]))
|
|
assert (await done(plugin, first["run_id"]))["status"] == "cancelled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_during_admission_tracks_then_cancels_child(plugin):
|
|
plugin[3].hold_admission = True
|
|
result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});")
|
|
await asyncio.wait_for(plugin[3].admission.wait(), 5)
|
|
await plugin[1]["cancel"]({"run_id": result["run_id"]}, context(plugin[3]))
|
|
# Response is deliberately delayed until after cancellation was requested.
|
|
plugin[3].return_admission.set()
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "cancelled"
|
|
assert final["steps"][0]["run_id"]
|
|
assert all(r.status == "interrupted" for r in plugin[3].runs.values())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"script",
|
|
[
|
|
"return process.env;",
|
|
"return agent.constructor('return process')();",
|
|
"return require('fs');",
|
|
"return fetch('https://example.com');",
|
|
"while(true) {}",
|
|
"const x=[]; while(true) x.push('x'.repeat(100000));",
|
|
"return await new Promise(()=>{});",
|
|
"return await agent({stepId:'x',agentRef:'forbidden',task:1});",
|
|
"for(let i=0;i<9;i++) await agent({stepId:'x'+i,agentRef:'lead_agent',task:i});",
|
|
"await agent({stepId:'x',agentRef:'lead_agent',task:1}); return await agent({stepId:'x',agentRef:'lead_agent',task:2});",
|
|
],
|
|
)
|
|
async def test_limits_and_ambient_capabilities_fail_closed(plugin, script):
|
|
result = await launch(plugin, script)
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "failed", final
|
|
assert len(plugin[3].starts) <= 8
|
|
assert not plugin[2].processes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"task,status",
|
|
[
|
|
("interrupt", "waiting_input"),
|
|
("clarification", "waiting_input"),
|
|
("fail", "failed"),
|
|
],
|
|
)
|
|
async def test_interrupt_and_failure_do_not_dispatch_downstream(plugin, task, status):
|
|
result = await launch(
|
|
plugin,
|
|
"await agent({stepId:'a',agentRef:'lead_agent',task:args.task}); return await agent({stepId:'b',agentRef:'lead_agent',task:2});",
|
|
args={"task": task},
|
|
)
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == status
|
|
assert len(plugin[3].starts) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_result_attribution_refuses_external_turn(plugin):
|
|
original = plugin[3].get_state
|
|
|
|
async def advanced(**kwargs):
|
|
state = await original(**kwargs)
|
|
state["values"]["messages"].extend([{"type": "human", "content": "later"}, {"type": "ai", "content": "wrong"}])
|
|
return state
|
|
|
|
plugin[3].get_state = advanced
|
|
result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:1});")
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "failed" and final["steps"][0]["status"] == "result_conflict"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_restart_retains_records_without_replaying(plugin):
|
|
result = await launch(plugin, "return 42;")
|
|
final = await done(plugin, result["run_id"])
|
|
await plugin[2].stop()
|
|
await plugin[2].start(None)
|
|
again = await plugin[1]["get"]({"run_id": result["run_id"]}, context(plugin[3]))
|
|
assert again["result"] == final["result"] == 42
|
|
assert not plugin[3].starts
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_uncertain_admission_is_retained_without_automatic_retry(plugin):
|
|
original = plugin[3].start
|
|
|
|
async def uncertain(**kwargs):
|
|
await original(**kwargs)
|
|
raise TimeoutError("Response lost after admission")
|
|
|
|
plugin[3].start = uncertain
|
|
script = "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});"
|
|
result = await launch(plugin, script)
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "needs_reconciliation"
|
|
assert final["steps"][0]["thread_id"] and final["steps"][0]["run_id"] is None
|
|
assert len(plugin[3].starts) == 1
|
|
assert final["steps"][0]["error"] == "Admission timed out; inspect the stable child thread, do not replay"
|
|
retry = await launch(plugin, script)
|
|
assert retry["run_id"] == result["run_id"] and len(plugin[3].starts) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"script",
|
|
[
|
|
"for(let i=0;i<21;i++)log('progress');return 1;",
|
|
"return 'x'.repeat(50000);",
|
|
"agent({stepId:'a',agentRef:'lead_agent',task:'blocked'}); return 'early';",
|
|
],
|
|
)
|
|
async def test_log_output_and_detached_call_limits(plugin, script):
|
|
result = await launch(plugin, script)
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "failed"
|
|
assert not plugin[2].processes
|
|
assert all(r.status != "running" for r in plugin[3].runs.values())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_service_shutdown_drains_children_and_worker(plugin):
|
|
result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});")
|
|
await asyncio.wait_for(plugin[3].entered.wait(), 5)
|
|
await plugin[2].stop()
|
|
assert not plugin[2].processes and not plugin[2].entries
|
|
assert all(r.status == "interrupted" for r in plugin[3].runs.values())
|
|
await plugin[2].start(None)
|
|
final = await plugin[1]["get"]({"run_id": result["run_id"]}, context(plugin[3]))
|
|
assert final["status"] == "needs_reconciliation"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_process_lock_prevents_multiple_services_recovering_live_work(plugin):
|
|
from deerflow_extension_workflows.service import Workflows
|
|
|
|
other = Workflows(plugin[2].config)
|
|
with pytest.raises(BlockingIOError):
|
|
await other.start(None)
|
|
assert other.available is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disabled_plugin_has_no_service_or_visible_tools(tmp_path, monkeypatch):
|
|
monkeypatch.syspath_prepend(str(ROOT / "examples/deerflow-extension-workflows"))
|
|
from deerflow.extensions.plugin_tools import build_plugin_tools
|
|
|
|
loaded, diagnostics = load_extensions([ExtensionSpec(use="deerflow_extension_workflows:install", config={"enabled": False})])
|
|
assert not diagnostics and not loaded.services and not build_plugin_tools(loaded)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_operator_configuration_rejects_unsupported_budget_and_limits(tmp_path, monkeypatch):
|
|
monkeypatch.syspath_prepend(str(ROOT / "examples/deerflow-extension-workflows"))
|
|
for extra in (
|
|
{"max_cost_usd": 1},
|
|
{"concurrency": 100},
|
|
{"max_calls": True},
|
|
{"agents": ["lead_agent", "lead_agent"]},
|
|
):
|
|
loaded, diagnostics = load_extensions(
|
|
[
|
|
ExtensionSpec(
|
|
use="deerflow_extension_workflows:install",
|
|
config={
|
|
"enabled": True,
|
|
"storage_path": str(tmp_path / "runs.sqlite"),
|
|
"agents": ["lead_agent"],
|
|
**extra,
|
|
},
|
|
)
|
|
]
|
|
)
|
|
assert diagnostics and not loaded.plugins and not loaded.services
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_real_gateway_lifecycle_bridge(plugin, gateway_runtime):
|
|
runs, *_ = gateway_runtime
|
|
result = await plugin[1]["launch"](
|
|
{
|
|
"request_id": "real",
|
|
"script": "const a=await agent({stepId:'a',agentRef:'lead_agent',task:'first'});return await agent({stepId:'b',agentRef:'lead_agent',task:a.output});",
|
|
"args": {},
|
|
},
|
|
context(runs),
|
|
)
|
|
await plugin[2].wait_idle(result["run_id"])
|
|
final = await plugin[1]["get"]({"run_id": result["run_id"]}, context(runs))
|
|
assert final["status"] == "completed", final
|
|
assert len(final["steps"]) == 2
|
|
for step in final["steps"]:
|
|
assert (await runs.get(thread_id=step["thread_id"], run_id=step["run_id"])).status == "success"
|
|
|
|
|
|
host = _host_fixture
|
|
gateway_runtime = _runtime_fixture
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_global_concurrency_and_owner_admission_caps(plugin):
|
|
service, runs = plugin[2], plugin[3]
|
|
service.global_slots = asyncio.Semaphore(1)
|
|
script = "return await agent({stepId:'x',agentRef:'lead_agent',task:'blocked'});"
|
|
first = await plugin[1]["launch"]({"request_id": "first", "script": script}, context(runs))
|
|
await asyncio.wait_for(runs.entered.wait(), 5)
|
|
second = await plugin[1]["launch"]({"request_id": "second", "script": script}, context(runs, "bob"))
|
|
# Wait until the second workflow has persisted a queued step, before asserting
|
|
# the shared slot prevented admission (no timing-only concurrency assertion).
|
|
async with asyncio.timeout(5):
|
|
while not (await plugin[1]["get"]({"run_id": second["run_id"]}, context(runs, "bob")))["steps"]:
|
|
await asyncio.sleep(0.002)
|
|
assert len(runs.starts) == 1
|
|
await plugin[1]["cancel"]({"run_id": second["run_id"]}, context(runs, "bob"))
|
|
await plugin[1]["cancel"]({"run_id": first["run_id"]}, context(runs))
|
|
await service.wait_idle(first["run_id"])
|
|
await service.wait_idle(second["run_id"])
|
|
assert len(runs.starts) == 1 and runs.peak == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_startup_marks_unfinished_records_interrupted_without_replay(plugin):
|
|
result = await launch(plugin, "return 1;")
|
|
await done(plugin, result["run_id"])
|
|
record = await asyncio.to_thread(plugin[2].store.get, "alice", result["run_id"])
|
|
record["status"] = "running"
|
|
record.pop("recovery_version", None)
|
|
await asyncio.to_thread(plugin[2].store.put, "alice", record)
|
|
await plugin[2].stop()
|
|
await plugin[2].start(None)
|
|
final = await plugin[1]["get"]({"run_id": result["run_id"]}, context(plugin[3]))
|
|
assert final["status"] == "interrupted"
|
|
assert not plugin[3].starts
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_model_tool_dispatch_uses_stamped_principal_and_handle(plugin):
|
|
from types import SimpleNamespace
|
|
|
|
from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY
|
|
|
|
from deerflow.extensions.plugin_tools import build_plugin_tools
|
|
|
|
# Exercise the production model-tool wrapper, not only the action handler.
|
|
class ScopedRuns(Runs):
|
|
def for_plugin(self, namespace):
|
|
assert namespace == "community.workflows"
|
|
return self
|
|
|
|
runs = ScopedRuns()
|
|
tools = build_plugin_tools(plugin[0])
|
|
tool = next(t for t in tools if "launch_workflow" in t.name)
|
|
runtime = SimpleNamespace(
|
|
context={
|
|
"thread_id": "parent",
|
|
AGENT_RUNS_CONTEXT_KEY: runs,
|
|
"user_id": "alice",
|
|
},
|
|
config={"configurable": {"user_id": "alice"}},
|
|
)
|
|
try:
|
|
reply = json.loads(await tool.coroutine(runtime=runtime, request_id="tool-run", script="return 7;", args={}))
|
|
await plugin[2].wait_idle(reply["run_id"])
|
|
final = await plugin[1]["get"]({"run_id": reply["run_id"]}, context(runs))
|
|
assert final["result"] == 7
|
|
finally:
|
|
await runs.close()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("with_child", [False, True])
|
|
async def test_workflow_deadline_is_failed_and_cleans_up(plugin, with_child):
|
|
service, runs = plugin[2], plugin[3]
|
|
service.config["timeout_seconds"] = 1
|
|
script = "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});" if with_child else "return await new Promise(()=>{});"
|
|
result = await launch(plugin, script)
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "failed", final
|
|
assert final["error"] == "Execution deadline exceeded"
|
|
assert not service.processes
|
|
if with_child:
|
|
assert len(runs.starts) == 1
|
|
assert final["steps"][0]["child_status"] == "interrupted"
|
|
assert all(run.status == "interrupted" for run in runs.runs.values())
|
|
|
|
|
|
def count_sends(service, monkeypatch):
|
|
sent = []
|
|
original = service.send
|
|
|
|
async def send(process, message):
|
|
sent.append(message)
|
|
await original(process, message)
|
|
|
|
monkeypatch.setattr(service, "send", send)
|
|
return sent
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_pending_agent_wait_does_not_consume_worker_turns(plugin, monkeypatch):
|
|
service, runs = plugin[2], plugin[3]
|
|
monkeypatch.setattr(sys.modules[type(service).__module__], "TURN_LIMIT", 20)
|
|
sent = count_sends(service, monkeypatch)
|
|
result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});")
|
|
await asyncio.wait_for(runs.entered.wait(), 5)
|
|
await asyncio.sleep(3)
|
|
runs.release.set()
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "completed", final
|
|
# Idle waiting must not round-trip to the worker every poll interval.
|
|
assert len(sent) <= 5, len(sent)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_worker_exit_while_agent_pending_fails_promptly(plugin):
|
|
_, actions, service, runs = plugin
|
|
service.config["timeout_seconds"] = 60
|
|
result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});")
|
|
await asyncio.wait_for(runs.entered.wait(), 5)
|
|
assert runs.active == 1
|
|
process = service.processes[result["run_id"]]
|
|
process.kill()
|
|
await asyncio.wait_for(process.wait(), 3)
|
|
|
|
# A dead idle worker must fail before the workflow's 60-second deadline,
|
|
# without waiting for the blocked Agent to complete by itself.
|
|
await asyncio.wait_for(service.wait_idle(result["run_id"]), 3)
|
|
final = await actions["get"]({"run_id": result["run_id"]}, context(runs))
|
|
assert final["status"] == "failed", final
|
|
assert final["error"] == "Worker exited while waiting for Agent calls"
|
|
assert final["steps"][0]["child_status"] == "interrupted"
|
|
assert all(run.status == "interrupted" for run in runs.runs.values())
|
|
assert not service.processes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_work_frames_still_count_against_worker_turn_budget(plugin, monkeypatch):
|
|
"""Work-carrying frames retain the turn cap (regression from PR #7025)."""
|
|
service = plugin[2]
|
|
monkeypatch.setattr(sys.modules[type(service).__module__], "TURN_LIMIT", 2)
|
|
result = await launch(
|
|
plugin,
|
|
"""
|
|
await agent({stepId:'a',agentRef:'lead_agent',task:'one'});
|
|
await agent({stepId:'b',agentRef:'lead_agent',task:'two'});
|
|
return 1;
|
|
""",
|
|
)
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "failed", final
|
|
assert final["error"] == "Worker turn limit"
|
|
assert not service.processes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cancel_while_agent_pending_stops_promptly(plugin):
|
|
_, actions, service, runs = plugin
|
|
result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});")
|
|
await asyncio.wait_for(runs.entered.wait(), 5)
|
|
await asyncio.sleep(0.3)
|
|
started = time.monotonic()
|
|
await actions["cancel"]({"run_id": result["run_id"]}, context(runs))
|
|
# Cancel latency is ~4 ms locally; the budget also covers the store writes
|
|
# and process reaping, which can spike on a loaded CI runner.
|
|
await asyncio.wait_for(service.wait_idle(result["run_id"]), 3)
|
|
assert time.monotonic() - started < 3
|
|
final = await actions["get"]({"run_id": result["run_id"]}, context(runs))
|
|
assert final["status"] == "cancelled", final
|
|
assert not service.processes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"script",
|
|
[
|
|
"for(;;) await null;",
|
|
"const p=agent({stepId:'a',agentRef:'lead_agent',task:'blocked'}); for(;;) await null;",
|
|
],
|
|
)
|
|
async def test_spinning_script_still_hits_turn_limit(plugin, monkeypatch, script):
|
|
service = plugin[2]
|
|
# Exercise the frame cap without racing 50 durable frame commits against
|
|
# the fixture's short wall deadline on loaded CI runners.
|
|
monkeypatch.setattr(sys.modules[type(service).__module__], "TURN_LIMIT", 3)
|
|
monkeypatch.setitem(service.config, "timeout_seconds", 30)
|
|
result = await launch(plugin, script)
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "failed", final
|
|
assert final["error"] == "Worker turn limit"
|
|
assert not service.processes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_external_supervisor_cancellation_is_interrupted(plugin):
|
|
service, runs = plugin[2], plugin[3]
|
|
result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});")
|
|
await asyncio.wait_for(runs.entered.wait(), 5)
|
|
service.workers[result["run_id"]].cancel()
|
|
final = await done(plugin, result["run_id"])
|
|
assert final["status"] == "interrupted", final
|
|
assert final["error"] == "Host interrupted; child cleanup required"
|
|
assert final["steps"][0]["child_status"] == "interrupted"
|
|
assert not service.processes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("operation", ["get", "get_state"])
|
|
@pytest.mark.parametrize("cleanup_fails", [False, True])
|
|
async def test_post_admission_timeout_reports_observation_and_cleans_up(plugin, operation, cleanup_fails):
|
|
service, runs = plugin[2], plugin[3]
|
|
|
|
async def timeout_once(**kwargs):
|
|
setattr(runs, operation, original)
|
|
if cleanup_fails:
|
|
runs.denied = True
|
|
raise TimeoutError("Slow child observation")
|
|
|
|
original = getattr(runs, operation)
|
|
setattr(runs, operation, timeout_once)
|
|
result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:args.task});", args={"task": "blocked" if operation == "get" else 1})
|
|
final = await done(plugin, result["run_id"])
|
|
step = final["steps"][0]
|
|
assert step["run_id"]
|
|
assert step["error"] == "Child observation timed out; interrupting tracked run"
|
|
assert not service.processes
|
|
if cleanup_fails:
|
|
assert final["status"] == step["status"] == "needs_reconciliation"
|
|
else:
|
|
assert final["status"] == step["status"] == "failed"
|
|
assert final["error"] == step["error"]
|
|
assert step["child_status"] in ("success", "interrupted")
|
|
assert all(run.status != "running" for run in runs.runs.values())
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_idempotent_retry_survives_allowlist_reordering_and_restart(plugin):
|
|
service, runs = plugin[2], plugin[3]
|
|
script = "return 42;"
|
|
first = await launch(plugin, script)
|
|
await done(plugin, first["run_id"])
|
|
await service.stop()
|
|
service.config["agents"] = list(reversed(service.config["agents"]))
|
|
await service.start(None)
|
|
retry = await launch(plugin, script)
|
|
assert retry["run_id"] == first["run_id"]
|
|
assert retry["status"] == "completed"
|
|
assert not runs.starts
|
|
service.config["agents"].append("new_agent")
|
|
with pytest.raises(ValueError, match="request_id already used"):
|
|
await launch(plugin, script)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("injection,later_turn", [(False, False), (True, False), (True, True)])
|
|
async def test_result_anchor_uses_visible_input_after_real_context_rewrite(plugin, injection, later_turn):
|
|
from langchain_core.messages import AIMessage, HumanMessage
|
|
|
|
from app.gateway.services import normalize_input
|
|
from deerflow.agents.middlewares.dynamic_context_middleware import DynamicContextMiddleware
|
|
from deerflow.runtime.serialization import serialize_channel_values_for_api
|
|
|
|
original = plugin[3].get_state
|
|
|
|
async def injected(**kwargs):
|
|
state = await original(**kwargs)
|
|
raw = state["values"]["messages"][0]
|
|
user = normalize_input({"messages": [raw]})["messages"][0]
|
|
assert isinstance(user, HumanMessage)
|
|
messages = DynamicContextMiddleware._make_reminder_and_user_messages(user, "Synthetic context", "Synthetic memory") if injection else [user]
|
|
messages.append(AIMessage(content='{"answer":42}'))
|
|
if later_turn:
|
|
messages.extend([HumanMessage(content="Later user input"), AIMessage(content="Wrong answer")])
|
|
state["values"] = serialize_channel_values_for_api({"messages": messages})
|
|
return state
|
|
|
|
plugin[3].get_state = injected
|
|
result = await launch(plugin, "return await agent({stepId:'a',agentRef:'lead_agent',task:'first'});")
|
|
final = await done(plugin, result["run_id"])
|
|
if later_turn:
|
|
assert final["status"] == "failed" and final["steps"][0]["status"] == "result_conflict"
|
|
else:
|
|
assert final["status"] == "completed", final
|
|
assert final["result"]["output"] == {"answer": 42}
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("shutdown", [False, True])
|
|
async def test_cancelled_launch_caller_keeps_committed_sqlite_launch_owned(plugin, monkeypatch, shutdown):
|
|
import threading
|
|
|
|
service, runs = plugin[2], plugin[3]
|
|
committed, release = threading.Event(), threading.Event()
|
|
original = service.store.create
|
|
|
|
def committed_before_response(owner, record):
|
|
result = original(owner, record)
|
|
committed.set()
|
|
assert release.wait(5)
|
|
return result
|
|
|
|
monkeypatch.setattr(service.store, "create", committed_before_response)
|
|
script = "return await agent({stepId:'a',agentRef:'lead_agent',task:'blocked'});"
|
|
caller = asyncio.create_task(launch(plugin, script))
|
|
stopping = None
|
|
try:
|
|
assert await asyncio.to_thread(committed.wait, 5)
|
|
caller.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await caller
|
|
if shutdown:
|
|
stopping = asyncio.create_task(service.stop())
|
|
await asyncio.sleep(0)
|
|
release.set()
|
|
if shutdown:
|
|
await asyncio.wait_for(stopping, 10)
|
|
await service.start(None)
|
|
retry = await launch(plugin, script)
|
|
if not shutdown:
|
|
await asyncio.wait_for(runs.entered.wait(), 5)
|
|
await plugin[1]["cancel"]({"run_id": retry["run_id"]}, context(runs))
|
|
final = await done(plugin, retry["run_id"])
|
|
assert final["status"] in ({"recoverable", "needs_reconciliation"} if shutdown else {"cancelled"}), final
|
|
assert len(runs.starts) <= 1
|
|
assert not service.entries and not service.workers and not service.processes
|
|
await plugin[1]["delete"]({"run_id": retry["run_id"]}, context(runs))
|
|
finally:
|
|
release.set()
|
|
if stopping:
|
|
await stopping
|
|
|
|
|
|
async def waiting(plugin, run_id, status="waiting_approval"):
|
|
async with asyncio.timeout(5):
|
|
while True:
|
|
result = await plugin[1]["get"]({"run_id": run_id}, context(plugin[3]))
|
|
if result["status"] == status:
|
|
return result
|
|
await asyncio.sleep(0.01)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase2_definition_versions_are_immutable_and_owner_scoped(plugin):
|
|
_, actions, _, runs = plugin
|
|
first = await actions["save_definition"]({"name": "inspection", "script": "return args;"}, context(runs))
|
|
assert await actions["save_definition"]({"name": "inspection", "script": "return args;"}, context(runs)) == first
|
|
second = await actions["save_definition"]({"name": "inspection", "script": "return args.product;"}, context(runs))
|
|
assert first["version"] != second["version"]
|
|
launched = await actions["launch"]({"request_id": "versioned", "definition": first, "args": {"product": "A"}}, context(runs))
|
|
assert (await done(plugin, launched["run_id"]))["result"] == {"product": "A"}
|
|
with pytest.raises(ValueError):
|
|
await actions["launch"]({"request_id": "stolen", "definition": first}, context(runs, "bob"))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase2_checkpoint_requires_exact_authenticated_decision_and_dispatches_once(plugin):
|
|
from dataclasses import replace
|
|
|
|
from deerflow_extension_api.plugins import ToolContext
|
|
|
|
_, actions, _, runs = plugin
|
|
launched = await launch(
|
|
plugin,
|
|
"""
|
|
const evidence = await agent({stepId:'evidence',agentRef:'lead_agent',task:'evidence'});
|
|
await humanCheckpoint({stepId:'confirm',sourceStepId:'evidence',artifactVersion:evidence.artifactVersion});
|
|
return await agent({stepId:'report',agentRef:'lead_agent',task:evidence.output});
|
|
""",
|
|
)
|
|
run_id = launched["run_id"]
|
|
paused = await waiting(plugin, run_id)
|
|
checkpoint = paused["checkpoints"][0]
|
|
assert len(runs.starts) == 1
|
|
decision = {"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": checkpoint["artifact_version"], "decision": "approve"}
|
|
human = replace(context(runs), interaction="session")
|
|
for invalid in (context(runs), replace(human, principal=ExtensionPrincipal("bob")), ToolContext(human.principal, human.settings, "chat", agent_runs=runs, interaction="session")):
|
|
with pytest.raises(ValueError):
|
|
await actions["decide"](decision, invalid)
|
|
with pytest.raises(ValueError):
|
|
await actions["decide"]({**decision, "artifact_version": "stale"}, human)
|
|
await asyncio.gather(*(actions["decide"](decision, human) for _ in range(3)))
|
|
result = await done(plugin, run_id)
|
|
assert result["status"] == "completed"
|
|
assert len(runs.starts) == 2
|
|
assert result["checkpoints"][0]["decided_by"] == "alice"
|
|
with pytest.raises(ValueError):
|
|
await actions["decide"]({**decision, "decision": "reject"}, human)
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase2_pause_resume_preserves_results_and_invalidates_configuration(plugin):
|
|
_, actions, _, runs = plugin
|
|
launched = await launch(
|
|
plugin,
|
|
"""
|
|
const a = await agent({stepId:'a',agentRef:'lead_agent',task:'original'});
|
|
await pause({stepId:'break'});
|
|
return a.output;
|
|
""",
|
|
)
|
|
run_id = launched["run_id"]
|
|
paused = await waiting(plugin, run_id, "paused")
|
|
runs.threads[runs.starts[0][0]]["values"]["messages"].append({"type": "ai", "content": "changed later"})
|
|
with pytest.raises(ValueError):
|
|
await actions["resume"]({"run_id": run_id, "step_id": "break", "attempt": 1, "fingerprint": "changed"}, context(runs))
|
|
await actions["resume"]({"run_id": run_id, "step_id": "break", "attempt": 1, "fingerprint": paused["fingerprint"]}, context(runs))
|
|
assert (await done(plugin, run_id))["result"] == {"echo": "original"}
|
|
assert len(runs.starts) == 1
|
|
launched = await launch(plugin, "await pause({stepId:'break'}); return 1;", request_id="changed-config")
|
|
paused = await waiting(plugin, launched["run_id"], "paused")
|
|
runs.revision = "modified"
|
|
with pytest.raises(ValueError, match="configuration"):
|
|
await actions["resume"]({"run_id": launched["run_id"], "step_id": "break", "attempt": 1, "fingerprint": paused["fingerprint"]}, context(runs))
|
|
await actions["cancel"]({"run_id": launched["run_id"]}, context(runs))
|
|
assert (await done(plugin, launched["run_id"]))["status"] == "cancelled"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("kind", ["reject", "cancel", "expiry"])
|
|
async def test_phase2_checkpoint_stop_never_dispatches_downstream(plugin, kind):
|
|
from dataclasses import replace
|
|
|
|
_, actions, service, runs = plugin
|
|
if kind == "expiry":
|
|
service.config["pause_timeout_seconds"] = 1
|
|
launched = await launch(
|
|
plugin,
|
|
"const a=await agent({stepId:'a',agentRef:'lead_agent',task:'data'});await humanCheckpoint({stepId:'confirm',sourceStepId:'a',artifactVersion:a.artifactVersion});return await agent({stepId:'b',agentRef:'lead_agent',task:'never'});",
|
|
)
|
|
run_id = launched["run_id"]
|
|
paused = await waiting(plugin, run_id)
|
|
if kind == "reject":
|
|
await actions["decide"]({"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": paused["checkpoints"][0]["artifact_version"], "decision": "reject"}, replace(context(runs), interaction="session"))
|
|
elif kind == "cancel":
|
|
await actions["cancel"]({"run_id": run_id}, context(runs))
|
|
result = await done(plugin, run_id)
|
|
assert result["status"] == ("cancelled" if kind == "cancel" else "failed")
|
|
assert len(runs.starts) == 1
|
|
assert not service.processes
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase2_pause_does_not_consume_execution_budget_and_duplicate_resume_is_fenced(plugin):
|
|
_, actions, service, runs = plugin
|
|
service.config["timeout_seconds"] = 1
|
|
launched = await launch(plugin, "await pause({stepId:'first'});await pause({stepId:'second'});return 42;")
|
|
run_id = launched["run_id"]
|
|
first = await waiting(plugin, run_id, "paused")
|
|
await asyncio.sleep(1.1)
|
|
payload = {"run_id": run_id, "step_id": "first", "attempt": 1, "fingerprint": first["fingerprint"]}
|
|
await actions["resume"](payload, context(runs))
|
|
async with asyncio.timeout(5):
|
|
while True:
|
|
second = await actions["get"]({"run_id": run_id}, context(runs))
|
|
if len(second["checkpoints"]) == 2:
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
await actions["resume"](payload, context(runs))
|
|
assert (await actions["get"]({"run_id": run_id}, context(runs)))["checkpoints"][1]["status"] == "pending"
|
|
await actions["resume"]({**payload, "step_id": "second"}, context(runs))
|
|
assert (await done(plugin, run_id))["result"] == 42
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize(
|
|
"script",
|
|
[
|
|
"await humanCheckpoint({stepId:'c',sourceStepId:'fake',artifactVersion:'approved'});",
|
|
"pause({stepId:'p'}); await agent({stepId:'a',agentRef:'lead_agent',task:'forbidden'});",
|
|
"await pause({stepId:'p'});",
|
|
],
|
|
)
|
|
async def test_phase2_fail_closed_checkpoint_and_restart(plugin, script):
|
|
_, actions, service, runs = plugin
|
|
launched = await launch(plugin, script)
|
|
run_id = launched["run_id"]
|
|
if script == "await pause({stepId:'p'});":
|
|
paused = await waiting(plugin, run_id, "paused")
|
|
with pytest.raises(ValueError):
|
|
await actions["delete"]({"run_id": run_id}, context(runs))
|
|
await service.stop()
|
|
# Simulate a crash-retained record and a new Gateway instance.
|
|
record = await asyncio.to_thread(service.store.get, "alice", run_id)
|
|
record["status"] = "paused"
|
|
await asyncio.to_thread(service.store.put, "alice", record)
|
|
from deerflow_extension_workflows.service import Workflows
|
|
|
|
replacement = Workflows(service.config)
|
|
await replacement.start(None)
|
|
try:
|
|
assert (await replacement.get({"run_id": run_id}, context(runs)))["status"] == "recoverable"
|
|
with pytest.raises(ValueError):
|
|
await replacement.resume({"run_id": run_id, "step_id": "p", "attempt": 1, "fingerprint": paused["fingerprint"]}, context(runs))
|
|
finally:
|
|
await replacement.stop()
|
|
else:
|
|
assert (await done(plugin, run_id))["status"] == "failed"
|
|
assert not runs.starts
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase2_real_gateway_output_checkpoint(plugin, gateway_runtime):
|
|
from dataclasses import replace
|
|
|
|
runs, *_ = gateway_runtime
|
|
_, actions, service, _ = plugin
|
|
launched = await actions["launch"](
|
|
{"request_id": "real-approval", "script": "const a=await agent({stepId:'a',agentRef:'lead_agent',task:'first'});await humanCheckpoint({stepId:'confirm',sourceStepId:'a',artifactVersion:a.artifactVersion});return a.output;"},
|
|
context(runs),
|
|
)
|
|
run_id = launched["run_id"]
|
|
paused = await waiting(plugin, run_id)
|
|
retained = await actions["get_step"]({"run_id": run_id, "step_id": "a"}, context(runs))
|
|
assert retained["artifact_version"] == paused["checkpoints"][0]["artifact_version"]
|
|
await actions["decide"]({"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": retained["artifact_version"], "decision": "approve"}, replace(context(runs), interaction="session"))
|
|
await service.wait_idle(run_id)
|
|
assert (await actions["get"]({"run_id": run_id}, context(runs)))["result"] == retained["output"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase2_disconnected_decision_still_publishes_exactly_once(plugin, monkeypatch):
|
|
from dataclasses import replace
|
|
|
|
_, actions, service, runs = plugin
|
|
launched = await launch(plugin, "const a=await agent({stepId:'a',agentRef:'lead_agent',task:'x'});await humanCheckpoint({stepId:'confirm',sourceStepId:'a',artifactVersion:a.artifactVersion});return 42;")
|
|
run_id = launched["run_id"]
|
|
paused = await waiting(plugin, run_id)
|
|
decision = {"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": paused["checkpoints"][0]["artifact_version"], "decision": "approve"}
|
|
persisted, release = asyncio.Event(), asyncio.Event()
|
|
original = service.save
|
|
|
|
async def hold(owner, record):
|
|
await original(owner, record)
|
|
if record["status"] == "waiting_approval" and record["checkpoints"][0]["status"] == "approved":
|
|
persisted.set()
|
|
await release.wait()
|
|
|
|
monkeypatch.setattr(service, "save", hold)
|
|
caller = asyncio.create_task(actions["decide"](decision, replace(context(runs), interaction="session")))
|
|
await asyncio.wait_for(persisted.wait(), 5)
|
|
caller.cancel()
|
|
with pytest.raises(asyncio.CancelledError):
|
|
await caller
|
|
release.set()
|
|
assert (await done(plugin, run_id))["result"] == 42
|
|
assert len(runs.starts) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_phase2_inspection_example_reviews_each_stage_and_final_output(plugin, monkeypatch):
|
|
from dataclasses import replace
|
|
|
|
_, actions, _, runs = plugin
|
|
original = runs.answer
|
|
|
|
async def answer(run, input):
|
|
await original(run, input)
|
|
task = json.loads(input["messages"][0]["content"])
|
|
if isinstance(task, dict) and task.get("instruction", "").startswith("Return JSON with verdict"):
|
|
runs.threads[run.thread_id]["values"]["messages"][-1]["content"] = json.dumps({"verdict": "ready"})
|
|
|
|
monkeypatch.setattr(runs, "answer", answer)
|
|
script = (ROOT / "examples/deerflow-extension-workflows/examples/inspection.workflow.js").read_text(encoding="utf-8")
|
|
launched = await launch(
|
|
plugin, script, args={"complianceAgent": "lead_agent", "ingredientAgent": "lead_agent", "reportAgent": "lead_agent", "reviewerAgent": "lead_agent", "product": "synthetic", "referenceDate": "2026-10-15", "formula": []}
|
|
)
|
|
run_id = launched["run_id"]
|
|
for step_id, calls in [("confirm-evidence", 1), ("confirm-findings", 2), ("confirm-final-report", 4)]:
|
|
async with asyncio.timeout(5):
|
|
while True:
|
|
result = await actions["get"]({"run_id": run_id}, context(runs))
|
|
checkpoint = next((c for c in result["checkpoints"] if c["step_id"] == step_id), None)
|
|
if checkpoint and result["status"] == "waiting_approval":
|
|
break
|
|
await asyncio.sleep(0.01)
|
|
assert len(runs.starts) == calls
|
|
await actions["decide"]({"run_id": run_id, "step_id": step_id, "attempt": 1, "artifact_version": checkpoint["artifact_version"], "decision": "approve"}, replace(context(runs), interaction="session"))
|
|
result = await done(plugin, run_id)
|
|
assert result["status"] == "completed", result
|
|
assert result["result"]["status"] == "human_confirmed_report"
|
|
assert result["result"]["artifactVersion"] == result["checkpoints"][-1]["artifact_version"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_immutable_file_review_and_downstream_consume_same_version(plugin):
|
|
import base64
|
|
from dataclasses import replace
|
|
|
|
_, actions, _, runs = plugin
|
|
launched = await launch(
|
|
plugin,
|
|
"""
|
|
await agent({stepId:'report',agentRef:'lead_agent',task:'create file'});
|
|
const file=await captureArtifact({stepId:'file',sourceStepId:'report',path:'/mnt/user-data/outputs/report.pdf'});
|
|
await humanCheckpoint({stepId:'confirm-file',sourceStepId:'file',artifactVersion:file.artifactVersion});
|
|
return await readArtifact({artifactVersion:file.artifactVersion});
|
|
""",
|
|
)
|
|
run_id = launched["run_id"]
|
|
paused = await waiting(plugin, run_id)
|
|
file = paused["artifacts"][0]
|
|
assert file["sha256"] and file["source_run_id"]
|
|
with pytest.raises(ValueError, match="approved"):
|
|
await actions["read_artifact"]({"run_id": run_id, "version": file["version"]}, context(runs))
|
|
preview = await actions["get_artifact"]({"run_id": run_id, "version": file["version"]}, context(runs))
|
|
runs.file_content = b"overwritten source after preview"
|
|
decision = {"run_id": run_id, "step_id": "confirm-file", "attempt": 1, "artifact_version": file["version"], "decision": "approve"}
|
|
with pytest.raises(ValueError):
|
|
await actions["decide"]({**decision, "artifact_version": "different-version"}, replace(context(runs), interaction="session"))
|
|
await actions["decide"](decision, replace(context(runs), interaction="session"))
|
|
result = await done(plugin, run_id)
|
|
assert result["status"] == "completed", result
|
|
assert result["result"] == preview
|
|
assert base64.b64decode(result["result"]["content_base64"]) == b"original file"
|
|
source = await actions["get_step"]({"run_id": run_id, "step_id": "report"}, context(runs))
|
|
assert source["configuration_evidence"]["run_id"] == source["run_id"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_file_reference_requires_approval_before_child_admission(plugin):
|
|
launched = await launch(
|
|
plugin,
|
|
"await agent({stepId:'a',agentRef:'lead_agent',task:'file'});"
|
|
"const f=await captureArtifact({stepId:'f',sourceStepId:'a',path:'/mnt/user-data/outputs/report.pdf'});"
|
|
"return await agent({stepId:'b',agentRef:'lead_agent',task:'read',artifacts:[f.artifactVersion]});",
|
|
)
|
|
result = await done(plugin, launched["run_id"])
|
|
assert result["status"] == "failed"
|
|
assert len(plugin[3].starts) == 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_approved_file_reference_is_bound_to_child_input(plugin):
|
|
from dataclasses import replace
|
|
|
|
_, actions, _, runs = plugin
|
|
launched = await launch(
|
|
plugin,
|
|
"await agent({stepId:'a',agentRef:'lead_agent',task:'file'});"
|
|
"const f=await captureArtifact({stepId:'f',sourceStepId:'a',path:'/mnt/user-data/outputs/report.pdf'});"
|
|
"await humanCheckpoint({stepId:'ok',sourceStepId:'f',artifactVersion:f.artifactVersion});return await agent({stepId:'b',agentRef:'lead_agent',task:'read',artifacts:[f.artifactVersion]});",
|
|
)
|
|
run_id = launched["run_id"]
|
|
paused = await waiting(plugin, run_id)
|
|
version = paused["artifacts"][0]["version"]
|
|
await actions["decide"]({"run_id": run_id, "step_id": "ok", "attempt": 1, "artifact_version": version, "decision": "approve"}, replace(context(runs), interaction="session"))
|
|
result = await done(plugin, run_id)
|
|
assert result["status"] == "completed", result
|
|
step = await actions["get_step"]({"run_id": run_id, "step_id": "b"}, context(runs))
|
|
task = json.loads(runs.starts[-1][1]["messages"][0]["content"])
|
|
assert step["input_artifacts"] == task["approved_artifacts"]
|
|
assert task["approved_artifacts"][0]["version"] == version
|
|
assert task["approved_artifacts"][0]["sha256"] == paused["artifacts"][0]["sha256"]
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
@pytest.mark.parametrize("duplicate", [False, True])
|
|
async def test_parallel_approved_file_consumers_reserve_atomically(plugin, monkeypatch, duplicate):
|
|
import threading
|
|
from dataclasses import replace
|
|
|
|
_, actions, service, runs = plugin
|
|
service.config["max_calls"] = 8 if duplicate else 2
|
|
script = """
|
|
await agent({stepId:'source',agentRef:'lead_agent',task:'file'});
|
|
const f=await captureArtifact({stepId:'file',sourceStepId:'source',path:'/mnt/user-data/outputs/report.pdf'});
|
|
await humanCheckpoint({stepId:'confirm',sourceStepId:'file',artifactVersion:f.artifactVersion});
|
|
return await Promise.all([
|
|
agent({stepId:'consumer',agentRef:'lead_agent',task:'read',artifacts:[f.artifactVersion]}),
|
|
agent({stepId:args.second,agentRef:'lead_agent',task:'read',artifacts:[f.artifactVersion]})]);
|
|
"""
|
|
launched = await launch(plugin, script, args={"second": "consumer" if duplicate else "other"})
|
|
run_id = launched["run_id"]
|
|
paused = await waiting(plugin, run_id)
|
|
original = service.store.artifact
|
|
barrier = threading.Barrier(2, timeout=5)
|
|
reads = []
|
|
lock = threading.Lock()
|
|
|
|
def synchronized(*args):
|
|
value = original(*args)
|
|
with lock:
|
|
reads.append(args)
|
|
decision_read = len(reads) == 1
|
|
if not decision_read:
|
|
barrier.wait()
|
|
return value
|
|
|
|
monkeypatch.setattr(service.store, "artifact", synchronized)
|
|
await actions["decide"]({"run_id": run_id, "step_id": "confirm", "attempt": 1, "artifact_version": paused["artifacts"][0]["version"], "decision": "approve"}, replace(context(runs), interaction="session"))
|
|
result = await done(plugin, run_id)
|
|
assert len(reads) == 3 and not barrier.broken
|
|
assert result["status"] == "failed", result
|
|
assert len(result["steps"]) <= 2
|
|
assert len(runs.starts) <= 2
|
|
assert len({s["step_id"] for s in result["steps"]}) == len(result["steps"])
|