1240 lines
50 KiB
Python
1240 lines
50 KiB
Python
"""Workflow run APIs: start / inspect / cancel / resume / retry / events / SSE stream."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import hashlib
|
||
import json
|
||
import logging
|
||
from datetime import UTC, datetime
|
||
from typing import Any
|
||
|
||
from fastapi import APIRouter, HTTPException, Query, Request
|
||
from fastapi.responses import StreamingResponse
|
||
from pydantic import BaseModel, Field
|
||
|
||
from app.gateway.deps import get_current_user, get_optional_user_from_request
|
||
from app.gateway.workflow_audit import audit_workflow
|
||
from app.gateway.workflow_resource_checks import build_resource_catalog, make_resource_checker
|
||
from deerflow.config.app_config import get_app_config
|
||
from deerflow.persistence.workflow_events.base import WorkflowEventStore
|
||
from deerflow.persistence.workflow_planning import (
|
||
WorkflowPlanningConflictError,
|
||
WorkflowPlanningProposalNotFoundError,
|
||
WorkflowPlanningStore,
|
||
)
|
||
from deerflow.persistence.workflow_runs.base import WorkflowRunStore
|
||
from deerflow.workflows.runtime.sink import envelope_from_row
|
||
from deerflow.workflows.schema_validation import schema_issues
|
||
from deerflow.workflows.schemas import WorkflowGraph
|
||
from deerflow.workflows.validator import validate_workflow_graph
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
router = APIRouter(prefix="/api/workflows", tags=["workflow-runs"])
|
||
|
||
_TERMINAL_STATUSES = ("completed", "failed", "cancelled")
|
||
_TERMINAL_EVENTS = ("run.completed", "run.failed", "run.cancelled")
|
||
_REPLAY_PAGE = 200
|
||
_DB_POLL_SECONDS = 0.8
|
||
_LIVE_WAIT_SECONDS = 0.25
|
||
_HEARTBEAT_SECONDS = 15.0
|
||
_STREAM_DEADLINE_SECONDS = 60 * 60
|
||
# Send one harmless, sufficiently sized comment as soon as the connection is
|
||
# accepted. This prevents a proxy from withholding the first small durable
|
||
# event until the run has already produced a batch of output. It is an SSE
|
||
# comment, so it does not affect the persisted event sequence or UI reducer.
|
||
_SSE_FLUSH_PADDING = " " * 2_048
|
||
|
||
|
||
class StartRunBody(BaseModel):
|
||
version_id: str | None = Field(default=None, alias="versionId")
|
||
inputs: dict[str, Any] = Field(default_factory=dict)
|
||
# 独立 Coze 工作流页早期使用 input/target；保留兼容入口，内部仍只使用 inputs。
|
||
input: dict[str, Any] | None = None
|
||
target: str | None = None
|
||
# 对话面板的显式定向任务：不是打断既有 run，而是从当前图中选取一个已
|
||
# 绑定的 Agent，投影成可审计的 start → agent → output 新运行。
|
||
target_agent_id: str | None = Field(default=None, alias="targetAgentId", max_length=191)
|
||
model_name: str | None = Field(default=None, alias="modelName", max_length=191)
|
||
# 规划会话中的已选候选图。二者必须同时给出；正式执行只读取服务端持久化
|
||
# 的 proposal 快照，永不接受浏览器直接塞入的临时 graph。
|
||
planning_session_id: str | None = Field(default=None, alias="planningSessionId", max_length=64)
|
||
proposal_id: str | None = Field(default=None, alias="proposalId", max_length=64)
|
||
idempotency_key: str | None = Field(default=None, alias="idempotencyKey", max_length=128)
|
||
|
||
model_config = {"populate_by_name": True}
|
||
|
||
def resolved_inputs(self) -> dict[str, Any]:
|
||
return self.inputs if self.inputs or self.input is None else self.input
|
||
|
||
|
||
class ResumeRunBody(BaseModel):
|
||
resume_token: str = Field(alias="resumeToken")
|
||
node_id: str | None = Field(default=None, alias="nodeId")
|
||
action: str = "submit"
|
||
values: dict[str, Any] = Field(default_factory=dict)
|
||
# 兼容独立页的 ResumeRunRequest：payload/message。
|
||
payload: dict[str, Any] | None = None
|
||
message: str | None = None
|
||
|
||
model_config = {"populate_by_name": True}
|
||
|
||
def resolved_values(self) -> dict[str, Any]:
|
||
return self.values if self.values or self.payload is None else self.payload
|
||
|
||
|
||
class FeedbackRunBody(BaseModel):
|
||
"""A targeted revision request for one completed Agent/Skill node."""
|
||
|
||
node_id: str = Field(alias="nodeId", min_length=1, max_length=128)
|
||
message: str = Field(min_length=1, max_length=8_000)
|
||
idempotency_key: str | None = Field(default=None, alias="idempotencyKey", max_length=128)
|
||
|
||
model_config = {"populate_by_name": True}
|
||
|
||
|
||
# ── shared plumbing ─────────────────────────────────────────────────────
|
||
|
||
|
||
def _cfg():
|
||
return get_app_config().workflows
|
||
|
||
|
||
def _require_enabled() -> None:
|
||
if not _cfg().enabled:
|
||
raise HTTPException(status_code=503, detail="Workflow Studio is disabled")
|
||
|
||
|
||
def _selected_model_name(value: str | None) -> str | None:
|
||
model_name = str(value or "").strip()
|
||
if not model_name:
|
||
return None
|
||
if get_app_config().get_model_config(model_name) is None:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={"code": "WORKFLOW_MODEL_INVALID", "message": "所选模型不存在或已不可用"},
|
||
)
|
||
return model_name
|
||
|
||
|
||
def _runs(request: Request) -> WorkflowRunStore:
|
||
store = getattr(request.app.state, "workflow_run_store", None)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Workflow run store not available")
|
||
return store
|
||
|
||
|
||
def _events(request: Request) -> WorkflowEventStore:
|
||
store = getattr(request.app.state, "workflow_event_store", None)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Workflow event store not available")
|
||
return store
|
||
|
||
|
||
def _definitions(request: Request):
|
||
store = getattr(request.app.state, "workflow_store", None)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Workflow store not available")
|
||
return store
|
||
|
||
|
||
def _planning(request: Request) -> WorkflowPlanningStore:
|
||
store = getattr(request.app.state, "workflow_planning_store", None)
|
||
if store is None:
|
||
raise HTTPException(status_code=503, detail="Workflow planning store not available")
|
||
return store
|
||
|
||
|
||
def _nudge(request: Request) -> None:
|
||
dispatcher = getattr(request.app.state, "workflow_dispatcher", None)
|
||
if dispatcher is not None:
|
||
dispatcher.nudge()
|
||
|
||
|
||
async def _is_admin(request: Request) -> bool:
|
||
user = await get_optional_user_from_request(request)
|
||
return user is not None and getattr(user, "system_role", None) == "admin"
|
||
|
||
|
||
async def _authorized_run(request: Request, run_id: str) -> dict[str, Any]:
|
||
run = await _runs(request).get_run(run_id)
|
||
if run is None:
|
||
raise HTTPException(status_code=404, detail={"code": "WORKFLOW_RUN_NOT_FOUND", "message": "运行不存在"})
|
||
user_id = await get_current_user(request)
|
||
if run.get("owner_id") != user_id and not await _is_admin(request):
|
||
raise HTTPException(status_code=403, detail={"code": "WORKFLOW_FORBIDDEN", "message": "无权访问该运行"})
|
||
return run
|
||
|
||
|
||
def _public_run(run: dict[str, Any]) -> dict[str, Any]:
|
||
"""Run view for HTTP clients: no lease internals, no resume token."""
|
||
context = run.get("context") or {}
|
||
return {
|
||
"id": run["id"],
|
||
"runId": run["id"],
|
||
"workflowId": run["workflow_id"],
|
||
"versionId": run["workflow_version_id"],
|
||
"status": run["status"],
|
||
"ownerId": run["owner_id"],
|
||
"input": run.get("input") or {},
|
||
"output": run.get("output"),
|
||
"error": run.get("error"),
|
||
"pendingInput": run.get("pending_input"),
|
||
"attempt": run.get("attempt"),
|
||
"retryOfRunId": run.get("retry_of_run_id"),
|
||
"startedAt": run.get("started_at"),
|
||
"finishedAt": run.get("finished_at"),
|
||
"createdAt": run.get("created_at"),
|
||
"updatedAt": run.get("updated_at"),
|
||
# The run row's counter holds the last seq handed out by the event log
|
||
# (the event store bumps it in the insert transaction); 0 = no events.
|
||
"lastSeq": int(run.get("next_event_seq") or 0),
|
||
"executionMode": context.get("executionMode") or "workflow",
|
||
"targetAgentId": context.get("targetAgentId"),
|
||
"targetAgentName": context.get("targetAgentName"),
|
||
"planningSessionId": context.get("planningSessionId"),
|
||
"proposalId": context.get("proposalId"),
|
||
"revisionOfRunId": context.get("revisionOfRunId"),
|
||
"feedback": context.get("feedback"),
|
||
}
|
||
|
||
|
||
async def _run_validation_issues(
|
||
request: Request,
|
||
*,
|
||
workflow_id: str,
|
||
graph: WorkflowGraph,
|
||
user_id: str,
|
||
) -> list[Any]:
|
||
"""Validate the exact graph snapshot about to enter the run queue.
|
||
|
||
Studio validation is advisory while an author is editing. Starting a run
|
||
is the enforcement boundary: do not queue an unbound SQL/HTTP/subworkflow
|
||
node only to fail it later after a dispatcher lease has been acquired.
|
||
"""
|
||
config = _cfg()
|
||
catalog = await build_resource_catalog(request, user_id=user_id)
|
||
checker = make_resource_checker(
|
||
catalog,
|
||
config=config,
|
||
current_workflow_id=workflow_id,
|
||
)
|
||
return validate_workflow_graph(
|
||
graph,
|
||
resource_checker=checker,
|
||
system_max_steps=config.max_steps,
|
||
system_max_loop_iterations=config.max_loop_iterations,
|
||
system_max_parallelism=config.max_parallelism,
|
||
)
|
||
|
||
|
||
def _agent_task_graph(
|
||
graph: WorkflowGraph,
|
||
target_agent_id: str,
|
||
model_name: str | None = None,
|
||
) -> WorkflowGraph:
|
||
"""Build an explicit, durable one-agent task graph from a canvas graph.
|
||
|
||
A chat composer cannot safely "message" a running node: that loses DAG
|
||
ordering, cancellation, event persistence and resource checks. Instead a
|
||
target selection starts a separate run whose graph is a small projection of
|
||
one *already bound* agent node. The original node id is preserved, so its
|
||
streamed output still highlights the corresponding canvas card.
|
||
"""
|
||
matches = [
|
||
node
|
||
for node in graph.nodes
|
||
if node.type == "agent"
|
||
and str(node.config.get("agentId") or node.config.get("agent_id") or "")
|
||
== target_agent_id
|
||
]
|
||
if not matches:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={
|
||
"code": "WORKFLOW_AGENT_TARGET_NOT_BOUND",
|
||
"message": "指定的智能体尚未绑定到当前画布，请先从节点资源拖入该智能体",
|
||
"details": {"targetAgentId": target_agent_id},
|
||
},
|
||
)
|
||
if len(matches) > 1:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={
|
||
"code": "WORKFLOW_AGENT_TARGET_AMBIGUOUS",
|
||
"message": "该智能体在画布中绑定了多个节点，请在画布中保留一个定向任务节点后重试",
|
||
"details": {"targetAgentId": target_agent_id, "nodeIds": [node.id for node in matches]},
|
||
},
|
||
)
|
||
|
||
agent = matches[0]
|
||
config = dict(agent.config)
|
||
raw_bindings = config.get("inputBindings") or config.get("input_bindings")
|
||
bindings = dict(raw_bindings) if isinstance(raw_bindings, dict) else {}
|
||
# Regardless of the authoring form, the chat task itself must be visible to
|
||
# the selected agent. Keep every existing binding intact for its role.
|
||
bindings["workflowInput"] = "{{ inputs }}"
|
||
config["inputBindings"] = bindings
|
||
config["agentName"] = str(config.get("agentName") or agent.name or target_agent_id)
|
||
if model_name:
|
||
config["modelName"] = model_name
|
||
|
||
return WorkflowGraph.model_validate(
|
||
{
|
||
"schemaVersion": "1.0",
|
||
"id": graph.id,
|
||
"name": f"{graph.name or graph.id} · {agent.name or target_agent_id}",
|
||
"inputSchema": {
|
||
"type": "object",
|
||
"required": ["query"],
|
||
"properties": {"query": {"type": "string", "minLength": 1}},
|
||
},
|
||
"outputSchema": {"type": "object", "properties": {"text": {"type": "string"}}},
|
||
"nodes": [
|
||
{"id": "__agent_task_start__", "type": "start", "name": "定向任务开始"},
|
||
{"id": agent.id, "type": "agent", "name": agent.name, "config": config},
|
||
{
|
||
"id": "__agent_task_output__",
|
||
"type": "output",
|
||
"name": "定向任务输出",
|
||
"config": {"mapping": {"text": f"{{{{ nodes.{agent.id}.data.text }}}}"}},
|
||
},
|
||
],
|
||
"edges": [
|
||
{"id": "__agent_task_start_to_agent__", "source": "__agent_task_start__", "target": agent.id},
|
||
{"id": "__agent_task_agent_to_output__", "source": agent.id, "target": "__agent_task_output__"},
|
||
],
|
||
}
|
||
)
|
||
|
||
|
||
async def _run_graph_snapshot(request: Request, run: dict[str, Any]) -> WorkflowGraph:
|
||
"""Return the immutable graph that actually produced ``run``.
|
||
|
||
Draft, planned, and targeted-agent runs keep their graph in ``context``;
|
||
a feedback revision must use that snapshot rather than today's editable
|
||
canvas or a newer published version.
|
||
"""
|
||
raw = (run.get("context") or {}).get("draftGraph")
|
||
if raw is None:
|
||
version = await _definitions(request).get_version(run["workflow_version_id"])
|
||
raw = (version or {}).get("graph")
|
||
try:
|
||
return WorkflowGraph.model_validate(raw or {})
|
||
except Exception as exc: # noqa: BLE001 - persisted malformed state must not be rerun
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={"code": "WORKFLOW_VERSION_NOT_FOUND", "message": "原运行的执行图不可恢复"},
|
||
) from exc
|
||
|
||
|
||
def _downstream_nodes(graph: WorkflowGraph, start_node_id: str) -> set[str]:
|
||
"""Return a node plus every ordinary downstream consumer of its output."""
|
||
adjacency: dict[str, list[str]] = {}
|
||
for edge in graph.edges:
|
||
adjacency.setdefault(edge.source, []).append(edge.target)
|
||
affected: set[str] = set()
|
||
pending = [start_node_id]
|
||
while pending:
|
||
current = pending.pop()
|
||
if current in affected:
|
||
continue
|
||
affected.add(current)
|
||
pending.extend(adjacency.get(current) or [])
|
||
return affected
|
||
|
||
|
||
# ── start / list / get / cancel / resume ────────────────────────────────
|
||
|
||
|
||
@router.post("/{workflow_id}/runs")
|
||
async def start_run(request: Request, workflow_id: str, body: StartRunBody) -> dict[str, Any]:
|
||
_require_enabled()
|
||
user_id = await get_current_user(request)
|
||
definitions = _definitions(request)
|
||
is_draft_run = str(body.target or "").strip().lower() == "draft"
|
||
planning_session_id = str(body.planning_session_id or "").strip()
|
||
proposal_id = str(body.proposal_id or "").strip()
|
||
if bool(planning_session_id) != bool(proposal_id):
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={"code": "WORKFLOW_PLANNING_INVALID", "message": "planningSessionId 和 proposalId 必须同时提供"},
|
||
)
|
||
is_planning_run = bool(planning_session_id)
|
||
if is_planning_run and body.target_agent_id:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={"code": "WORKFLOW_PLANNING_INVALID", "message": "确认候选流程时不能同时指定单个智能体"},
|
||
)
|
||
|
||
planning_session: dict[str, Any] | None = None
|
||
planning_graph_raw: dict[str, Any] | None = None
|
||
if is_planning_run:
|
||
planning_session = await _planning(request).get_session(planning_session_id)
|
||
if planning_session is None:
|
||
raise HTTPException(status_code=404, detail={"code": "WORKFLOW_PLANNING_NOT_FOUND", "message": "规划会话不存在"})
|
||
if planning_session.get("workflow_id") != workflow_id:
|
||
raise HTTPException(status_code=400, detail={"code": "WORKFLOW_PLANNING_INVALID", "message": "候选流程不属于当前工作流"})
|
||
if planning_session.get("owner_id") != user_id and not await _is_admin(request):
|
||
raise HTTPException(status_code=403, detail={"code": "WORKFLOW_FORBIDDEN", "message": "无权确认该候选流程"})
|
||
selected = next((item for item in planning_session.get("proposals") or [] if item.get("id") == proposal_id), None)
|
||
if selected is None:
|
||
raise HTTPException(status_code=404, detail={"code": "WORKFLOW_PROPOSAL_NOT_FOUND", "message": "候选流程不存在"})
|
||
if planning_session.get("status") == "confirmed":
|
||
if planning_session.get("selected_proposal_id") != proposal_id:
|
||
raise HTTPException(status_code=409, detail={"code": "WORKFLOW_PLANNING_CONFLICT", "message": "规划会话已确认了另一条候选流程"})
|
||
existing_run_id = planning_session.get("confirmed_run_id")
|
||
existing_run = await _runs(request).get_run(str(existing_run_id)) if existing_run_id else None
|
||
if existing_run is not None:
|
||
return {**_public_run(existing_run), "created": False}
|
||
raise HTTPException(status_code=409, detail={"code": "WORKFLOW_PLANNING_CONFLICT", "message": "规划会话已确认，请刷新后重试"})
|
||
if planning_session.get("selected_proposal_id") != proposal_id:
|
||
raise HTTPException(status_code=409, detail={"code": "WORKFLOW_PLANNING_CONFLICT", "message": "请先选择该候选流程"})
|
||
planning_graph_raw = selected.get("graph") if isinstance(selected.get("graph"), dict) else {}
|
||
|
||
if is_draft_run:
|
||
# Test-run semantics: execute the CURRENT draft graph (what the canvas
|
||
# shows right now), versioned as the literal "draft". The compiled
|
||
# execution graph travels in run.context (never the canvas document),
|
||
# and the executor prefers it over any published version.
|
||
definition = await definitions.get_definition(workflow_id, include_draft=True)
|
||
else:
|
||
definition = await definitions.get_definition(workflow_id, include_draft=False)
|
||
if definition is None:
|
||
raise HTTPException(status_code=404, detail="工作流不存在")
|
||
if definition.get("owner_id") != user_id and not await _is_admin(request):
|
||
raise HTTPException(status_code=403, detail={"code": "WORKFLOW_FORBIDDEN", "message": "无权运行该工作流"})
|
||
|
||
draft_graph_raw: dict[str, Any] | None = None
|
||
if is_draft_run:
|
||
draft_graph_raw = definition.get("draft_graph") or {}
|
||
if not draft_graph_raw.get("nodes"):
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={
|
||
"code": "WORKFLOW_DRAFT_EMPTY",
|
||
"message": "草稿还没有可运行的执行图，请先在画布保存一次",
|
||
},
|
||
)
|
||
|
||
version_id = body.version_id
|
||
version: dict[str, Any] | None = None
|
||
if is_planning_run:
|
||
# The graph itself is frozen into run.context; this version label is an
|
||
# audit marker and deliberately has no workflow_versions row.
|
||
version_id = f"plan:{planning_session_id}"
|
||
elif not version_id and not is_draft_run:
|
||
versions = await definitions.list_versions(workflow_id)
|
||
if not versions:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={"code": "WORKFLOW_VERSION_NOT_FOUND", "message": "工作流尚未发布任何版本"},
|
||
)
|
||
version_id = versions[0]["id"]
|
||
if is_draft_run:
|
||
version_id = "draft"
|
||
elif not is_planning_run:
|
||
version = await definitions.get_version(version_id)
|
||
if version is None or version.get("workflow_id") != workflow_id:
|
||
raise HTTPException(status_code=404, detail={"code": "WORKFLOW_VERSION_NOT_FOUND", "message": "版本不存在"})
|
||
|
||
try:
|
||
graph = WorkflowGraph.model_validate(
|
||
planning_graph_raw if is_planning_run else (draft_graph_raw if is_draft_run else (version.get("graph") or {})),
|
||
)
|
||
except Exception as exc: # noqa: BLE001 - malformed persisted graph must not reach a worker
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={
|
||
"code": "WORKFLOW_SCHEMA_INVALID",
|
||
"message": f"执行图不符合 WorkflowGraph v1.0: {exc}",
|
||
},
|
||
) from exc
|
||
|
||
target_agent_id = str(body.target_agent_id or "").strip()
|
||
model_name = _selected_model_name(body.model_name)
|
||
if target_agent_id:
|
||
graph = _agent_task_graph(graph, target_agent_id, model_name)
|
||
|
||
issues = await _run_validation_issues(
|
||
request,
|
||
workflow_id=workflow_id,
|
||
graph=graph,
|
||
user_id=user_id,
|
||
)
|
||
if issues:
|
||
first = issues[0].model_dump(by_alias=True)
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={
|
||
"code": first.get("code") or "WORKFLOW_SCHEMA_INVALID",
|
||
"message": f"当前执行图不可运行：{first.get('message') or '请检查节点配置'}",
|
||
"nodeId": first.get("nodeId"),
|
||
"details": {"issues": [issue.model_dump(by_alias=True) for issue in issues]},
|
||
},
|
||
)
|
||
inputs = body.resolved_inputs()
|
||
if is_planning_run and not inputs and planning_session is not None:
|
||
inputs = {"query": planning_session["query"]}
|
||
missing = [key for key in (graph.input_schema.get("required") or []) if inputs.get(str(key)) in (None, "")]
|
||
if missing:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={
|
||
"code": "WORKFLOW_INPUT_INVALID",
|
||
"message": f"缺少必填输入：{'、'.join(str(m) for m in missing)}",
|
||
"details": {"missing": missing},
|
||
},
|
||
)
|
||
# Full JSON Schema pass so type/enum/nested mistakes fail here with a
|
||
# structured 400 instead of surfacing mid-run as a node error.
|
||
schema_errors = schema_issues(inputs, graph.input_schema or {})
|
||
if schema_errors:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={
|
||
"code": "WORKFLOW_INPUT_INVALID",
|
||
"message": "输入不符合 schema：" + "；".join(e["message"] for e in schema_errors[:3]),
|
||
"details": {"errors": schema_errors},
|
||
},
|
||
)
|
||
|
||
runs = _runs(request)
|
||
active = [r for r in await runs.list_runs(owner_id=user_id, limit=200) if r["status"] in ("queued", "running")]
|
||
if len(active) >= _cfg().max_concurrent_runs_per_user:
|
||
raise HTTPException(
|
||
status_code=429,
|
||
detail={
|
||
"code": "WORKFLOW_LIMIT_EXCEEDED",
|
||
"message": f"同时运行的工作流已达上限（{_cfg().max_concurrent_runs_per_user}）",
|
||
},
|
||
)
|
||
|
||
# Scope the idempotency key per workflow so two workflows can share a client key.
|
||
key = f"{workflow_id}:{body.idempotency_key}" if body.idempotency_key else None
|
||
context: dict[str, Any] = {"depth": 0}
|
||
if is_draft_run or target_agent_id or is_planning_run:
|
||
# The draft snapshot rides with the run; the executor prefers it over
|
||
# the published version. Execution graph only — never the canvas doc.
|
||
# A directed agent task uses the same durable snapshot mechanism even
|
||
# when the source graph originated from a published version.
|
||
context["draftGraph"] = graph.model_dump(by_alias=True)
|
||
if target_agent_id:
|
||
target_node = next(node for node in graph.nodes if node.type == "agent")
|
||
context.update(
|
||
{
|
||
"executionMode": "agent_task",
|
||
"targetAgentId": target_agent_id,
|
||
"targetAgentName": target_node.name or target_agent_id,
|
||
}
|
||
)
|
||
if is_planning_run:
|
||
context.update(
|
||
{
|
||
"executionMode": "planned_workflow",
|
||
"planningSessionId": planning_session_id,
|
||
"proposalId": proposal_id,
|
||
}
|
||
)
|
||
run, created = await runs.create_run(
|
||
{
|
||
"workflow_id": workflow_id,
|
||
"workflow_version_id": version_id,
|
||
"owner_id": user_id,
|
||
"input": inputs,
|
||
"idempotency_key": key,
|
||
"context": context,
|
||
}
|
||
)
|
||
if created:
|
||
if is_planning_run:
|
||
try:
|
||
await _planning(request).confirm_proposal(
|
||
planning_session_id,
|
||
proposal_id,
|
||
run_id=run["id"],
|
||
)
|
||
except WorkflowPlanningProposalNotFoundError as exc:
|
||
await runs.finalize_cancel(run["id"])
|
||
raise HTTPException(status_code=409, detail={"code": "WORKFLOW_PLANNING_CONFLICT", "message": "候选流程已失效，请重新选择"}) from exc
|
||
except WorkflowPlanningConflictError as exc:
|
||
await runs.finalize_cancel(run["id"])
|
||
raise HTTPException(status_code=409, detail={"code": "WORKFLOW_PLANNING_CONFLICT", "message": str(exc)}) from exc
|
||
await _events(request).append(
|
||
run_id=run["id"],
|
||
workflow_id=workflow_id,
|
||
version_id=version_id,
|
||
event_type="run.created",
|
||
payload={
|
||
"inputs": inputs,
|
||
**({"executionMode": "agent_task", "targetAgentId": target_agent_id} if target_agent_id else {}),
|
||
**({"executionMode": "planned_workflow", "planningSessionId": planning_session_id, "proposalId": proposal_id} if is_planning_run else {}),
|
||
},
|
||
)
|
||
audit_workflow("run.start", user_id=user_id, workflowId=workflow_id, runId=run["id"], versionId=version_id)
|
||
_nudge(request)
|
||
return {**_public_run(run), "created": created}
|
||
|
||
|
||
@router.get("/{workflow_id}/runs")
|
||
async def list_workflow_runs(
|
||
request: Request,
|
||
workflow_id: str,
|
||
status: str | None = None,
|
||
limit: int = Query(default=50, ge=1, le=200),
|
||
offset: int = Query(default=0, ge=0),
|
||
) -> dict[str, Any]:
|
||
_require_enabled()
|
||
user_id = await get_current_user(request)
|
||
owner = None if await _is_admin(request) else user_id
|
||
rows = await _runs(request).list_runs(workflow_id=workflow_id, owner_id=owner, status=status, limit=limit, offset=offset)
|
||
return {"runs": [_public_run(r) for r in rows], "hasMore": len(rows) >= limit}
|
||
|
||
|
||
@router.get("/runs/{run_id}")
|
||
async def get_run(request: Request, run_id: str) -> dict[str, Any]:
|
||
_require_enabled()
|
||
run = await _authorized_run(request, run_id)
|
||
node_runs = await _runs(request).list_node_runs(run_id, current_only=True)
|
||
return {
|
||
**_public_run(run),
|
||
"nodeRuns": [
|
||
{
|
||
"nodeId": r["node_id"],
|
||
"nodeType": r["node_type"],
|
||
"status": r["status"],
|
||
"attempt": r["attempt"],
|
||
"iteration": r["iteration"],
|
||
"durationMs": r["duration_ms"],
|
||
"error": r["error"],
|
||
"startedAt": r["started_at"],
|
||
"finishedAt": r["finished_at"],
|
||
}
|
||
for r in node_runs
|
||
],
|
||
}
|
||
|
||
|
||
@router.get("/runs/{run_id}/artifacts")
|
||
async def list_run_artifacts(request: Request, run_id: str) -> dict[str, Any]:
|
||
_require_enabled()
|
||
await _authorized_run(request, run_id)
|
||
rows = await _runs(request).list_artifacts(run_id)
|
||
return {
|
||
"artifacts": [
|
||
{
|
||
"artifactId": r["id"],
|
||
"nodeId": r["node_id"],
|
||
"name": r["name"],
|
||
"mimeType": r["mime_type"],
|
||
"sizeBytes": r["size_bytes"],
|
||
"preview": r["preview"],
|
||
}
|
||
for r in rows
|
||
]
|
||
}
|
||
|
||
|
||
# ── artifact content / download ─────────────────────────────────────────
|
||
|
||
_PREVIEW_MAX_BYTES = 256 * 1024
|
||
_TEXT_MIME_PREFIXES = ("text/", "application/json", "application/javascript", "application/x-yaml", "application/xml", "application/csv")
|
||
|
||
|
||
def _is_textual(mime_type: str) -> bool:
|
||
return (mime_type or "").lower().startswith(_TEXT_MIME_PREFIXES)
|
||
|
||
|
||
def _read_prefix(path: str, limit: int) -> tuple[bytes, bool]:
|
||
"""Read up to ``limit + 1`` bytes; the extra byte detects truncation."""
|
||
with open(path, "rb") as handle:
|
||
data = handle.read(limit + 1)
|
||
return data[:limit], len(data) > limit
|
||
|
||
|
||
def _file_exists(path: str) -> bool:
|
||
import os
|
||
|
||
return bool(path) and os.path.isfile(path)
|
||
|
||
|
||
async def _authorized_artifact(request: Request, run_id: str, artifact_id: str) -> dict[str, Any]:
|
||
"""Run-level authorization first, then artifact-must-belong-to-run.
|
||
|
||
The file path is resolved exclusively from the registered artifact row —
|
||
clients address artifacts by id and can never pass a sandbox path in.
|
||
"""
|
||
await _authorized_run(request, run_id)
|
||
artifact = await _runs(request).get_artifact(run_id, artifact_id)
|
||
if artifact is None:
|
||
raise HTTPException(
|
||
status_code=404,
|
||
detail={"code": "WORKFLOW_ARTIFACT_NOT_FOUND", "message": "制品不存在"},
|
||
)
|
||
return artifact
|
||
|
||
|
||
@router.get("/runs/{run_id}/artifacts/{artifact_id}")
|
||
async def get_run_artifact_content(request: Request, run_id: str, artifact_id: str) -> dict[str, Any]:
|
||
"""Size-bounded text preview for file cards; binaries stay download-only."""
|
||
from starlette.concurrency import run_in_threadpool
|
||
|
||
from deerflow.workflows.events import artifact_kind
|
||
|
||
_require_enabled()
|
||
artifact = await _authorized_artifact(request, run_id, artifact_id)
|
||
mime = str(artifact.get("mime_type") or "application/octet-stream")
|
||
kind = artifact_kind(mime)
|
||
base = {
|
||
"artifactId": artifact["id"],
|
||
"name": artifact["name"],
|
||
"kind": kind,
|
||
"mimeType": mime,
|
||
"sizeBytes": artifact.get("size_bytes"),
|
||
}
|
||
|
||
path = str(artifact.get("path") or "")
|
||
file_ok = bool(path) and await run_in_threadpool(_file_exists, path)
|
||
if file_ok:
|
||
raw, truncated = (b"", False)
|
||
if _is_textual(mime):
|
||
raw, truncated = await run_in_threadpool(_read_prefix, path, _PREVIEW_MAX_BYTES)
|
||
return {
|
||
**base,
|
||
"text": raw.decode("utf-8", errors="replace"),
|
||
"truncated": truncated,
|
||
"source": "file",
|
||
}
|
||
return {
|
||
**base,
|
||
"binary": True,
|
||
"downloadUrl": f"/api/workflows/runs/{run_id}/artifacts/{artifact_id}/download",
|
||
"source": "file",
|
||
}
|
||
|
||
preview = artifact.get("preview")
|
||
if preview:
|
||
return {**base, "text": str(preview), "truncated": True, "source": "preview"}
|
||
raise HTTPException(
|
||
status_code=404,
|
||
detail={"code": "WORKFLOW_ARTIFACT_CONTENT_UNAVAILABLE", "message": "制品内容不可用（文件已过期）"},
|
||
)
|
||
|
||
|
||
@router.get("/runs/{run_id}/artifacts/{artifact_id}/download")
|
||
async def download_run_artifact(request: Request, run_id: str, artifact_id: str):
|
||
from urllib.parse import quote
|
||
|
||
from starlette.concurrency import run_in_threadpool
|
||
from starlette.responses import FileResponse, Response
|
||
|
||
_require_enabled()
|
||
artifact = await _authorized_artifact(request, run_id, artifact_id)
|
||
mime = str(artifact.get("mime_type") or "application/octet-stream")
|
||
name = str(artifact.get("name") or artifact_id)
|
||
disposition = f"attachment; filename*=UTF-8''{quote(name)}"
|
||
path = str(artifact.get("path") or "")
|
||
|
||
if path and await run_in_threadpool(_file_exists, path):
|
||
return FileResponse(path, media_type=mime, headers={"Content-Disposition": disposition})
|
||
preview = artifact.get("preview")
|
||
if preview:
|
||
return Response(
|
||
content=str(preview),
|
||
media_type=mime if _is_textual(mime) else "text/plain",
|
||
headers={"Content-Disposition": disposition},
|
||
)
|
||
raise HTTPException(
|
||
status_code=404,
|
||
detail={"code": "WORKFLOW_ARTIFACT_CONTENT_UNAVAILABLE", "message": "制品文件已过期，无法下载"},
|
||
)
|
||
|
||
|
||
@router.post("/runs/{run_id}/cancel")
|
||
async def cancel_run(request: Request, run_id: str) -> dict[str, Any]:
|
||
_require_enabled()
|
||
run = await _authorized_run(request, run_id)
|
||
if run["status"] in _TERMINAL_STATUSES:
|
||
return {**_public_run(run), "cancelled": False}
|
||
updated = await _runs(request).request_cancel(run_id)
|
||
if updated is None:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={"code": "WORKFLOW_RUN_CONFLICT", "message": "当前状态无法取消"},
|
||
)
|
||
await _events(request).append(
|
||
run_id=run_id,
|
||
workflow_id=run["workflow_id"],
|
||
version_id=run["workflow_version_id"],
|
||
event_type="run.cancel_requested",
|
||
payload={},
|
||
)
|
||
_nudge(request)
|
||
audit_workflow("run.cancel", user_id=await get_current_user(request), runId=run_id, workflowId=run["workflow_id"])
|
||
return {**_public_run(updated), "cancelled": True}
|
||
|
||
|
||
@router.post("/runs/{run_id}/resume")
|
||
async def resume_run(request: Request, run_id: str, body: ResumeRunBody) -> dict[str, Any]:
|
||
_require_enabled()
|
||
run = await _authorized_run(request, run_id)
|
||
if run["status"] != "awaiting_input":
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={"code": "WORKFLOW_RUN_NOT_RESUMABLE", "message": "运行当前不在等待人工输入"},
|
||
)
|
||
pending = run.get("pending_input") or {}
|
||
node_id = body.node_id or str(pending.get("nodeId") or "")
|
||
if not node_id:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={"code": "WORKFLOW_HUMAN_INPUT_INVALID", "message": "缺少 nodeId"},
|
||
)
|
||
|
||
runs = _runs(request)
|
||
# Single conditional update: token validation + payload write + token
|
||
# consumption + status flip all in one statement, so two concurrent resume
|
||
# requests with the same token cannot interleave — exactly one succeeds and
|
||
# the loser gets 409 without ever touching the run.
|
||
resumed = await runs.resume_with_payload(
|
||
run_id,
|
||
body.resume_token,
|
||
{
|
||
"nodeId": node_id,
|
||
"action": body.action,
|
||
"values": body.resolved_values(),
|
||
**({"message": body.message} if body.message else {}),
|
||
},
|
||
context=dict(run.get("context") or {}),
|
||
)
|
||
if resumed is None:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={"code": "WORKFLOW_RESUME_TOKEN_INVALID", "message": "恢复令牌无效或已被使用"},
|
||
)
|
||
await _events(request).append(
|
||
run_id=run_id,
|
||
workflow_id=run["workflow_id"],
|
||
version_id=run["workflow_version_id"],
|
||
event_type="run.resumed",
|
||
payload={"nodeId": node_id, "action": body.action},
|
||
node_id=node_id,
|
||
)
|
||
_nudge(request)
|
||
audit_workflow("run.resume", user_id=await get_current_user(request), runId=run_id, nodeId=node_id)
|
||
return _public_run(resumed)
|
||
|
||
|
||
@router.post("/runs/{run_id}/retry", status_code=201)
|
||
async def retry_run(request: Request, run_id: str) -> dict[str, Any]:
|
||
"""Start a new run from a failed one.
|
||
|
||
Completed node outputs are copied onto the new run so the engine resumes
|
||
from the first unfinished node. If there is nothing to replay, the new run
|
||
is a full replay of the same version and inputs, still marked
|
||
``retry_of_run_id``.
|
||
"""
|
||
_require_enabled()
|
||
run = await _authorized_run(request, run_id)
|
||
if run["status"] != "failed":
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={"code": "WORKFLOW_RUN_NOT_RETRYABLE", "message": "只有失败的运行可以重试"},
|
||
)
|
||
|
||
user_id = await get_current_user(request)
|
||
runs = _runs(request)
|
||
active = [r for r in await runs.list_runs(owner_id=user_id, limit=200) if r["status"] in ("queued", "running")]
|
||
if len(active) >= _cfg().max_concurrent_runs_per_user:
|
||
raise HTTPException(
|
||
status_code=429,
|
||
detail={
|
||
"code": "WORKFLOW_LIMIT_EXCEEDED",
|
||
"message": f"同时运行的工作流已达上限（{_cfg().max_concurrent_runs_per_user}）",
|
||
},
|
||
)
|
||
|
||
context = dict(run.get("context") or {})
|
||
context.pop("resume_payload", None)
|
||
context["retry_of_run_id"] = run_id
|
||
new_run, _created = await runs.create_run(
|
||
{
|
||
"workflow_id": run["workflow_id"],
|
||
"workflow_version_id": run["workflow_version_id"],
|
||
"owner_id": user_id,
|
||
"input": run.get("input") or {},
|
||
"retry_of_run_id": run_id,
|
||
"context": context,
|
||
}
|
||
)
|
||
for row in await runs.list_node_runs(run_id, current_only=True):
|
||
if row.get("status") != "completed":
|
||
continue
|
||
await runs.record_node_run(
|
||
{
|
||
"run_id": new_run["id"],
|
||
"node_id": row["node_id"],
|
||
"node_type": row.get("node_type") or "unknown",
|
||
"status": "completed",
|
||
"attempt": 1,
|
||
"iteration": int(row.get("iteration") or 0),
|
||
"output": row.get("output"),
|
||
"duration_ms": row.get("duration_ms"),
|
||
"is_current": True,
|
||
}
|
||
)
|
||
await _events(request).append(
|
||
run_id=new_run["id"],
|
||
workflow_id=new_run["workflow_id"],
|
||
version_id=new_run["workflow_version_id"],
|
||
event_type="run.created",
|
||
payload={"retryOfRunId": run_id, "inputs": new_run.get("input") or {}},
|
||
)
|
||
_nudge(request)
|
||
audit_workflow("run.retry", user_id=user_id, runId=new_run["id"], retryOfRunId=run_id)
|
||
return {**_public_run(new_run), "created": True}
|
||
|
||
|
||
@router.post("/runs/{run_id}/feedback", status_code=201)
|
||
async def revise_run_from_node_feedback(
|
||
request: Request,
|
||
run_id: str,
|
||
body: FeedbackRunBody,
|
||
) -> dict[str, Any]:
|
||
"""Create an auditable local rerun from feedback on one Agent/Skill node.
|
||
|
||
Completed snapshots in the original run are immutable. Instead of erasing
|
||
them, this creates a revision run, copies completed nodes outside the
|
||
target's downstream impact set, and re-executes the target plus its
|
||
consumers. The feedback is injected only into that target's prompt through
|
||
``RunContext.env.nodeFeedback``. This first public slice deliberately
|
||
accepts terminal, non-loop graphs only; a live run must be paused with a
|
||
normal ``human_input`` card or explicitly cancelled before revision so two
|
||
workers can never produce competing side effects.
|
||
"""
|
||
_require_enabled()
|
||
run = await _authorized_run(request, run_id)
|
||
if run["status"] not in _TERMINAL_STATUSES:
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={
|
||
"code": "WORKFLOW_RUN_CONFLICT",
|
||
"message": "运行仍在执行；请先使用人工协助卡补充信息，或停止后再发起局部重跑",
|
||
},
|
||
)
|
||
|
||
graph = await _run_graph_snapshot(request, run)
|
||
if any(node.type == "loop" for node in graph.nodes):
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={
|
||
"code": "WORKFLOW_RUN_CONFLICT",
|
||
"message": "包含循环节点的流程暂不支持定点重跑，请重新执行整条流程",
|
||
},
|
||
)
|
||
node_id = body.node_id.strip()
|
||
target = graph.node_map().get(node_id)
|
||
if target is None:
|
||
raise HTTPException(
|
||
status_code=404,
|
||
detail={"code": "WORKFLOW_NODE_NOT_FOUND", "message": "指定节点不在原运行图中"},
|
||
)
|
||
if target.type not in {"agent", "skill"}:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={
|
||
"code": "WORKFLOW_NODE_NOT_FOUND",
|
||
"message": "目前只能对智能体或技能节点发起定点反馈",
|
||
"nodeId": node_id,
|
||
},
|
||
)
|
||
|
||
feedback_message = body.message.strip()
|
||
if not feedback_message:
|
||
raise HTTPException(
|
||
status_code=400,
|
||
detail={"code": "WORKFLOW_INPUT_INVALID", "message": "修订说明不能为空"},
|
||
)
|
||
|
||
user_id = await get_current_user(request)
|
||
issues = await _run_validation_issues(
|
||
request,
|
||
workflow_id=run["workflow_id"],
|
||
graph=graph,
|
||
user_id=user_id,
|
||
)
|
||
if issues:
|
||
first = issues[0].model_dump(by_alias=True)
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={
|
||
"code": first.get("code") or "WORKFLOW_SCHEMA_INVALID",
|
||
"message": "原执行图的依赖已不可用，无法创建修订运行",
|
||
"nodeId": first.get("nodeId"),
|
||
"details": {"issues": [issue.model_dump(by_alias=True) for issue in issues]},
|
||
},
|
||
)
|
||
|
||
current_rows = await _runs(request).list_node_runs(run_id, current_only=True)
|
||
target_row = next((row for row in current_rows if row.get("node_id") == node_id), None)
|
||
if target_row is None or target_row.get("status") != "completed":
|
||
raise HTTPException(
|
||
status_code=409,
|
||
detail={
|
||
"code": "WORKFLOW_RUN_CONFLICT",
|
||
"message": "只能对原运行中已完成的智能体节点发起定点反馈",
|
||
"nodeId": node_id,
|
||
},
|
||
)
|
||
|
||
runs = _runs(request)
|
||
active = [row for row in await runs.list_runs(owner_id=user_id, limit=200) if row["status"] in ("queued", "running")]
|
||
if len(active) >= _cfg().max_concurrent_runs_per_user:
|
||
raise HTTPException(
|
||
status_code=429,
|
||
detail={
|
||
"code": "WORKFLOW_LIMIT_EXCEEDED",
|
||
"message": f"同时运行的工作流已达上限（{_cfg().max_concurrent_runs_per_user}）",
|
||
},
|
||
)
|
||
|
||
affected = _downstream_nodes(graph, node_id)
|
||
reusable_rows = [
|
||
row
|
||
for row in current_rows
|
||
if row.get("status") == "completed" and row.get("node_id") not in affected
|
||
]
|
||
reused_node_ids = sorted(str(row["node_id"]) for row in reusable_rows)
|
||
fingerprint = hashlib.sha256(
|
||
f"{run_id}\x1f{node_id}\x1f{feedback_message}".encode()
|
||
).hexdigest()[:24]
|
||
# Do not truncate an arbitrary 128-character client key: that could make
|
||
# two different keys with the same prefix collide. Hashing preserves a
|
||
# compact, stable key for both client-supplied retries and server-derived
|
||
# semantic retries.
|
||
client_key = str(body.idempotency_key or fingerprint)
|
||
client_fingerprint = hashlib.sha256(client_key.encode()).hexdigest()[:24]
|
||
idempotency_key = f"revision:{run_id}:{client_fingerprint}"
|
||
parent_context = dict(run.get("context") or {})
|
||
parent_context.pop("resume_payload", None)
|
||
parent_context.pop("loop_state", None)
|
||
parent_env = parent_context.get("env")
|
||
env = dict(parent_env) if isinstance(parent_env, dict) else {}
|
||
env["nodeFeedback"] = {
|
||
node_id: {
|
||
"message": feedback_message,
|
||
"sourceRunId": run_id,
|
||
}
|
||
}
|
||
parent_context.update(
|
||
{
|
||
"env": env,
|
||
"revisionOfRunId": run_id,
|
||
"feedback": {
|
||
"targetNodeId": node_id,
|
||
"message": feedback_message,
|
||
"affectedNodeIds": sorted(affected),
|
||
"reusedNodeIds": reused_node_ids,
|
||
"submittedAt": datetime.now(UTC).isoformat(),
|
||
},
|
||
}
|
||
)
|
||
new_run, created = await runs.create_run(
|
||
{
|
||
"workflow_id": run["workflow_id"],
|
||
"workflow_version_id": run["workflow_version_id"],
|
||
"owner_id": user_id,
|
||
"input": run.get("input") or {},
|
||
"idempotency_key": idempotency_key,
|
||
"retry_of_run_id": run_id,
|
||
"context": parent_context,
|
||
}
|
||
)
|
||
if not created:
|
||
return {**_public_run(new_run), "created": False, "reused": True}
|
||
|
||
for row in reusable_rows:
|
||
await runs.record_node_run(
|
||
{
|
||
"run_id": new_run["id"],
|
||
"node_id": row["node_id"],
|
||
"node_type": row.get("node_type") or "unknown",
|
||
"status": "completed",
|
||
"attempt": 1,
|
||
"iteration": int(row.get("iteration") or 0),
|
||
"output": row.get("output"),
|
||
"duration_ms": row.get("duration_ms"),
|
||
"is_current": True,
|
||
}
|
||
)
|
||
await _events(request).append(
|
||
run_id=new_run["id"],
|
||
workflow_id=new_run["workflow_id"],
|
||
version_id=new_run["workflow_version_id"],
|
||
event_type="run.created",
|
||
payload={
|
||
"revisionOfRunId": run_id,
|
||
"targetNodeId": node_id,
|
||
"affectedNodeIds": sorted(affected),
|
||
"reusedNodeIds": reused_node_ids,
|
||
},
|
||
)
|
||
_nudge(request)
|
||
audit_workflow(
|
||
"run.feedback_revision",
|
||
user_id=user_id,
|
||
runId=new_run["id"],
|
||
revisionOfRunId=run_id,
|
||
nodeId=node_id,
|
||
affectedNodeIds=sorted(affected),
|
||
)
|
||
return {
|
||
**_public_run(new_run),
|
||
"created": True,
|
||
"reused": False,
|
||
"affectedNodeIds": sorted(affected),
|
||
"reusedNodeIds": reused_node_ids,
|
||
}
|
||
|
||
|
||
# ── events / SSE ────────────────────────────────────────────────────────
|
||
|
||
|
||
@router.get("/runs/{run_id}/events")
|
||
async def list_run_events(
|
||
request: Request,
|
||
run_id: str,
|
||
after: int = Query(default=0, ge=0),
|
||
after_seq: int | None = Query(default=None, alias="after_seq", ge=0),
|
||
limit: int = Query(default=_REPLAY_PAGE, ge=1, le=1000),
|
||
) -> dict[str, Any]:
|
||
_require_enabled()
|
||
await _authorized_run(request, run_id)
|
||
cursor = after_seq if after_seq is not None else after
|
||
rows = await _events(request).list_after(run_id, after_seq=cursor, limit=limit)
|
||
return {"events": [envelope_from_row(r) for r in rows], "lastSeq": rows[-1]["seq"] if rows else cursor}
|
||
|
||
|
||
def _frame(payload: dict[str, Any]) -> str:
|
||
body = json.dumps(payload, ensure_ascii=False, separators=(",", ":"), default=str)
|
||
return f"id: {payload.get('seq', 0)}\nevent: {payload.get('event', 'message')}\ndata: {body}\n\n"
|
||
|
||
|
||
def _replay_cursor(request: Request, after: int) -> int:
|
||
"""``Last-Event-ID`` wins over ``?after=``; invalid headers fall back."""
|
||
raw = request.headers.get("last-event-id")
|
||
if raw:
|
||
try:
|
||
return max(0, int(str(raw).strip()))
|
||
except (TypeError, ValueError):
|
||
pass
|
||
return after
|
||
|
||
|
||
@router.get("/runs/{run_id}/stream")
|
||
async def stream_run(
|
||
request: Request,
|
||
run_id: str,
|
||
after: int = Query(default=0, ge=0),
|
||
after_seq: int | None = Query(default=None, alias="after_seq", ge=0),
|
||
) -> StreamingResponse:
|
||
"""Replay from ``after`` then tail live.
|
||
|
||
The database is authoritative for ordering: live frames whose ``seq`` we have
|
||
not yet replayed are held back and picked up by the next DB poll, so a
|
||
reconnecting client never sees a gap or an out-of-order ``seq``.
|
||
``Last-Event-ID`` takes priority over ``after``.
|
||
"""
|
||
_require_enabled()
|
||
run = await _authorized_run(request, run_id)
|
||
events = _events(request)
|
||
runs = _runs(request)
|
||
hub = getattr(request.app.state, "workflow_live_hub", None)
|
||
cursor_start = _replay_cursor(request, after_seq if after_seq is not None else after)
|
||
|
||
async def generator():
|
||
cursor = cursor_start
|
||
live_queue = hub.subscribe(run_id) if hub is not None else None
|
||
loop = asyncio.get_running_loop()
|
||
deadline = loop.time() + _STREAM_DEADLINE_SECONDS
|
||
next_poll = 0.0
|
||
last_beat = loop.time()
|
||
try:
|
||
yield f": stream-ready{_SSE_FLUSH_PADDING}\n\n"
|
||
while loop.time() < deadline:
|
||
if await request.is_disconnected():
|
||
return
|
||
|
||
if loop.time() >= next_poll:
|
||
while True:
|
||
rows = await events.list_after(run_id, after_seq=cursor, limit=_REPLAY_PAGE)
|
||
if not rows:
|
||
break
|
||
for row in rows:
|
||
yield _frame(envelope_from_row(row))
|
||
cursor = max(cursor, int(row["seq"]))
|
||
last_beat = loop.time()
|
||
if row["event_type"] in _TERMINAL_EVENTS:
|
||
return
|
||
if len(rows) < _REPLAY_PAGE:
|
||
break
|
||
current = await runs.get_run(run_id)
|
||
if current is None:
|
||
return
|
||
if current["status"] in _TERMINAL_STATUSES:
|
||
# Terminal state with no terminal event (e.g. a crash
|
||
# before the event landed): synthesise a closing frame
|
||
# so the client is never left hanging.
|
||
yield _frame(
|
||
{
|
||
"schemaVersion": "1.0",
|
||
"runId": run_id,
|
||
"workflowId": current["workflow_id"],
|
||
"versionId": current["workflow_version_id"],
|
||
"seq": cursor,
|
||
"event": f"run.{current['status']}",
|
||
"timestamp": datetime.now(UTC).isoformat(),
|
||
"data": {
|
||
"status": current["status"],
|
||
"output": current.get("output"),
|
||
"error": current.get("error"),
|
||
"synthesised": True,
|
||
},
|
||
}
|
||
)
|
||
return
|
||
next_poll = loop.time() + _DB_POLL_SECONDS
|
||
|
||
if live_queue is not None:
|
||
try:
|
||
event = await asyncio.wait_for(live_queue.get(), timeout=_LIVE_WAIT_SECONDS)
|
||
except TimeoutError:
|
||
event = None
|
||
if event is not None and int(event.get("seq") or 0) > cursor:
|
||
yield _frame(event)
|
||
cursor = int(event["seq"])
|
||
last_beat = loop.time()
|
||
if event.get("event") in _TERMINAL_EVENTS:
|
||
return
|
||
else:
|
||
await asyncio.sleep(_LIVE_WAIT_SECONDS)
|
||
|
||
if loop.time() - last_beat >= _HEARTBEAT_SECONDS:
|
||
yield ": heartbeat\n\n"
|
||
last_beat = loop.time()
|
||
finally:
|
||
if live_queue is not None and hub is not None:
|
||
hub.unsubscribe(run_id, live_queue)
|
||
|
||
return StreamingResponse(
|
||
generator(),
|
||
media_type="text/event-stream",
|
||
headers={
|
||
"Cache-Control": "no-cache, no-transform",
|
||
"Connection": "keep-alive",
|
||
"X-Accel-Buffering": "no",
|
||
"X-Workflow-Run-Status": run["status"],
|
||
},
|
||
)
|