573 lines
23 KiB
Python
573 lines
23 KiB
Python
"""Run lifecycle service layer.
|
||
|
||
Centralizes the business logic for creating runs, formatting SSE
|
||
frames, and consuming stream bridge events. Router modules
|
||
(``thread_runs``, ``runs``) are thin HTTP handlers that delegate here.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
import re
|
||
from collections.abc import Mapping
|
||
from typing import Any
|
||
|
||
from fastapi import HTTPException, Request
|
||
from langchain_core.messages import HumanMessage
|
||
|
||
from app.gateway.deps import get_run_context, get_run_manager, get_stream_bridge
|
||
from app.gateway.utils import sanitize_log_param
|
||
from deerflow.config.system_settings import load_system_settings
|
||
from deerflow.runtime import (
|
||
END_SENTINEL,
|
||
HEARTBEAT_SENTINEL,
|
||
ConflictError,
|
||
DisconnectMode,
|
||
RunManager,
|
||
RunRecord,
|
||
RunStatus,
|
||
StreamBridge,
|
||
UnsupportedStrategyError,
|
||
run_agent,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# SSE formatting
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def format_sse(event: str, data: Any, *, event_id: str | None = None) -> str:
|
||
"""Format a single SSE frame.
|
||
|
||
Field order: ``event:`` -> ``data:`` -> ``id:`` (optional) -> blank line.
|
||
This matches the LangGraph Platform wire format consumed by the
|
||
``useStream`` React hook and the Python ``langgraph-sdk`` SSE decoder.
|
||
"""
|
||
payload = json.dumps(data, default=str, ensure_ascii=False)
|
||
parts = [f"event: {event}", f"data: {payload}"]
|
||
if event_id:
|
||
parts.append(f"id: {event_id}")
|
||
parts.append("")
|
||
parts.append("")
|
||
return "\n".join(parts)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Input / config helpers
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
def normalize_stream_modes(raw: list[str] | str | None) -> list[str]:
|
||
"""Normalize the stream_mode parameter to a list.
|
||
|
||
Default matches what ``useStream`` expects: values + messages-tuple.
|
||
"""
|
||
if raw is None:
|
||
return ["values"]
|
||
if isinstance(raw, str):
|
||
return [raw]
|
||
return raw if raw else ["values"]
|
||
|
||
|
||
def normalize_input(raw_input: dict[str, Any] | None) -> dict[str, Any]:
|
||
"""Convert LangGraph Platform input format to LangChain state dict."""
|
||
if raw_input is None:
|
||
return {}
|
||
messages = raw_input.get("messages")
|
||
if messages and isinstance(messages, list):
|
||
converted = []
|
||
for msg in messages:
|
||
if isinstance(msg, dict):
|
||
role = msg.get("role", msg.get("type", "user"))
|
||
content = msg.get("content", "")
|
||
if role in ("user", "human"):
|
||
converted.append(HumanMessage(content=content))
|
||
else:
|
||
# TODO: handle other message types (system, ai, tool)
|
||
converted.append(HumanMessage(content=content))
|
||
else:
|
||
converted.append(msg)
|
||
return {**raw_input, "messages": converted}
|
||
return raw_input
|
||
|
||
|
||
_DEFAULT_ASSISTANT_ID = "lead_agent"
|
||
|
||
# Built-in LangGraph graph IDs registered in langgraph.json.
|
||
# These are NOT stored in the agents DB and must bypass the custom-agent access check.
|
||
_BUILTIN_GRAPH_IDS: frozenset[str] = frozenset({"lead_agent", "canvas_agent", "ai_writing"})
|
||
|
||
_RAG_CITATION_PROMPT = """当前对话已开启参考文献模式。
|
||
如果本轮调用了搜索、知识库、检索类技能或工具，请基于本轮所有检索结果整体回答。
|
||
你必须把同一轮内多次检索得到的资料视为同一批参考文献，按出现顺序从 [1] 开始连续编号，不要让每个检索词各自从 [1] 重新编号。
|
||
回答正文中的关键结论必须标注对应编号，例如 [1]、[2]，多个来源可写为 [1][3][5]。
|
||
不要引用不存在的编号，不要使用技能内部自带的编号，不要把参考文献完整列表堆在正文末尾。
|
||
如果没有任何可用检索结果，请直接说明未检索到可靠参考资料。"""
|
||
|
||
|
||
def _prepend_prompt_prefix(graph_input: dict[str, Any], prefix: str) -> dict[str, Any]:
|
||
"""Return ``graph_input`` with the prefix *tagged* onto its first HumanMessage.
|
||
|
||
The prefix is stored in ``additional_kwargs['prompt_prefix']`` rather than
|
||
concatenated into ``content``. This keeps the persisted/displayed message
|
||
clean — the user never sees the prefix — while ``PromptPrefixMiddleware``
|
||
injects it into the content of the request actually sent to the LLM.
|
||
|
||
Leaves the input unchanged when there are no messages or the first message
|
||
is not a human turn.
|
||
"""
|
||
if not prefix:
|
||
return graph_input
|
||
messages = graph_input.get("messages")
|
||
if not isinstance(messages, list) or not messages:
|
||
return graph_input
|
||
new_messages = list(messages)
|
||
for idx, msg in enumerate(new_messages):
|
||
if isinstance(msg, HumanMessage):
|
||
extra = dict(getattr(msg, "additional_kwargs", None) or {})
|
||
extra["prompt_prefix"] = prefix
|
||
new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra})
|
||
break
|
||
return {**graph_input, "messages": new_messages}
|
||
|
||
|
||
def _prompt_prefix_opted_in(body: Any) -> bool:
|
||
"""Detect whether the caller asked for the prompt prefix to be applied."""
|
||
context = getattr(body, "context", None)
|
||
if isinstance(context, Mapping) and context.get("prompt_prefix_enabled"):
|
||
return True
|
||
metadata = getattr(body, "metadata", None)
|
||
if isinstance(metadata, Mapping) and metadata.get("prompt_prefix_enabled"):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _rag_mode_requested(body: Any) -> bool:
|
||
context = getattr(body, "context", None)
|
||
if isinstance(context, Mapping) and context.get("rag_mode_enabled"):
|
||
return True
|
||
metadata = getattr(body, "metadata", None)
|
||
if isinstance(metadata, Mapping) and (
|
||
metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
|
||
):
|
||
return True
|
||
return False
|
||
|
||
|
||
def _thread_rag_mode_enabled(record: Any) -> bool:
|
||
if not isinstance(record, Mapping):
|
||
return False
|
||
metadata = record.get("metadata")
|
||
return isinstance(metadata, Mapping) and bool(
|
||
metadata.get("rag_mode_enabled") or metadata.get("rag_mode")
|
||
)
|
||
|
||
|
||
def _append_invisible_prompt(graph_input: dict[str, Any], prompt: str) -> dict[str, Any]:
|
||
messages = graph_input.get("messages")
|
||
if not isinstance(messages, list) or not messages:
|
||
return graph_input
|
||
new_messages = list(messages)
|
||
for idx, msg in enumerate(new_messages):
|
||
if not isinstance(msg, HumanMessage):
|
||
continue
|
||
extra = dict(getattr(msg, "additional_kwargs", None) or {})
|
||
existing = str(extra.get("prompt_prefix") or "").strip()
|
||
extra["prompt_prefix"] = f"{existing}\n\n{prompt}".strip() if existing else prompt
|
||
new_messages[idx] = msg.model_copy(update={"additional_kwargs": extra})
|
||
break
|
||
return {**graph_input, "messages": new_messages}
|
||
|
||
|
||
# Whitelist of run-context keys that the langgraph-compat layer forwards from
|
||
# ``body.context`` into the run config. ``config["context"]`` exists in
|
||
# LangGraph >=0.6, but these values must be written to both ``configurable``
|
||
# (for legacy ``_get_runtime_config`` consumers) and ``context`` because
|
||
# LangGraph >=1.1.9 no longer makes ``ToolRuntime.context`` fall back to
|
||
# ``configurable`` for consumers like ``setup_agent``.
|
||
_CONTEXT_CONFIGURABLE_KEYS: frozenset[str] = frozenset(
|
||
{
|
||
"agent_id",
|
||
"model_name",
|
||
"mode",
|
||
"thinking_enabled",
|
||
"reasoning_effort",
|
||
"rag_mode_enabled",
|
||
"writing_mode",
|
||
"writing_artifact_path",
|
||
"canvas_artifact_title",
|
||
"canvas_artifact_type",
|
||
"is_plan_mode",
|
||
"subagent_enabled",
|
||
"max_concurrent_subagents",
|
||
"agent_name",
|
||
"is_bootstrap",
|
||
"excluded_tools",
|
||
"skill_stop_names",
|
||
"skill_stop_message",
|
||
}
|
||
)
|
||
|
||
|
||
def merge_run_context_overrides(config: dict[str, Any], context: Mapping[str, Any] | None) -> None:
|
||
"""Merge whitelisted keys from ``body.context`` into both ``config['configurable']``
|
||
and ``config['context']`` so they are visible to legacy configurable readers and
|
||
to LangGraph ``ToolRuntime.context`` consumers (e.g. the ``setup_agent`` tool —
|
||
see issue #2677)."""
|
||
if not context:
|
||
return
|
||
configurable = config.setdefault("configurable", {})
|
||
runtime_context = config.setdefault("context", {})
|
||
for key in _CONTEXT_CONFIGURABLE_KEYS:
|
||
if key in context:
|
||
if isinstance(configurable, dict):
|
||
configurable.setdefault(key, context[key])
|
||
if isinstance(runtime_context, dict):
|
||
runtime_context.setdefault(key, context[key])
|
||
|
||
|
||
def resolve_agent_factory(assistant_id: str | None):
|
||
"""Resolve the agent factory callable from config.
|
||
|
||
Built-in graphs (``lead_agent``, ``canvas_agent``, ``ai_writing``) are dispatched
|
||
directly to their own factory. Custom agents share the ``lead_agent`` factory
|
||
with ``agent_id`` injected into configurable/context by :func:`build_run_config`.
|
||
"""
|
||
if assistant_id == "canvas_agent":
|
||
from deerflow.agents.canvas_agent.agent import make_canvas_agent
|
||
|
||
return make_canvas_agent
|
||
|
||
if assistant_id == "ai_writing":
|
||
from deerflow.agents.ai_writing.graph import make_ai_writing_graph
|
||
|
||
return make_ai_writing_graph
|
||
|
||
from deerflow.agents.lead_agent.agent import make_lead_agent
|
||
|
||
return make_lead_agent
|
||
|
||
|
||
def build_run_config(
|
||
thread_id: str,
|
||
request_config: dict[str, Any] | None,
|
||
metadata: dict[str, Any] | None,
|
||
*,
|
||
assistant_id: str | None = None,
|
||
) -> dict[str, Any]:
|
||
"""Build a RunnableConfig dict for the agent.
|
||
|
||
When *assistant_id* refers to a custom agent (anything other than
|
||
``"lead_agent"`` / ``None``), the id is forwarded as ``agent_id`` in
|
||
whichever runtime options container is active: ``context`` for
|
||
LangGraph >= 0.6.0 requests, otherwise ``configurable``.
|
||
``make_lead_agent`` reads this key to load the matching
|
||
``agents/<id>/SOUL.md`` and per-agent config — without it the agent
|
||
silently runs as the default lead agent.
|
||
|
||
This mirrors the channel manager's ``_resolve_run_params`` logic so that
|
||
the LangGraph Platform-compatible HTTP API and the IM channel path behave
|
||
identically.
|
||
"""
|
||
config: dict[str, Any] = {"recursion_limit": 100}
|
||
if request_config:
|
||
# LangGraph >= 0.6.0 introduced ``context`` as the preferred way to
|
||
# pass thread-level data and rejects requests that include both
|
||
# ``configurable`` and ``context``. If the caller already sends
|
||
# ``context``, honour it and skip our own ``configurable`` dict.
|
||
if "context" in request_config:
|
||
if "configurable" in request_config:
|
||
logger.warning(
|
||
"build_run_config: client sent both 'context' and 'configurable'; preferring 'context' (LangGraph >= 0.6.0). thread_id=%s, caller_configurable keys=%s",
|
||
thread_id,
|
||
list(request_config.get("configurable", {}).keys()),
|
||
)
|
||
context_value = request_config["context"]
|
||
if context_value is None:
|
||
context = {}
|
||
elif isinstance(context_value, Mapping):
|
||
context = dict(context_value)
|
||
else:
|
||
raise ValueError("request config 'context' must be a mapping or null.")
|
||
config["context"] = context
|
||
else:
|
||
configurable = {"thread_id": thread_id}
|
||
configurable.update(request_config.get("configurable", {}))
|
||
config["configurable"] = configurable
|
||
for k, v in request_config.items():
|
||
if k not in ("configurable", "context"):
|
||
config[k] = v
|
||
else:
|
||
config["configurable"] = {"thread_id": thread_id}
|
||
|
||
# Inject custom agent id when the caller specified a non-default assistant.
|
||
# Built-in graphs (lead_agent, canvas_agent) are dispatched via their own
|
||
# factory and must NOT receive an agent_id override.
|
||
if assistant_id and assistant_id not in _BUILTIN_GRAPH_IDS:
|
||
normalized = assistant_id.strip()
|
||
if not normalized or not re.fullmatch(r"[A-Za-z0-9_-]+", normalized):
|
||
raise ValueError(f"Invalid assistant_id {assistant_id!r}: must contain only letters, digits, hyphens, and underscores.")
|
||
if "configurable" in config:
|
||
target = config["configurable"]
|
||
elif "context" in config:
|
||
target = config["context"]
|
||
else:
|
||
target = config.setdefault("configurable", {})
|
||
if target is not None and "agent_id" not in target:
|
||
target["agent_id"] = normalized
|
||
if metadata:
|
||
config.setdefault("metadata", {}).update(metadata)
|
||
return config
|
||
|
||
|
||
def _collect_requested_agent_ids(body: Any) -> set[str]:
|
||
"""Collect agent ids from assistant_id/config/context for access checks."""
|
||
agent_ids: set[str] = set()
|
||
if body.assistant_id and body.assistant_id not in _BUILTIN_GRAPH_IDS:
|
||
agent_ids.add(str(body.assistant_id))
|
||
|
||
def collect(container: Mapping[str, Any] | None) -> None:
|
||
if not container:
|
||
return
|
||
for key in ("agent_id", "agent_name"):
|
||
value = container.get(key)
|
||
if value:
|
||
agent_ids.add(str(value))
|
||
|
||
context = getattr(body, "context", None)
|
||
if isinstance(context, Mapping):
|
||
collect(context)
|
||
|
||
config = body.config or {}
|
||
if isinstance(config, Mapping):
|
||
nested_context = config.get("context")
|
||
nested_configurable = config.get("configurable")
|
||
if isinstance(nested_context, Mapping):
|
||
collect(nested_context)
|
||
if isinstance(nested_configurable, Mapping):
|
||
collect(nested_configurable)
|
||
|
||
return agent_ids
|
||
|
||
|
||
async def _sync_legacy_agent_records(store) -> None:
|
||
"""Import existing folder-backed agents as built-in records."""
|
||
from deerflow.config.agents_config import list_custom_agents
|
||
|
||
for agent_cfg in list_custom_agents():
|
||
agent_id = agent_cfg.id or agent_cfg.name
|
||
if not agent_id:
|
||
continue
|
||
await store.ensure_builtin(
|
||
{
|
||
"id": agent_id,
|
||
"name": agent_cfg.name,
|
||
"description": agent_cfg.description or "",
|
||
"published": True,
|
||
}
|
||
)
|
||
|
||
|
||
async def _enforce_agent_access(body: Any, request: Request) -> None:
|
||
agent_ids = _collect_requested_agent_ids(body)
|
||
if not agent_ids:
|
||
return
|
||
|
||
from app.gateway.deps import get_agent_store
|
||
from deerflow.config.agents_config import validate_agent_id
|
||
from deerflow.runtime.user_context import get_effective_user_id
|
||
|
||
store = get_agent_store(request)
|
||
await _sync_legacy_agent_records(store)
|
||
user_id = get_effective_user_id()
|
||
for agent_id in agent_ids:
|
||
try:
|
||
normalized = validate_agent_id(agent_id)
|
||
except ValueError as exc:
|
||
raise HTTPException(status_code=422, detail=str(exc)) from exc
|
||
if normalized is None:
|
||
continue
|
||
if await store.get_visible(normalized, user_id) is None:
|
||
raise HTTPException(status_code=404, detail=f"Agent {normalized} not found")
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Run lifecycle
|
||
# ---------------------------------------------------------------------------
|
||
|
||
|
||
async def start_run(
|
||
body: Any,
|
||
thread_id: str,
|
||
request: Request,
|
||
) -> RunRecord:
|
||
"""Create a RunRecord and launch the background agent task.
|
||
|
||
Parameters
|
||
----------
|
||
body : RunCreateRequest
|
||
The validated request body (typed as Any to avoid circular import
|
||
with the router module that defines the Pydantic model).
|
||
thread_id : str
|
||
Target thread.
|
||
request : Request
|
||
FastAPI request — used to retrieve singletons from ``app.state``.
|
||
"""
|
||
bridge = get_stream_bridge(request)
|
||
run_mgr = get_run_manager(request)
|
||
run_ctx = get_run_context(request)
|
||
|
||
disconnect = DisconnectMode.cancel if body.on_disconnect == "cancel" else DisconnectMode.continue_
|
||
|
||
await _enforce_agent_access(body, request)
|
||
|
||
try:
|
||
record = await run_mgr.create_or_reject(
|
||
thread_id,
|
||
body.assistant_id,
|
||
on_disconnect=disconnect,
|
||
metadata=body.metadata or {},
|
||
kwargs={"input": body.input, "config": body.config},
|
||
multitask_strategy=body.multitask_strategy,
|
||
)
|
||
except ConflictError as exc:
|
||
raise HTTPException(status_code=409, detail=str(exc)) from exc
|
||
except UnsupportedStrategyError as exc:
|
||
raise HTTPException(status_code=501, detail=str(exc)) from exc
|
||
|
||
# Upsert thread metadata so the thread appears in /threads/search,
|
||
# even for threads that were never explicitly created via POST /threads
|
||
# (e.g. stateless runs).
|
||
requested_rag_mode = _rag_mode_requested(body)
|
||
effective_rag_mode = requested_rag_mode
|
||
try:
|
||
existing = await run_ctx.thread_store.get(thread_id)
|
||
effective_rag_mode = requested_rag_mode or _thread_rag_mode_enabled(existing)
|
||
if existing is None:
|
||
metadata = dict(body.metadata or {})
|
||
if effective_rag_mode:
|
||
metadata["rag_mode_enabled"] = True
|
||
metadata["rag_mode"] = True
|
||
await run_ctx.thread_store.create(
|
||
thread_id,
|
||
assistant_id=body.assistant_id,
|
||
metadata=metadata,
|
||
)
|
||
else:
|
||
await run_ctx.thread_store.update_status(thread_id, "running")
|
||
if requested_rag_mode and not _thread_rag_mode_enabled(existing):
|
||
await run_ctx.thread_store.update_metadata(
|
||
thread_id,
|
||
{"rag_mode_enabled": True, "rag_mode": True},
|
||
)
|
||
effective_rag_mode = True
|
||
# SDK 的 ``client.threads.create()`` 不传 assistant_id —— 先建出来空的 thread，
|
||
# 真正提交 run 时才知道用哪个 assistant。这里回填一次，避免历史列表 / 按
|
||
# assistant 筛选时拿不到关联。
|
||
if not existing.get("assistant_id") and body.assistant_id:
|
||
await run_ctx.thread_store.update_assistant_id(thread_id, body.assistant_id)
|
||
except Exception:
|
||
logger.warning("Failed to upsert thread_meta for %s (non-fatal)", sanitize_log_param(thread_id))
|
||
|
||
agent_factory = resolve_agent_factory(body.assistant_id)
|
||
graph_input = normalize_input(body.input)
|
||
|
||
# Admin-configurable prompt prefix: when the caller opted in via
|
||
# context.prompt_prefix_enabled, tag the human message with the configured
|
||
# prefix (stored in additional_kwargs, not concatenated into content, so the
|
||
# user never sees it — PromptPrefixMiddleware injects it for the LLM only).
|
||
# The frontend only sends the flag on the opening turn, so this naturally
|
||
# applies once. We intentionally do NOT gate on thread-row existence: the
|
||
# thread_meta row is frequently pre-created, which previously skipped the
|
||
# prefix for genuine first turns.
|
||
if _prompt_prefix_opted_in(body):
|
||
prefix_settings = load_system_settings().prompt_prefix
|
||
if prefix_settings.enabled and prefix_settings.prompt_prefix.strip():
|
||
graph_input = _prepend_prompt_prefix(graph_input, prefix_settings.prompt_prefix.strip())
|
||
if effective_rag_mode:
|
||
graph_input = _append_invisible_prompt(graph_input, _RAG_CITATION_PROMPT)
|
||
config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
|
||
|
||
# Merge DeerFlow-specific context overrides into both ``configurable`` and ``context``.
|
||
# The ``context`` field is a custom extension for the langgraph-compat layer
|
||
# that carries agent configuration (model_name, thinking_enabled, etc.).
|
||
# Only agent-relevant keys are forwarded; unknown keys (e.g. thread_id) are ignored.
|
||
body_context = dict(getattr(body, "context", None) or {})
|
||
if effective_rag_mode:
|
||
body_context["rag_mode_enabled"] = True
|
||
# Promote whitelisted top-level body fields into the context map so
|
||
# callers can pass them either nested under ``context`` or as flat
|
||
# top-level fields (``excluded_tools``, ``skill_stop_names``, …).
|
||
for top_level_key in ("excluded_tools", "skill_stop_names", "skill_stop_message"):
|
||
value = getattr(body, top_level_key, None)
|
||
if value is not None and top_level_key not in body_context:
|
||
body_context[top_level_key] = value
|
||
merge_run_context_overrides(config, body_context or None)
|
||
|
||
stream_modes = normalize_stream_modes(body.stream_mode)
|
||
|
||
task = asyncio.create_task(
|
||
run_agent(
|
||
bridge,
|
||
run_mgr,
|
||
record,
|
||
ctx=run_ctx,
|
||
agent_factory=agent_factory,
|
||
graph_input=graph_input,
|
||
config=config,
|
||
stream_modes=stream_modes,
|
||
stream_subgraphs=body.stream_subgraphs,
|
||
interrupt_before=body.interrupt_before,
|
||
interrupt_after=body.interrupt_after,
|
||
# 透传 SDK ``Command``：resume / update / goto。Pregel 会据此从最近
|
||
# checkpoint 处恢复，而不是把 graph_input 当新一轮输入跑。
|
||
command=getattr(body, "command", None),
|
||
)
|
||
)
|
||
record.task = task
|
||
|
||
# Title sync is handled by worker.py's finally block which reads the
|
||
# title from the checkpoint and calls thread_store.update_display_name
|
||
# after the run completes.
|
||
|
||
return record
|
||
|
||
|
||
async def sse_consumer(
|
||
bridge: StreamBridge,
|
||
record: RunRecord,
|
||
request: Request,
|
||
run_mgr: RunManager,
|
||
):
|
||
"""Async generator that yields SSE frames from the bridge.
|
||
|
||
The ``finally`` block implements ``on_disconnect`` semantics:
|
||
- ``cancel``: abort the background task on client disconnect.
|
||
- ``continue``: let the task run; events are discarded.
|
||
"""
|
||
last_event_id = request.headers.get("Last-Event-ID")
|
||
try:
|
||
async for entry in bridge.subscribe(record.run_id, last_event_id=last_event_id):
|
||
if await request.is_disconnected():
|
||
break
|
||
|
||
if entry is HEARTBEAT_SENTINEL:
|
||
yield ": heartbeat\n\n"
|
||
continue
|
||
|
||
if entry is END_SENTINEL:
|
||
yield format_sse("end", None, event_id=entry.id or None)
|
||
return
|
||
|
||
yield format_sse(entry.event, entry.data, event_id=entry.id or None)
|
||
|
||
finally:
|
||
if record.status in (RunStatus.pending, RunStatus.running):
|
||
if record.on_disconnect == DisconnectMode.cancel:
|
||
await run_mgr.cancel(record.run_id)
|