mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 13:16:24 +00:00
cfe2bec371
* feat(workflows): add leased multi-Gateway coordination and fenced dispatch * test(workflows): validate deployed failover and canonicalize config revisions * fix(workflows): share namespace across registration and lease fencing * fix(workflows): recover interrupted cancellation without admission caps * fix(workflows): retain execution lease through child cleanup * docs: trim guidance padding to fit merged instruction budget * fix(workflows): release rejected admission slots and normalize gateway routing --------- Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
860 lines
41 KiB
Python
860 lines
41 KiB
Python
"""Supervised workflow service using only the public AgentRuns capability."""
|
|
|
|
import asyncio
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import logging
|
|
import os
|
|
import re
|
|
import sys
|
|
import tempfile
|
|
import time
|
|
import uuid
|
|
from pathlib import Path
|
|
|
|
from deerflow_extension_api import AgentRunError
|
|
|
|
from .artifacts import Artifacts
|
|
from .cluster import Cluster
|
|
from .constants import NAMESPACE as NAMESPACE
|
|
from .lifecycle import Lifecycle
|
|
from .postgres import LeaseLost
|
|
from .recovery import RECOVERY_VERSION, Recovery
|
|
from .store import TERMINAL, Store
|
|
|
|
FRAME = 48 * 1024
|
|
TURN_LIMIT = 10000
|
|
AGENT_TASK_BYTES = 16000
|
|
AGENT_OUTPUT_BYTES = 12000
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def encoded(value, limit=FRAME - 1):
|
|
raw = json.dumps(value, ensure_ascii=True, allow_nan=False, sort_keys=True)
|
|
if len(raw.encode("utf-8")) > limit:
|
|
raise ValueError("JSON byte limit exceeded")
|
|
return raw
|
|
|
|
|
|
def identity(value):
|
|
return hashlib.sha256(encoded(value).encode()).hexdigest()
|
|
|
|
|
|
def text(value, limit):
|
|
if not isinstance(value, str) or not value.strip() or len(value.encode("utf-8")) > limit:
|
|
raise ValueError("Invalid text or UTF-8 byte limit")
|
|
return value
|
|
|
|
|
|
def pending_input(messages):
|
|
# Any unanswered human-input artifact in this one-step conversation stops the script.
|
|
return any(isinstance(m.get("artifact"), dict) and isinstance(m["artifact"].get("human_input"), dict) and m["artifact"]["human_input"].get("kind") == "human_input_request" for m in messages if m.get("type", m.get("role")) == "tool")
|
|
|
|
|
|
def project(record, detail=True):
|
|
result = {
|
|
key: record[key]
|
|
for key in (
|
|
"id",
|
|
"status",
|
|
"created_at",
|
|
"script_hash",
|
|
"args_hash",
|
|
"limits",
|
|
"error",
|
|
)
|
|
}
|
|
result["run_id"] = record["id"]
|
|
result.update(
|
|
fingerprint=record["fingerprint"],
|
|
definition=record.get("definition"),
|
|
configuration_revision=record.get("configuration_revision"),
|
|
)
|
|
if detail:
|
|
result.update(
|
|
checkpoints=copy.deepcopy(record.get("checkpoints", [])),
|
|
artifacts=copy.deepcopy(record.get("artifacts", [])),
|
|
pause_expires_at=record.get("pause_expires_at"),
|
|
script=record["script"],
|
|
args=record["args"],
|
|
logs=record["logs"],
|
|
result=record.get("result"),
|
|
steps=[
|
|
{
|
|
k: v
|
|
for k, v in step.items()
|
|
if k
|
|
not in (
|
|
"input",
|
|
"output",
|
|
"configuration_evidence",
|
|
"input_artifacts",
|
|
)
|
|
}
|
|
for step in record["steps"]
|
|
],
|
|
)
|
|
encoded(result, 60 * 1024)
|
|
return result
|
|
|
|
|
|
class Workflows(Cluster, Recovery, Lifecycle, Artifacts):
|
|
def __init__(self, config):
|
|
self.config = config
|
|
self.clustered = bool(config.get("storage_url"))
|
|
self.holder_id = uuid.uuid4().hex
|
|
self.heartbeats = {}
|
|
self.store = None
|
|
self.available = False
|
|
self.workers = {}
|
|
self.processes = {}
|
|
self.entries = {}
|
|
self.admissions = set()
|
|
self.launch_lock = asyncio.Lock()
|
|
self.control_lock = asyncio.Lock()
|
|
self.write_lock = asyncio.Lock()
|
|
self.global_slots = asyncio.Semaphore(config["global_concurrency"])
|
|
self.lockfile = None
|
|
|
|
async def start(self, deps):
|
|
def initialize():
|
|
import fcntl
|
|
|
|
path = Path(self.config["storage_path"])
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
lock = None if self.clustered else path.with_suffix(path.suffix + ".lock").open("a+b")
|
|
try:
|
|
if self.clustered:
|
|
from .postgres import PostgresStore
|
|
|
|
store = PostgresStore(self.config["storage_url"], self.holder_id, self.config)
|
|
else:
|
|
fcntl.flock(lock.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
|
|
store = Store(path)
|
|
store.interrupt_unfinished()
|
|
# Include worker code and engine version in the conservative
|
|
# replay compatibility guard; upgrades never silently replay.
|
|
from importlib.metadata import version
|
|
|
|
signature = hashlib.sha256()
|
|
for name in (
|
|
"constants.py",
|
|
"worker.py",
|
|
"service.py",
|
|
"recovery.py",
|
|
"lifecycle.py",
|
|
"artifacts.py",
|
|
"cluster.py",
|
|
"postgres.py",
|
|
):
|
|
signature.update(Path(__file__).with_name(name).read_bytes())
|
|
signature.update(version("quickjs").encode())
|
|
return lock, store, signature.hexdigest()
|
|
except BaseException:
|
|
if lock:
|
|
lock.close()
|
|
raise
|
|
|
|
self.lockfile, self.store, self.runtime_version = await asyncio.to_thread(initialize)
|
|
if self.clustered and deps is not None:
|
|
from sqlalchemy import text as sql_text
|
|
|
|
if deps.session_factory is None:
|
|
raise ValueError("Cluster workflows require the host PostgreSQL session factory")
|
|
async with deps.session_factory() as session:
|
|
if session.get_bind().dialect.name != "postgresql":
|
|
raise ValueError("Cluster workflows require host PostgreSQL persistence")
|
|
host_identity = tuple((await session.execute(sql_text("SELECT current_database(),current_schema(),inet_server_addr()::text,inet_server_port()"))).one())
|
|
if host_identity != await asyncio.to_thread(self.store.database_identity):
|
|
raise ValueError("Workflow and host must share the same PostgreSQL database and schema")
|
|
self.available = True
|
|
|
|
async def stop(self):
|
|
async with self.launch_lock:
|
|
self.available = False
|
|
for entry in self.entries.values():
|
|
entry["suspending"] = entry["record"]["status"] != "cancelling"
|
|
entry["cancel"].set()
|
|
workers = list(self.workers.values())
|
|
# Admissions waiting for launch_lock see unavailable; committed admissions
|
|
# held the lock through worker publication and are included above.
|
|
if self.admissions:
|
|
await asyncio.gather(*list(self.admissions), return_exceptions=True)
|
|
if workers:
|
|
await asyncio.gather(*workers, return_exceptions=True)
|
|
if self.heartbeats:
|
|
heartbeats = list(self.heartbeats.values())
|
|
for heartbeat in heartbeats:
|
|
heartbeat.cancel()
|
|
await asyncio.gather(*heartbeats, return_exceptions=True)
|
|
self.heartbeats.clear()
|
|
if self.lockfile:
|
|
await asyncio.to_thread(self.lockfile.close)
|
|
self.lockfile = None
|
|
|
|
async def wait_idle(self, run_id):
|
|
worker = self.workers.get(run_id)
|
|
if worker:
|
|
await asyncio.shield(worker)
|
|
|
|
def owner(self, context):
|
|
if not self.available or context.settings.get("enabled") is not True:
|
|
raise ValueError("Workflow service unavailable or disabled")
|
|
return context.principal.user_id
|
|
|
|
async def save(self, owner, record):
|
|
# SQLite commits survive cancellation of to_thread. Keep the lock and
|
|
# observe the new revision before propagating cancellation to cleanup.
|
|
task = asyncio.create_task(self.save_drained(owner, record))
|
|
try:
|
|
await asyncio.shield(task)
|
|
except asyncio.CancelledError:
|
|
while not task.done():
|
|
try:
|
|
await asyncio.wait({task})
|
|
except asyncio.CancelledError:
|
|
continue
|
|
if not task.cancelled():
|
|
task.exception()
|
|
raise
|
|
|
|
async def save_drained(self, owner, record):
|
|
async with self.write_lock:
|
|
try:
|
|
record["revision"] = await asyncio.to_thread(self.store.put, owner, copy.deepcopy(record))
|
|
except LeaseLost:
|
|
entry = self.entries.get(record["id"])
|
|
if entry:
|
|
entry["lease_lost"] = True
|
|
entry["cancel"].set()
|
|
raise
|
|
|
|
async def launch(self, payload, context):
|
|
owner = self.owner(context)
|
|
if context.agent_runs is None:
|
|
raise ValueError("Session-bound AgentRuns required; PAT/internal callers unsupported")
|
|
if set(payload) - {"request_id", "script", "args", "definition"}:
|
|
raise ValueError("Unexpected launch fields")
|
|
definition = payload.get("definition")
|
|
if definition is not None:
|
|
if "script" in payload or not isinstance(definition, dict):
|
|
raise ValueError("Choose a script or an exact definition version")
|
|
definition = await self.get_definition(definition, context)
|
|
script = definition.pop("script")
|
|
else:
|
|
script = text(payload.get("script"), 16000)
|
|
encoded(script, 16000)
|
|
request_id = text(payload.get("request_id"), 128)
|
|
args = copy.deepcopy(payload.get("args", {}))
|
|
if not isinstance(args, dict):
|
|
raise ValueError("args must be a JSON object")
|
|
encoded(args, 4000)
|
|
limits = {
|
|
k: self.config[k]
|
|
for k in (
|
|
"concurrency",
|
|
"global_concurrency",
|
|
"max_workflows",
|
|
"max_workflows_per_owner",
|
|
"admission_timeout_seconds",
|
|
"max_calls",
|
|
"timeout_seconds",
|
|
"pause_timeout_seconds",
|
|
"cpu_seconds",
|
|
"memory_bytes",
|
|
)
|
|
}
|
|
script_hash = hashlib.sha256(script.encode()).hexdigest()
|
|
args_hash = identity(args)
|
|
fingerprint = identity(
|
|
{
|
|
"script_hash": script_hash,
|
|
"args_hash": args_hash,
|
|
"limits": limits,
|
|
"agents": sorted(self.config["agents"]),
|
|
"definition": definition,
|
|
}
|
|
)
|
|
record = {
|
|
"id": uuid.uuid4().hex,
|
|
"recovery_version": RECOVERY_VERSION,
|
|
"runtime_version": self.runtime_version,
|
|
"journal": [],
|
|
"definition": definition,
|
|
"configuration_revision": await context.agent_runs.configuration_revision(assistant_ids=sorted(self.config["agents"]), durable=True) if hasattr(context.agent_runs, "configuration_revision") else None,
|
|
"operator_revision": identity({**self.config, "agents": sorted(self.config["agents"])}),
|
|
"checkpoints": [],
|
|
"artifacts": [],
|
|
"request_id": request_id,
|
|
"fingerprint": fingerprint,
|
|
"script_hash": script_hash,
|
|
"args_hash": args_hash,
|
|
"script": script,
|
|
"args": args,
|
|
"limits": limits,
|
|
"created_at": time.time(),
|
|
"status": "queued",
|
|
"error": None,
|
|
"steps": [],
|
|
"logs": [],
|
|
"result": None,
|
|
}
|
|
# SQLite threads survive caller cancellation. Retain a service-owned task
|
|
# through commit AND supervisor publication, shielding only the caller's wait.
|
|
admission = asyncio.create_task(self.admit(owner, record, context))
|
|
self.admissions.add(admission)
|
|
admission.add_done_callback(self.admission_finished)
|
|
return await asyncio.shield(admission)
|
|
|
|
def admission_finished(self, task):
|
|
self.admissions.discard(task)
|
|
if not task.cancelled():
|
|
error = task.exception()
|
|
if error is not None:
|
|
logger.error(
|
|
"Workflow background action failed (%s); inspect retained workflow and child records",
|
|
type(error).__name__,
|
|
)
|
|
|
|
async def admit(self, owner, record, context):
|
|
async with self.launch_lock:
|
|
self.owner(context)
|
|
request_id, fingerprint = record["request_id"], record["fingerprint"]
|
|
current = [e for e in self.entries.values() if e["owner"] == owner]
|
|
# Capacity admission happens before persist, but matching retries bypass it.
|
|
retry = await asyncio.to_thread(self.store.by_request, owner, request_id)
|
|
if retry:
|
|
if retry["fingerprint"] != fingerprint:
|
|
raise ValueError("request_id already used with different script or arguments")
|
|
return {"run_id": retry["id"], "status": retry["status"]}
|
|
if len(self.entries) >= self.config["max_workflows"] or len(current) >= self.config["max_workflows_per_owner"]:
|
|
raise ValueError("Active workflow limit")
|
|
record, created = await asyncio.to_thread(self.store.create, owner, record)
|
|
if created:
|
|
self.publish(self.new_entry(owner, record, context.agent_runs))
|
|
return {"run_id": record["id"], "status": record["status"]}
|
|
|
|
def new_entry(self, owner, record, runs):
|
|
return {
|
|
"owner": owner,
|
|
"record": record,
|
|
"runs": runs,
|
|
"cancel": asyncio.Event(),
|
|
"tasks": set(),
|
|
"slots": asyncio.Semaphore(self.config["concurrency"]),
|
|
"fault": None,
|
|
}
|
|
|
|
def publish(self, entry):
|
|
run_id = entry["record"]["id"]
|
|
self.entries[run_id] = entry
|
|
if self.clustered:
|
|
self.heartbeats[run_id] = asyncio.create_task(self.follow_lease(entry))
|
|
task = asyncio.create_task(self.execute(entry))
|
|
self.workers[run_id] = task
|
|
task.add_done_callback(lambda t: self.worker_finished(run_id, t))
|
|
|
|
def worker_finished(self, run_id, task):
|
|
heartbeat = self.heartbeats.pop(run_id, None)
|
|
if heartbeat:
|
|
heartbeat.cancel()
|
|
self.workers.pop(run_id, None)
|
|
self.entries.pop(run_id, None)
|
|
if not task.cancelled() and task.exception() is not None:
|
|
logger.error(
|
|
"Workflow %s supervisor failed (%s); inspect retained child records",
|
|
run_id,
|
|
type(task.exception()).__name__,
|
|
)
|
|
|
|
async def get(self, payload, context):
|
|
owner = self.owner(context)
|
|
return project(await asyncio.to_thread(self.store.get, owner, text(payload.get("run_id"), 64)))
|
|
|
|
async def list(self, payload, context):
|
|
owner = self.owner(context)
|
|
records = await asyncio.to_thread(self.store.list, owner)
|
|
return {"runs": [project(r, False) for r in records], "limit": 20}
|
|
|
|
async def cancel(self, payload, context):
|
|
task = asyncio.create_task(self.apply_cancel(payload, context))
|
|
self.admissions.add(task)
|
|
task.add_done_callback(self.admission_finished)
|
|
return await asyncio.shield(task)
|
|
|
|
async def apply_cancel(self, payload, context):
|
|
# Serialize against recovery publication. A recovered record cannot be
|
|
# published after a concurrent cancellation has committed.
|
|
async with self.launch_lock:
|
|
owner = self.owner(context)
|
|
run_id = text(payload.get("run_id"), 64)
|
|
record = await asyncio.to_thread(self.store.get, owner, run_id)
|
|
entry = self.entries.get(run_id)
|
|
if self.clustered and record["status"] not in TERMINAL:
|
|
await self.remote_command(owner, record, context, {"kind": "cancel"})
|
|
return {"run_id": run_id, "status": "cancelling"}
|
|
if self.clustered and record["status"] == "recoverable":
|
|
return await self.cancel_recovered(owner, run_id, context)
|
|
if entry:
|
|
entry["cancel"].set()
|
|
entry["suspending"] = False
|
|
if entry["record"]["status"] not in TERMINAL:
|
|
entry["record"]["status"] = "cancelling"
|
|
await self.save(owner, entry["record"])
|
|
return {"run_id": run_id, "status": "cancelling"}
|
|
if record["status"] == "recoverable":
|
|
if context.agent_runs is None:
|
|
raise ValueError("Session-bound AgentRuns required")
|
|
record["status"] = "cancelling"
|
|
await self.save(owner, record)
|
|
for row in record["steps"]:
|
|
if row.get("run_id") and row["status"] != "completed":
|
|
await self.cancel_child(context.agent_runs, row)
|
|
uncertain = any(s["status"] not in ("completed", "queued", "cancelled") for s in record["steps"])
|
|
record.update(
|
|
status="needs_reconciliation" if uncertain else "cancelled",
|
|
error="Inspect uncertain child effects" if uncertain else None,
|
|
)
|
|
await self.save(owner, record)
|
|
return {"run_id": run_id, "status": record["status"]}
|
|
|
|
async def delete(self, payload, context):
|
|
owner = self.owner(context)
|
|
run_id = text(payload.get("run_id"), 64)
|
|
if run_id in self.entries:
|
|
raise ValueError("Workflow is still cleaning up")
|
|
await asyncio.to_thread(self.store.delete, owner, run_id)
|
|
return {"deleted": True}
|
|
|
|
async def step(self, entry, message, *, replay=False):
|
|
owner, record, runs = entry["owner"], entry["record"], entry["runs"]
|
|
payload = message.get("payload")
|
|
response = {"id": message["id"]}
|
|
row = None
|
|
try:
|
|
if not isinstance(payload, dict) or not {"stepId", "agentRef", "task"} <= set(payload) or set(payload) - {"stepId", "agentRef", "task", "artifacts"}:
|
|
raise ValueError("agent requires stepId, agentRef, task")
|
|
step_id = text(payload["stepId"], 80)
|
|
if not re.fullmatch(r"[A-Za-z0-9_-]{1,80}", step_id):
|
|
raise ValueError("stepId must contain ASCII letters, digits, underscores or hyphens")
|
|
if payload["agentRef"] not in self.config["agents"]:
|
|
raise ValueError("Agent reference not permitted by operator")
|
|
versions = payload.get("artifacts", [])
|
|
if not isinstance(versions, list) or len(versions) > 16 or any(not isinstance(v, str) for v in versions):
|
|
raise ValueError("artifacts must contain at most 16 version IDs")
|
|
references = []
|
|
for version in versions:
|
|
if not self.approved_artifact(record, version):
|
|
raise ValueError("File version has not been approved")
|
|
info, _ = await asyncio.to_thread(self.store.artifact, owner, record["id"], version)
|
|
references.append(
|
|
{
|
|
"run_id": record["id"],
|
|
**{k: info[k] for k in ("version", "sha256", "name", "size")},
|
|
}
|
|
)
|
|
task = payload["task"]
|
|
if references:
|
|
task = {
|
|
"task": task,
|
|
"approved_artifacts": references,
|
|
"artifact_instruction": "Read these exact versions using read_workflow_artifact (run_id, version, offset). Follow next_offset until null. Do not use source file paths.",
|
|
}
|
|
# Stable identity/input are recorded BEFORE thread creation or run admission.
|
|
thread_id = "workflow-" + identity([record["id"], step_id])[:32]
|
|
message_id = "workflow-" + identity([record["id"], step_id, 1])[:32]
|
|
input = {
|
|
"messages": [
|
|
{
|
|
"role": "user",
|
|
"id": message_id,
|
|
"content": encoded(task, AGENT_TASK_BYTES),
|
|
}
|
|
]
|
|
}
|
|
existing = next((s for s in record["steps"] if s["step_id"] == step_id), None) if replay else None
|
|
if existing:
|
|
row = existing
|
|
if row["input"] != input or row["agent_ref"] != payload["agentRef"]:
|
|
raise ValueError("Recovery step input changed")
|
|
if row["status"] == "completed":
|
|
response["value"] = self.step_result(row)
|
|
return response
|
|
if row["status"] not in ("queued", "running"):
|
|
raise ValueError("Child outcome requires reconciliation")
|
|
# Revalidate after file IO; validation and reservation must not yield.
|
|
if not existing:
|
|
if any(s["step_id"] == step_id for s in record["steps"] + record.get("checkpoints", []) + record.get("artifacts", [])):
|
|
raise ValueError("Step ID already used")
|
|
if len(record["steps"]) >= self.config["max_calls"]:
|
|
entry["fault"] = "Total Agent call limit"
|
|
raise ValueError(entry["fault"])
|
|
row = {
|
|
"step_id": step_id,
|
|
"attempt": 1,
|
|
"agent_ref": payload["agentRef"],
|
|
"thread_id": thread_id,
|
|
"run_id": None,
|
|
"status": "queued",
|
|
"input": input,
|
|
"input_artifacts": references,
|
|
"idempotency_key": f"{record['id']}:{message_id}:1",
|
|
}
|
|
record["steps"].append(row)
|
|
await self.save(owner, record)
|
|
async with entry["slots"], self.global_slots, self.cluster_slot(entry, row):
|
|
if entry["cancel"].is_set() or entry["fault"]:
|
|
row["status"] = "cancelled"
|
|
raise ValueError("Workflow stopping")
|
|
if record.get("configuration_revision") is not None:
|
|
try:
|
|
await self.check_configuration(entry)
|
|
except Exception:
|
|
entry["fault"] = "Agent configuration changed or unavailable"
|
|
row["status"] = "failed"
|
|
if row.get("run_id") is None:
|
|
row["admission_rejected"] = True
|
|
raise
|
|
if row["status"] == "queued":
|
|
row["status"] = "creating_thread"
|
|
await self.save(owner, record)
|
|
# Never cancel these awaits just because a workflow cancel was requested.
|
|
# Keep tracking the response so an admitted child cannot disappear.
|
|
async with asyncio.timeout(self.config["admission_timeout_seconds"]):
|
|
await runs.create_thread(
|
|
assistant_id=payload["agentRef"],
|
|
thread_id=thread_id,
|
|
metadata={"title": f"Workflow {record['id'][:8]} / {step_id}"},
|
|
)
|
|
if entry["cancel"].is_set():
|
|
row["status"] = "cancelled"
|
|
raise ValueError("Workflow stopping")
|
|
row["status"] = "admitting"
|
|
await self.save(owner, record)
|
|
run = await runs.start(
|
|
thread_id=thread_id,
|
|
input=input,
|
|
idempotency_key=row["idempotency_key"],
|
|
**self.dispatch_fence(record),
|
|
)
|
|
row.update(
|
|
run_id=run.run_id,
|
|
status="running",
|
|
effective_assistant_id=run.assistant_id,
|
|
)
|
|
await self.save(owner, record)
|
|
else:
|
|
run = await runs.get(thread_id=thread_id, run_id=row["run_id"])
|
|
while True:
|
|
if entry["cancel"].is_set() or entry["fault"]:
|
|
if not entry.get("lease_lost"):
|
|
await self.cancel_child(runs, row, **({"record": record} if self.clustered else {}))
|
|
raise ValueError("Workflow stopping")
|
|
async with asyncio.timeout(5):
|
|
run = await runs.get(thread_id=thread_id, run_id=run.run_id)
|
|
if run.status not in ("pending", "running"):
|
|
break
|
|
await asyncio.sleep(0.02)
|
|
row.update(child_status=run.status, stop_reason=run.stop_reason)
|
|
async with asyncio.timeout(5):
|
|
state = await runs.get_state(thread_id=thread_id)
|
|
messages = state.get("values", {}).get("messages", [])
|
|
if state.get("next") or state.get("interrupts") or pending_input(messages):
|
|
row["status"] = "waiting_input"
|
|
entry["fault"] = "waiting_input"
|
|
raise ValueError("Child requires input; workflow resume cannot answer child interrupts")
|
|
if run.status != "success":
|
|
row["status"] = "failed"
|
|
raise ValueError("Agent run did not succeed")
|
|
# Dynamic context keeps X on a hidden system reminder and moves
|
|
# the visible human input to X__user. Never anchor on the reminder
|
|
# or hidden memory; genuinely later human messages still conflict.
|
|
marker = next(
|
|
(
|
|
i
|
|
for i, m in enumerate(messages)
|
|
if m.get("type", m.get("role")) in ("human", "user")
|
|
and not m.get("additional_kwargs", {}).get("hide_from_ui")
|
|
and (m.get("id") in (message_id, f"{message_id}__user") or m.get("additional_kwargs", {}).get("run_id") == run.run_id)
|
|
),
|
|
None,
|
|
)
|
|
if marker is None or any(m.get("type", m.get("role")) in ("human", "user") for m in messages[marker + 1 :]):
|
|
row["status"] = "result_conflict"
|
|
raise ValueError("Thread advanced before result collection")
|
|
answers = [m for m in messages[marker + 1 :] if m.get("type", m.get("role")) in ("ai", "assistant") and not m.get("tool_calls")]
|
|
if not answers:
|
|
raise ValueError("Agent completed without an answer")
|
|
content = answers[-1].get("content")
|
|
if isinstance(content, str):
|
|
try:
|
|
output = json.loads(content)
|
|
except ValueError:
|
|
output = content
|
|
else:
|
|
output = content
|
|
encoded(output, AGENT_OUTPUT_BYTES)
|
|
# Evidence comes from the run's completed host assembly, never a
|
|
# fresh read of mutable configuration after the Agent finishes.
|
|
async with asyncio.timeout(5):
|
|
proof = await runs.configuration_evidence(thread_id=thread_id, run_id=run.run_id)
|
|
if proof.get("run_id") != run.run_id or proof.get("thread_id") != thread_id or proof.get("source") != "host_assembly":
|
|
raise ValueError("Host configuration evidence does not match child run")
|
|
row["configuration_evidence"] = proof
|
|
row["configuration_version"] = proof["version"]
|
|
# Capture before any later turn can alter the thread checkpoint.
|
|
row.update(
|
|
status="completed",
|
|
output=copy.deepcopy(output),
|
|
artifact_version=identity([record["id"], step_id, row["attempt"], output]),
|
|
)
|
|
response["value"] = self.step_result(row)
|
|
except TimeoutError:
|
|
if row and row["run_id"] is not None:
|
|
# Admission is known. Final cleanup interrupts/observes the tracked
|
|
# run and escalates to reconciliation only if that cleanup fails.
|
|
row["status"] = "failed"
|
|
entry["fault"] = "Child observation timed out; interrupting tracked run"
|
|
else:
|
|
if row:
|
|
row["status"] = "needs_reconciliation"
|
|
entry["fault"] = "Admission timed out; inspect the stable child thread, do not replay"
|
|
response["error"] = entry["fault"]
|
|
except AgentRunError as exc:
|
|
if row:
|
|
if row.get("run_id") is None and exc.status_code in (400, 403, 404, 409, 422):
|
|
row["admission_rejected"] = True
|
|
row["status"] = "failed" if exc.status_code in (400, 403, 404, 409, 422) else "needs_reconciliation"
|
|
response["error"] = f"AgentRuns rejected operation ({exc.status_code})"
|
|
entry["fault"] = response["error"]
|
|
except Exception as exc:
|
|
if row and row["status"] in ("creating_thread", "admitting"):
|
|
row["status"] = "needs_reconciliation"
|
|
entry["fault"] = "Uncertain admission; inspect the child thread, do not replay"
|
|
elif row and row["status"] == "running":
|
|
row["status"] = "failed"
|
|
response["error"] = str(exc)[:240] if isinstance(exc, ValueError) else "Agent step failed"
|
|
finally:
|
|
if replay and "error" in response:
|
|
entry["fault"] = "Recovered child outcome requires reconciliation"
|
|
if row:
|
|
row["status"] = "needs_reconciliation"
|
|
if row and "error" in response:
|
|
row["error"] = response["error"]
|
|
await self.save(owner, record)
|
|
return response
|
|
|
|
@staticmethod
|
|
def step_result(row):
|
|
return {
|
|
"artifactVersion": row["artifact_version"],
|
|
"output": copy.deepcopy(row["output"]),
|
|
"threadId": row["thread_id"],
|
|
"runId": row["run_id"],
|
|
"status": "success",
|
|
}
|
|
|
|
async def cancel_child(self, runs, row, record=None):
|
|
try:
|
|
async with asyncio.timeout(5):
|
|
run = await runs.get(thread_id=row["thread_id"], run_id=row["run_id"])
|
|
if run.status in ("pending", "running"):
|
|
await runs.cancel(thread_id=row["thread_id"], run_id=row["run_id"], **(self.dispatch_fence(record) if record is not None else {}))
|
|
run = await runs.wait(thread_id=row["thread_id"], run_id=row["run_id"], timeout=4)
|
|
row["child_status"] = run.status
|
|
if row["status"] == "running":
|
|
row["status"] = "cancelled"
|
|
except Exception:
|
|
row["status"] = "needs_reconciliation"
|
|
|
|
async def execute(self, entry):
|
|
owner, record, runs = entry["owner"], entry["record"], entry["runs"]
|
|
process = None
|
|
workspace = None
|
|
final_state = {}
|
|
try:
|
|
if record.get("budget_deadline") is not None:
|
|
record["remaining_seconds"] = max(0, record["budget_deadline"] - time.time())
|
|
record["budget_deadline"] = time.time() + record.get("remaining_seconds", self.config["timeout_seconds"])
|
|
record["status"] = "running"
|
|
await self.save(owner, record)
|
|
workspace = await asyncio.to_thread(tempfile.TemporaryDirectory, prefix="deerflow-workflow-")
|
|
directory = workspace.name
|
|
# -I excludes cwd/PYTHONPATH and the environment contains no host credentials.
|
|
env = {"PATH": os.defpath, "LANG": "C.UTF-8"}
|
|
process = await asyncio.create_subprocess_exec(
|
|
sys.executable,
|
|
"-I",
|
|
str(Path(__file__).with_name("worker.py")),
|
|
stdin=asyncio.subprocess.PIPE,
|
|
stdout=asyncio.subprocess.PIPE,
|
|
stderr=asyncio.subprocess.DEVNULL,
|
|
cwd=directory,
|
|
env=env,
|
|
limit=FRAME,
|
|
)
|
|
self.processes[record["id"]] = process
|
|
await self.send(
|
|
process,
|
|
{
|
|
"script": record["script"],
|
|
"args": record["args"],
|
|
"cpu_seconds": self.config["cpu_seconds"],
|
|
"memory_bytes": self.config["memory_bytes"],
|
|
},
|
|
)
|
|
async with asyncio.timeout(record.get("remaining_seconds", self.config["timeout_seconds"])) as deadline:
|
|
entry["deadline"] = deadline
|
|
turns = 0
|
|
while True:
|
|
if entry["cancel"].is_set() or entry["fault"]:
|
|
raise ValueError(entry["fault"] or "Cancelled")
|
|
turns += 1
|
|
if turns > TURN_LIMIT:
|
|
raise ValueError("Worker turn limit")
|
|
raw = await process.stdout.readline()
|
|
if not raw or len(raw) > FRAME:
|
|
raise ValueError("Worker exited or exceeded frame limit")
|
|
if entry["cancel"].is_set() or entry["fault"]:
|
|
raise ValueError(entry["fault"] or "Cancelled")
|
|
frame = json.loads(raw)
|
|
messages = frame["messages"]
|
|
if not isinstance(messages, list) or len(messages) > 64:
|
|
raise ValueError("Worker message limit")
|
|
replayed = await self.begin_frame(entry, frame)
|
|
if replayed is not None:
|
|
await self.send(process, {"responses": replayed})
|
|
continue
|
|
if any(m.get("type") in ("pause", "humanCheckpoint") for m in messages):
|
|
if len(messages) != 1 or entry["tasks"] or frame["outcome"] is not None:
|
|
raise ValueError("Checkpoint must be awaited alone, after all Agent calls")
|
|
response = await self.checkpoint(entry, messages[0], deadline)
|
|
await self.deliver_frame(entry, process, [response])
|
|
continue
|
|
if any(m.get("type") in ("captureArtifact", "readArtifact") for m in messages):
|
|
if len(messages) != 1 or entry["tasks"] or frame["outcome"] is not None:
|
|
raise ValueError("Artifact operations must be awaited alone")
|
|
operation = self.capture if messages[0]["type"] == "captureArtifact" else self.consume
|
|
response = await operation(entry, messages[0])
|
|
await self.deliver_frame(entry, process, [response])
|
|
continue
|
|
for message_index, message in enumerate(messages):
|
|
if message.get("type") == "log":
|
|
log_id = f"{entry['frame_index']}:{message_index}"
|
|
if log_id in record.get("log_ids", []):
|
|
continue
|
|
if len(record["logs"]) >= 20:
|
|
raise ValueError("Log limit")
|
|
log = text(message.get("text"), 160)
|
|
encoded(log, 160)
|
|
record["logs"].append(log)
|
|
record.setdefault("log_ids", []).append(log_id)
|
|
await self.save(owner, record)
|
|
elif message.get("type") == "agent":
|
|
if type(message.get("id")) is not int or len(entry["tasks"]) >= self.config["max_calls"]:
|
|
raise ValueError("Pending Agent call limit")
|
|
entry["tasks"].add(
|
|
asyncio.create_task(
|
|
self.step(
|
|
entry,
|
|
message,
|
|
replay=entry.get("recover_frame", False),
|
|
)
|
|
)
|
|
)
|
|
else:
|
|
raise ValueError("Unsupported worker message")
|
|
if frame["outcome"] is not None:
|
|
outcome = frame["outcome"]
|
|
if entry["tasks"]:
|
|
raise ValueError("Script returned before awaiting Agent calls")
|
|
if outcome.get("type") != "result":
|
|
message = outcome.get("message")
|
|
raise ValueError(message[:240] if isinstance(message, str) else "Script failed or exceeded limits")
|
|
encoded(outcome["value"], 12000)
|
|
final_state.update(status="completed", result=outcome["value"])
|
|
break
|
|
# Poll cancel/fault/worker exit while a response is pending. An idle worker can only progress
|
|
# on a response, so keep waiting here instead of sending empty frames (turns).
|
|
if entry["tasks"]:
|
|
while True:
|
|
finished, _ = await asyncio.wait(
|
|
entry["tasks"],
|
|
timeout=0.02,
|
|
return_when=asyncio.FIRST_COMPLETED,
|
|
)
|
|
if process.returncode is not None:
|
|
raise ValueError("Worker exited while waiting for Agent calls")
|
|
if finished or not frame.get("idle", False) or entry["cancel"].is_set() or entry["fault"]:
|
|
break
|
|
else:
|
|
await asyncio.sleep(0.002)
|
|
finished = [t for t in entry["tasks"] if t.done()]
|
|
responses = [t.result() for t in finished]
|
|
entry["tasks"].difference_update(finished)
|
|
await self.deliver_frame(entry, process, responses)
|
|
except TimeoutError:
|
|
# Keep this outside asyncio.timeout: its own cancellation must first
|
|
# leave the context manager so it can be translated to TimeoutError.
|
|
final_state.update(
|
|
status="failed",
|
|
error="Checkpoint wait deadline exceeded" if record["status"] in ("paused", "waiting_approval") else "Execution deadline exceeded",
|
|
)
|
|
except asyncio.CancelledError:
|
|
entry["cancel"].set()
|
|
final_state.update(status="interrupted", error="Host interrupted; child cleanup required")
|
|
except Exception as exc:
|
|
final_state.update(
|
|
status="cancelled" if entry["cancel"].is_set() else "waiting_input" if entry["fault"] == "waiting_input" else "failed",
|
|
error=str(exc)[:240],
|
|
)
|
|
finally:
|
|
entry["cancel"].set()
|
|
if process:
|
|
if process.returncode is None:
|
|
try:
|
|
process.kill()
|
|
except ProcessLookupError:
|
|
pass
|
|
await process.wait()
|
|
self.processes.pop(record["id"], None)
|
|
if workspace:
|
|
await asyncio.to_thread(workspace.cleanup)
|
|
# Admission tasks run to a bounded completion; no lost response on workflow cancel.
|
|
if entry["tasks"]:
|
|
await asyncio.gather(*entry["tasks"], return_exceptions=True)
|
|
if entry.get("lease_lost"):
|
|
return
|
|
cleanup = [self.cancel_child(runs, row, **({"record": record} if self.clustered else {})) for row in record["steps"] if row.get("run_id") and row["status"] not in ("completed", "waiting_input", "cancelled")]
|
|
if cleanup:
|
|
await asyncio.gather(*cleanup)
|
|
if self.clustered:
|
|
for row in record["steps"]:
|
|
if self.slot_releasable(row):
|
|
await asyncio.to_thread(self.store.release_slot, owner, record, row["step_id"])
|
|
# Child saves must not publish a terminal workflow and retire its
|
|
# lease before all fenced cleanup and reservation releases finish.
|
|
record.update(final_state)
|
|
if any(s["status"] == "needs_reconciliation" for s in record["steps"]):
|
|
record.update(
|
|
status="needs_reconciliation",
|
|
error="Some child outcomes are uncertain; inspect their threads/runs before new work",
|
|
)
|
|
if entry.get("suspending") and record["status"] not in (
|
|
"completed",
|
|
"needs_reconciliation",
|
|
):
|
|
record.update(status="recoverable", error="Gateway stopped; sign in to recover")
|
|
if any(s["status"] not in ("queued", "completed", "running") for s in record["steps"]):
|
|
record.update(
|
|
status="needs_reconciliation",
|
|
error="Gateway stopped with unfinished child effects; inspect child runs",
|
|
)
|
|
await self.save(owner, record)
|
|
self.entries.pop(record["id"], None)
|
|
|
|
async def send(self, process, message):
|
|
process.stdin.write((encoded(message) + "\n").encode())
|
|
await process.stdin.drain()
|