719 lines
28 KiB
Python
719 lines
28 KiB
Python
"""Workflow run APIs: start / inspect / cancel / resume / retry / events / SSE stream."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
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 deerflow.config.app_config import get_app_config
|
||
from deerflow.persistence.workflow_events.base import WorkflowEventStore
|
||
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
|
||
|
||
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
|
||
|
||
|
||
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
|
||
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
|
||
|
||
|
||
# ── 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 _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 _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."""
|
||
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),
|
||
}
|
||
|
||
|
||
# ── 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"
|
||
|
||
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
|
||
if 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"
|
||
else:
|
||
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": "版本不存在"})
|
||
|
||
graph = WorkflowGraph.model_validate(draft_graph_raw if is_draft_run else (version.get("graph") or {}))
|
||
inputs = body.resolved_inputs()
|
||
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:
|
||
# The draft snapshot rides with the run; the executor prefers it over
|
||
# the published version. Execution graph only — never the canvas doc.
|
||
context["draftGraph"] = draft_graph_raw
|
||
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:
|
||
await _events(request).append(
|
||
run_id=run["id"],
|
||
workflow_id=workflow_id,
|
||
version_id=version_id,
|
||
event_type="run.created",
|
||
payload={"inputs": inputs},
|
||
)
|
||
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}
|
||
|
||
|
||
# ── 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:
|
||
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"],
|
||
},
|
||
)
|