mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-29 01:15:59 +00:00
1346 lines
58 KiB
Python
1346 lines
58 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 AsyncIterator, Mapping
|
|
from contextlib import asynccontextmanager
|
|
from types import SimpleNamespace
|
|
from typing import Any
|
|
|
|
from fastapi import HTTPException, Request
|
|
from langchain_core.messages import BaseMessage
|
|
from langchain_core.messages.utils import convert_to_messages
|
|
from langgraph.types import Command
|
|
|
|
from app.gateway.auth_disabled import AUTH_SOURCE_INTERNAL
|
|
from app.gateway.deps import get_checkpointer, get_local_provider, get_run_context, get_run_manager, get_stream_bridge
|
|
from app.gateway.internal_auth import (
|
|
INTERNAL_OWNER_USER_ID_HEADER_NAME,
|
|
INTERNAL_SYSTEM_ROLE,
|
|
get_internal_user,
|
|
get_trusted_internal_owner_user_id,
|
|
)
|
|
from app.gateway.run_models import RunCreateRequest
|
|
from app.gateway.utils import sanitize_log_param
|
|
from deerflow.agents.middlewares.dynamic_context_middleware import _DYNAMIC_CONTEXT_REMINDER_KEY, _REMINDER_DATE_KEY
|
|
from deerflow.agents.middlewares.view_image_middleware import _IMAGE_CONTEXT_MESSAGE_MARKER_KEY
|
|
from deerflow.config.app_config import get_app_config
|
|
from deerflow.runtime import (
|
|
END_SENTINEL,
|
|
HEARTBEAT_SENTINEL,
|
|
ORPHAN_RECOVERY_STOP_REASON,
|
|
CheckpointStateAccessor,
|
|
ConflictError,
|
|
DisconnectMode,
|
|
RunContext,
|
|
RunManager,
|
|
RunRecord,
|
|
RunStatus,
|
|
StreamBridge,
|
|
StreamGap,
|
|
ThreadOperationKind,
|
|
UnsupportedStrategyError,
|
|
build_state_mutation_graph,
|
|
run_agent,
|
|
)
|
|
from deerflow.runtime.checkpoint_mode import (
|
|
INTERNAL_CHECKPOINT_MODE_KEY,
|
|
CheckpointModeMismatchError,
|
|
checkpoint_tuple_uses_delta,
|
|
inject_checkpoint_mode,
|
|
)
|
|
from deerflow.runtime.checkpoint_state import graph_state_schema
|
|
from deerflow.runtime.goal import goal_thread_lock
|
|
from deerflow.runtime.runs.naming import resolve_root_run_name
|
|
from deerflow.runtime.secret_context import redact_config_secrets
|
|
from deerflow.runtime.stream_modes import normalize_stream_modes
|
|
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
|
from deerflow.utils.messages import ORIGINAL_USER_CONTENT_KEY
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@asynccontextmanager
|
|
async def reserve_checkpoint_write(
|
|
request: Request,
|
|
thread_id: str,
|
|
*,
|
|
user_id: str | None = None,
|
|
) -> AsyncIterator[None]:
|
|
"""Serialize an out-of-run checkpoint writer against all thread operations."""
|
|
run_manager = get_run_manager(request)
|
|
async with goal_thread_lock(thread_id):
|
|
async with run_manager.reserve_thread_operation(
|
|
thread_id,
|
|
kind=ThreadOperationKind.checkpoint_write,
|
|
user_id=user_id,
|
|
):
|
|
yield
|
|
|
|
|
|
_TERMINAL_RUN_STATUSES = {
|
|
RunStatus.success,
|
|
RunStatus.error,
|
|
RunStatus.timeout,
|
|
RunStatus.interrupted,
|
|
}
|
|
|
|
_THREAD_METADATA_SETUP_TIMEOUT_SECONDS = 5.0
|
|
|
|
_SERVER_OWNED_MESSAGE_METADATA_KEYS = frozenset(
|
|
{
|
|
_DYNAMIC_CONTEXT_REMINDER_KEY,
|
|
_REMINDER_DATE_KEY,
|
|
_IMAGE_CONTEXT_MESSAGE_MARKER_KEY,
|
|
}
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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)
|
|
|
|
|
|
def _run_is_terminal(record: RunRecord) -> bool:
|
|
return record.status in _TERMINAL_RUN_STATUSES
|
|
|
|
|
|
def _consume_task_result(task: asyncio.Task) -> None:
|
|
"""Retrieve a detached task's exception without propagating cancellation."""
|
|
if not task.cancelled():
|
|
task.exception()
|
|
|
|
|
|
def _log_thread_metadata_task_result(task: asyncio.Task, *, thread_id: str) -> None:
|
|
"""Log detached metadata setup failures while ignoring cancellation."""
|
|
if task.cancelled():
|
|
return
|
|
try:
|
|
task.result()
|
|
except asyncio.CancelledError:
|
|
return
|
|
except Exception:
|
|
logger.warning(
|
|
"Failed to ensure thread_meta for %s after worker detached (non-fatal)",
|
|
sanitize_log_param(thread_id),
|
|
exc_info=True,
|
|
)
|
|
|
|
|
|
async def _ensure_thread_metadata(
|
|
run_ctx: RunContext,
|
|
record: RunRecord,
|
|
*,
|
|
owner_user_id: str | None,
|
|
) -> None:
|
|
"""Ensure an admitted run's thread exists without delaying task attachment."""
|
|
thread_store = run_ctx.thread_store
|
|
existing = await thread_store.get(record.thread_id)
|
|
if existing is None and owner_user_id:
|
|
unscoped = await thread_store.get(record.thread_id, user_id=None)
|
|
if unscoped is not None:
|
|
if unscoped.get("user_id") != owner_user_id:
|
|
await thread_store.update_owner(record.thread_id, owner_user_id, user_id=None)
|
|
existing = await thread_store.get(record.thread_id)
|
|
if existing is None:
|
|
await thread_store.create(
|
|
record.thread_id,
|
|
assistant_id=record.assistant_id,
|
|
metadata=record.metadata,
|
|
)
|
|
|
|
|
|
async def _terminal_record_stream_missing(bridge: StreamBridge, record: RunRecord) -> bool:
|
|
"""True when a terminal run has no retained stream on bridges that can tell."""
|
|
if not _run_is_terminal(record):
|
|
return False
|
|
stream_exists = getattr(bridge, "stream_exists", None)
|
|
if stream_exists is None:
|
|
return False
|
|
try:
|
|
return not bool(await stream_exists(record.run_id))
|
|
except Exception:
|
|
logger.debug(
|
|
"Failed to probe stream existence for terminal run %s",
|
|
sanitize_log_param(record.run_id),
|
|
exc_info=True,
|
|
)
|
|
return False
|
|
|
|
|
|
async def _orphan_recovery_observed_after_heartbeat(
|
|
record: RunRecord,
|
|
run_mgr: RunManager,
|
|
) -> bool:
|
|
"""Return whether durable orphan recovery is the consumer's liveness edge.
|
|
|
|
A normal terminal status is not sufficient: the producer persists status
|
|
before publishing its final error/data frames and END. Orphan recovery is
|
|
different because the producer is known to be gone and the durable
|
|
``stop_reason`` is written atomically with the terminal status. Only that
|
|
explicit signal may synthesize END after a heartbeat.
|
|
"""
|
|
if not record.store_only:
|
|
return False
|
|
refreshed = await run_mgr.get(record.run_id, user_id=record.user_id)
|
|
return refreshed is not None and _run_is_terminal(refreshed) and refreshed.stop_reason == ORPHAN_RECOVERY_STOP_REASON
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Input / config helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _strip_external_message_metadata(message: Any) -> Any:
|
|
"""Remove server-owned metadata from an untrusted input message."""
|
|
if not isinstance(message, BaseMessage):
|
|
return message
|
|
additional_kwargs = dict(message.additional_kwargs)
|
|
additional_kwargs.pop(ORIGINAL_USER_CONTENT_KEY, None)
|
|
for key in _SERVER_OWNED_MESSAGE_METADATA_KEYS:
|
|
additional_kwargs.pop(key, None)
|
|
if additional_kwargs == message.additional_kwargs:
|
|
return message
|
|
return message.model_copy(update={"additional_kwargs": additional_kwargs})
|
|
|
|
|
|
def normalize_input(raw_input: dict[str, Any] | None, *, trusted_internal: bool = False) -> dict[str, Any]:
|
|
"""Convert LangGraph Platform input format to LangChain state dict.
|
|
|
|
Delegates dict→message coercion to ``langchain_core.messages.utils.convert_to_messages``
|
|
so that ``additional_kwargs`` (e.g. uploaded-file metadata — gh #3132), ``id``,
|
|
``name``, and non-human roles (ai/system/tool) survive unchanged. An earlier
|
|
hand-rolled version only forwarded ``content`` and collapsed every role to
|
|
``HumanMessage``, which silently stripped frontend-supplied attachments.
|
|
|
|
Malformed message dicts (missing ``role``/``type``/``content``, unsupported
|
|
role, etc.) raise ``HTTPException(400)`` with the offending index, instead
|
|
of bubbling up as a 500. The gateway is a system boundary, so per-entry
|
|
validation errors are the right shape for clients to retry against.
|
|
|
|
``original_user_content``, dynamic-context reminder markers, and the
|
|
transient view-image context marker are server-owned. External callers
|
|
cannot supply them; trusted internal channel calls may preserve metadata
|
|
they added before invoking this boundary.
|
|
"""
|
|
if raw_input is None:
|
|
return {}
|
|
messages = raw_input.get("messages")
|
|
if messages and isinstance(messages, list):
|
|
converted: list[Any] = []
|
|
for index, msg in enumerate(messages):
|
|
if isinstance(msg, BaseMessage):
|
|
converted.append(msg)
|
|
elif isinstance(msg, dict):
|
|
try:
|
|
converted.extend(convert_to_messages([msg]))
|
|
except (ValueError, TypeError, NotImplementedError) as exc:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Invalid message at input.messages[{index}]: {exc}",
|
|
) from exc
|
|
else:
|
|
converted.append(msg)
|
|
if not trusted_internal:
|
|
converted = [_strip_external_message_metadata(message) for message in converted]
|
|
return {**raw_input, "messages": converted}
|
|
return raw_input
|
|
|
|
|
|
_DEFAULT_ASSISTANT_ID = "lead_agent"
|
|
|
|
|
|
# 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(
|
|
{
|
|
"model_name",
|
|
"mode",
|
|
"thinking_enabled",
|
|
"reasoning_effort",
|
|
"is_plan_mode",
|
|
"subagent_enabled",
|
|
"max_concurrent_subagents",
|
|
"max_total_subagents",
|
|
"agent_name",
|
|
"is_bootstrap",
|
|
}
|
|
)
|
|
|
|
# Keys honored only for internally-authenticated callers (the scheduler path).
|
|
# ``non_interactive`` strips ``ask_clarification`` from the lead-agent toolset;
|
|
# arbitrary HTTP/IM clients must not be able to force autonomous execution.
|
|
_CONTEXT_INTERNAL_CALLER_KEYS: frozenset[str] = frozenset({"non_interactive"})
|
|
|
|
# Server-owned authorization identity fields. These must never be accepted from
|
|
# client-supplied ``body.config.context`` or ``body.config.configurable``. They
|
|
# are either produced by Gateway auth state or admitted from a separately
|
|
# authenticated internal request channel.
|
|
# ``is_internal`` — derived from ``request.state.auth_source``
|
|
# ``authz_attributes`` — Phase 1A has no Gateway-side producer; always cleared.
|
|
# ``channel_user_id`` — accepted only from trusted internal ``body.context``.
|
|
_SERVER_OWNED_AUTHZ_CONTEXT_KEYS: frozenset[str] = frozenset({"is_internal", "authz_attributes", "channel_user_id"})
|
|
|
|
# Keys forwarded from ``body.context`` into ``config['context']`` ONLY (the
|
|
# runtime context that becomes ``ToolRuntime.context`` / ``runtime.context``),
|
|
# never into ``config['configurable']``. These are read by tools and
|
|
# middlewares from ``runtime.context`` and have no reason to live in
|
|
# ``configurable`` — and ``configurable`` is persisted in checkpoints, so
|
|
# keeping secrets like ``github_token`` out of it avoids writing a
|
|
# short-lived installation token into the checkpoint store.
|
|
#
|
|
# ``github_token`` — App installation token minted by the GitHub
|
|
# channel; the bash tool exposes it as
|
|
# ``GH_TOKEN``/``GITHUB_TOKEN`` so ``gh`` and
|
|
# ``git`` push as the bot, not the host user.
|
|
# ``disable_clarification`` — set for non-interactive channels (GitHub
|
|
# webhooks) so ClarificationMiddleware proceeds
|
|
# instead of dead-ending the run.
|
|
_CONTEXT_RUNTIME_ONLY_KEYS: frozenset[str] = frozenset({"github_token", "disable_clarification"})
|
|
|
|
|
|
def strip_internal_context_keys(config: dict[str, Any]) -> None:
|
|
"""Drop internal-only keys a non-internal caller smuggled into the run config.
|
|
|
|
Gating :func:`merge_run_context_overrides` is not enough on its own:
|
|
``build_run_config`` copies a client-supplied ``body.config['context']`` /
|
|
``body.config['configurable']`` verbatim, so the same keys must be scrubbed
|
|
from both sections after the config is assembled.
|
|
"""
|
|
for section in ("context", "configurable"):
|
|
value = config.get(section)
|
|
if isinstance(value, dict):
|
|
for key in _CONTEXT_INTERNAL_CALLER_KEYS:
|
|
value.pop(key, None)
|
|
|
|
|
|
def merge_run_context_overrides(config: dict[str, Any], context: Mapping[str, Any] | None, *, internal: bool = False) -> 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).
|
|
|
|
``user_id`` is intentionally propagated into ``config['context']`` in addition to
|
|
the whitelisted keys, so non-web callers (e.g. IM channels) that supply identity in
|
|
``body.context`` keep it on ``ToolRuntime.context``. It is merged with
|
|
``setdefault`` so a server-authenticated id stamped by
|
|
:func:`inject_authenticated_user_context` always wins over the client-supplied one.
|
|
|
|
:data:`_CONTEXT_INTERNAL_CALLER_KEYS` are also forwarded when ``internal``
|
|
is True; for non-internal callers those keys are dropped from client requests
|
|
by :func:`strip_internal_context_keys`.
|
|
|
|
A second set of keys (``_CONTEXT_RUNTIME_ONLY_KEYS`` — e.g. ``github_token``,
|
|
``disable_clarification``) is forwarded into ``config['context']`` only, never
|
|
``configurable``. These are secrets / runtime flags read by tools and middlewares
|
|
from ``runtime.context``; keeping them out of ``configurable`` avoids persisting a
|
|
short-lived token in the checkpoint store.
|
|
"""
|
|
if not context:
|
|
return
|
|
configurable = config.setdefault("configurable", {})
|
|
runtime_context = config.setdefault("context", {})
|
|
keys = _CONTEXT_CONFIGURABLE_KEYS | _CONTEXT_INTERNAL_CALLER_KEYS if internal else _CONTEXT_CONFIGURABLE_KEYS
|
|
for key in 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])
|
|
# Context-only keys (secrets / runtime flags) land in ``config['context']``
|
|
# only — never ``configurable`` (which is persisted in checkpoints).
|
|
for key in _CONTEXT_RUNTIME_ONLY_KEYS:
|
|
if key in context and isinstance(runtime_context, dict):
|
|
runtime_context.setdefault(key, context[key])
|
|
if "user_id" in context and isinstance(runtime_context, dict):
|
|
runtime_context.setdefault("user_id", context["user_id"])
|
|
|
|
|
|
async def resolve_trusted_internal_owner_for_attribution(request: Request, owner_user_id: str | None) -> Any | None:
|
|
"""Resolve the DeerFlow user used only for trusted internal attribution."""
|
|
|
|
if not owner_user_id:
|
|
return None
|
|
user = getattr(request.state, "user", None)
|
|
if getattr(user, "system_role", None) != INTERNAL_SYSTEM_ROLE:
|
|
return None
|
|
try:
|
|
return await get_local_provider().get_user(owner_user_id)
|
|
except Exception:
|
|
logger.exception("Failed to resolve trusted internal owner %s", sanitize_log_param(owner_user_id))
|
|
return None
|
|
|
|
|
|
def inject_authenticated_user_context(
|
|
config: dict[str, Any],
|
|
request: Request,
|
|
*,
|
|
internal_owner_user: Any | None = None,
|
|
request_context: Mapping[str, Any] | None = None,
|
|
) -> None:
|
|
"""Stamp the authenticated user into the run context for background tools.
|
|
|
|
Tool execution may happen after the request handler has returned, so tools
|
|
that persist user-scoped files should not rely only on ambient ContextVars.
|
|
The value comes from server-side auth state, never from client context.
|
|
|
|
``request_context.channel_user_id`` is the sole exception: it is honored
|
|
only after ``request.state.auth_source`` proves the caller is internal.
|
|
Values copied through the free-form RunnableConfig are always cleared.
|
|
"""
|
|
|
|
# --- Server-owned authorization identity fields ---
|
|
# Clear any client-forged values from both config sections, then write the
|
|
# authoritative is_internal. This runs before ALL early returns so that
|
|
# even user_id-is-None paths get a defined is_internal value.
|
|
runtime_context = config.setdefault("context", {})
|
|
if not isinstance(runtime_context, dict):
|
|
raise TypeError("run context must be a mapping")
|
|
for key in _SERVER_OWNED_AUTHZ_CONTEXT_KEYS:
|
|
runtime_context.pop(key, None)
|
|
configurable = config.get("configurable")
|
|
if isinstance(configurable, dict):
|
|
for key in _SERVER_OWNED_AUTHZ_CONTEXT_KEYS:
|
|
configurable.pop(key, None)
|
|
auth_source = getattr(getattr(request, "state", None), "auth_source", None)
|
|
runtime_context["is_internal"] = auth_source == AUTH_SOURCE_INTERNAL
|
|
if auth_source == AUTH_SOURCE_INTERNAL and request_context is not None:
|
|
channel_user_id = request_context.get("channel_user_id")
|
|
if channel_user_id is not None:
|
|
runtime_context["channel_user_id"] = channel_user_id
|
|
|
|
user = getattr(request.state, "user", None)
|
|
user_id = getattr(user, "id", None)
|
|
if user_id is None:
|
|
return
|
|
|
|
if getattr(user, "system_role", None) == INTERNAL_SYSTEM_ROLE:
|
|
runtime_context = config.setdefault("context", {})
|
|
if not isinstance(runtime_context, dict):
|
|
return
|
|
if internal_owner_user is None:
|
|
runtime_context.pop("user_role", None)
|
|
runtime_context.pop("oauth_provider", None)
|
|
runtime_context.pop("oauth_id", None)
|
|
return
|
|
owner_user_id = getattr(internal_owner_user, "id", None)
|
|
if owner_user_id is not None:
|
|
runtime_context["user_id"] = str(owner_user_id)
|
|
runtime_context["user_role"] = getattr(internal_owner_user, "system_role", None)
|
|
runtime_context["oauth_provider"] = getattr(internal_owner_user, "oauth_provider", None)
|
|
runtime_context["oauth_id"] = getattr(internal_owner_user, "oauth_id", None)
|
|
return
|
|
|
|
runtime_context = config.setdefault("context", {})
|
|
if isinstance(runtime_context, dict):
|
|
runtime_context["user_id"] = str(user_id)
|
|
runtime_context["user_role"] = getattr(user, "system_role", None)
|
|
runtime_context["oauth_provider"] = getattr(user, "oauth_provider", None)
|
|
runtime_context["oauth_id"] = getattr(user, "oauth_id", None)
|
|
|
|
|
|
def resolve_agent_factory(assistant_id: str | None):
|
|
"""Resolve the agent factory callable from config.
|
|
|
|
Custom agents are implemented as ``lead_agent`` + an ``agent_name``
|
|
injected into ``configurable`` or ``context`` — see
|
|
:func:`build_run_config`. All ``assistant_id`` values therefore map to the
|
|
same factory; the routing happens inside ``make_lead_agent`` when it reads
|
|
``cfg["agent_name"]``.
|
|
"""
|
|
from deerflow.agents.lead_agent.agent import make_lead_agent
|
|
|
|
return make_lead_agent
|
|
|
|
|
|
# Lead-agent recursion budget bounds. The Gateway must NOT trust a
|
|
# client-supplied ``recursion_limit`` verbatim: an arbitrarily large value lets
|
|
# a single run execute unbounded LangGraph super-steps (each at least one LLM
|
|
# call), enabling runaway API cost / DoS. ``_DEFAULT_RECURSION_LIMIT`` is the
|
|
# server default when the client sends nothing; the hard ceiling any client
|
|
# value is clamped to is configurable via ``AppConfig.max_recursion_limit``.
|
|
_DEFAULT_RECURSION_LIMIT = 100
|
|
_DEFAULT_MAX_RECURSION_LIMIT = 1000
|
|
|
|
|
|
def _resolve_max_recursion_limit() -> int:
|
|
"""Resolve the clamp ceiling from ``AppConfig.max_recursion_limit``.
|
|
|
|
Falls back to ``_DEFAULT_MAX_RECURSION_LIMIT`` when the app config cannot be
|
|
loaded (e.g. no ``config.yaml`` in a bare unit-test environment) so that the
|
|
clamp still applies rather than crashing the run-config assembly.
|
|
"""
|
|
try:
|
|
return get_app_config().max_recursion_limit
|
|
except Exception:
|
|
return _DEFAULT_MAX_RECURSION_LIMIT
|
|
|
|
|
|
def _clamp_recursion_limit(value: Any, max_limit: int) -> int:
|
|
"""Clamp a client-supplied ``recursion_limit`` into a safe server range.
|
|
|
|
Non-integer values (including ``bool``, an ``int`` subclass) and non-positive
|
|
values fall back to ``_DEFAULT_RECURSION_LIMIT``; valid positive integers are
|
|
capped at ``max_limit`` (from ``AppConfig.max_recursion_limit``).
|
|
"""
|
|
if isinstance(value, bool) or not isinstance(value, int) or value <= 0:
|
|
return _DEFAULT_RECURSION_LIMIT
|
|
return min(value, max_limit)
|
|
|
|
|
|
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 name is forwarded as ``agent_name`` in
|
|
both ``configurable`` and ``context`` so it is visible to legacy
|
|
configurable readers and to LangGraph ``ToolRuntime.context`` consumers
|
|
(e.g. the ``setup_agent`` tool, which since LangGraph >=1.1.9 no longer
|
|
falls back from ``context`` to ``configurable``). An explicit
|
|
``agent_name`` in either container takes precedence over the value
|
|
derived from ``assistant_id``. ``make_lead_agent`` reads this key to
|
|
load the matching ``agents/<name>/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.
|
|
"""
|
|
# Lead-agent recursion budget (LangGraph super-steps for the lead graph
|
|
# only). Independent of subagent depth: a `task()` dispatch runs the whole
|
|
# subagent inside ONE lead tools-node step, and subagents enforce their own
|
|
# limit via `subagents.max_turns`. Do not conflate this 100 with the
|
|
# general-purpose subagent's max_turns.
|
|
config: dict[str, Any] = {"recursion_limit": _DEFAULT_RECURSION_LIMIT}
|
|
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") or {}).keys()),
|
|
)
|
|
context_value = request_config["context"]
|
|
if context_value is None:
|
|
context = {}
|
|
elif isinstance(context_value, Mapping):
|
|
# Strip caller-supplied ``__``-prefixed keys: those are the
|
|
# harness's private run-context channels (skill secret-binding
|
|
# sources, the active-secret set, the run journal). A caller must
|
|
# not be able to seed them and forge internal state — e.g. a
|
|
# forged ``__slash_skill_secret_source`` would otherwise bypass the
|
|
# skill enabled/allowlist/declaration gates (#3938). Legitimate
|
|
# caller keys (``secrets``, ``user_id``, model overrides) never use
|
|
# the ``__`` prefix.
|
|
context = {key: value for key, value in context_value.items() if not (isinstance(key, str) and key.startswith("__"))}
|
|
else:
|
|
raise ValueError("request config 'context' must be a mapping or null.")
|
|
context["thread_id"] = thread_id
|
|
config["context"] = context
|
|
# The checkpointer always scopes state by configurable["thread_id"],
|
|
# regardless of whether the caller drives the run via context (e.g.
|
|
# request-scoped secrets, #3861). thread_id comes from the URL path,
|
|
# not caller config, so mirror it here while keeping secret-bearing
|
|
# context keys out of configurable.
|
|
config["configurable"] = {"thread_id": thread_id}
|
|
else:
|
|
configurable = {"thread_id": thread_id}
|
|
configurable.update(request_config.get("configurable") or {})
|
|
configurable["thread_id"] = thread_id
|
|
config["configurable"] = configurable
|
|
for k, v in request_config.items():
|
|
if k not in ("configurable", "context"):
|
|
config[k] = v
|
|
# Never trust a client-supplied recursion_limit verbatim: clamp it to a
|
|
# safe server range so a single run cannot execute unbounded LangGraph
|
|
# super-steps (runaway LLM cost / DoS). Applied after the passthrough so
|
|
# it overrides whatever the client sent.
|
|
if "recursion_limit" in request_config:
|
|
max_limit = _resolve_max_recursion_limit()
|
|
clamped = _clamp_recursion_limit(request_config["recursion_limit"], max_limit)
|
|
if clamped != request_config["recursion_limit"]:
|
|
logger.warning(
|
|
"build_run_config: clamped client recursion_limit %r -> %d (max %d). thread_id=%s",
|
|
request_config["recursion_limit"],
|
|
clamped,
|
|
max_limit,
|
|
thread_id,
|
|
)
|
|
config["recursion_limit"] = clamped
|
|
else:
|
|
config["configurable"] = {"thread_id": thread_id}
|
|
|
|
# Inject custom agent name when the caller specified a non-default assistant.
|
|
# Honour an explicit agent_name in either runtime options container.
|
|
if assistant_id and assistant_id != _DEFAULT_ASSISTANT_ID:
|
|
normalized = assistant_id.strip().lower().replace("_", "-")
|
|
if not normalized or not re.fullmatch(r"[a-z0-9-]+", normalized):
|
|
raise ValueError(f"Invalid assistant_id {assistant_id!r}: must contain only letters, digits, and hyphens after normalization.")
|
|
configurable = config.setdefault("configurable", {})
|
|
runtime_context = config.setdefault("context", {})
|
|
explicit_agent_name: str | None = None
|
|
if isinstance(configurable, dict) and isinstance(configurable.get("agent_name"), str):
|
|
explicit_agent_name = configurable["agent_name"]
|
|
elif isinstance(runtime_context, dict) and isinstance(runtime_context.get("agent_name"), str):
|
|
explicit_agent_name = runtime_context["agent_name"]
|
|
effective_agent_name = explicit_agent_name or normalized
|
|
if isinstance(configurable, dict):
|
|
configurable["agent_name"] = effective_agent_name
|
|
if isinstance(runtime_context, dict):
|
|
runtime_context["agent_name"] = effective_agent_name
|
|
config.setdefault("run_name", resolve_root_run_name(config, normalized))
|
|
for section in ("configurable", "context"):
|
|
external_values = config.get(section)
|
|
if isinstance(external_values, dict):
|
|
external_values.pop(INTERNAL_CHECKPOINT_MODE_KEY, None)
|
|
|
|
if metadata:
|
|
config.setdefault("metadata", {}).update(metadata)
|
|
return config
|
|
|
|
|
|
def build_checkpoint_state_mutation_accessor(
|
|
request: Request,
|
|
*,
|
|
thread_id: str,
|
|
as_node: str,
|
|
checkpoint_id: str | None = None,
|
|
state_schema: Any | None = None,
|
|
) -> tuple[CheckpointStateAccessor, dict[str, Any]]:
|
|
"""Build a state-only graph whose writer node finishes immediately.
|
|
|
|
``state_schema`` should be the thread's effective schema (from
|
|
:func:`graph_state_schema` on the assistant graph) whenever the write
|
|
carries materialized state; with the base-schema fallback, channels
|
|
contributed by custom middleware are silently discarded.
|
|
"""
|
|
mode = getattr(request.app.state, "checkpoint_channel_mode", "full")
|
|
config: dict[str, Any] = {
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": "",
|
|
}
|
|
}
|
|
if checkpoint_id is not None:
|
|
config["configurable"]["checkpoint_id"] = checkpoint_id
|
|
inject_checkpoint_mode(config, mode)
|
|
|
|
graph = build_state_mutation_graph(as_node, mode, state_schema)
|
|
accessor = CheckpointStateAccessor.bind(
|
|
graph,
|
|
get_checkpointer(request),
|
|
store=getattr(request.app.state, "store", None),
|
|
mode=mode,
|
|
)
|
|
return accessor, config
|
|
|
|
|
|
# Cache of factory-built accessor graphs. Accessor operations (aget_state /
|
|
# aupdate_state) never execute graph nodes or middleware, so per-request
|
|
# variations (user, model, skills) cannot affect materialization semantics;
|
|
# the compiled graph is stable per (assistant_id, mode, app_config). The
|
|
# factory and app_config identities are re-validated on every call so patched
|
|
# factories take effect immediately and a config.yaml hot-reload (which
|
|
# rebuilds the AppConfig object) never serves a stale compiled graph — the
|
|
# cached reference keeps the old config alive, so id-reuse cannot produce a
|
|
# false hit. Bounded: cleared when too many distinct assistants appear.
|
|
_STATE_ACCESSOR_GRAPH_CACHE_MAX = 64
|
|
_state_accessor_graph_cache: dict[tuple[str | None, str], tuple[Any, Any, Any]] = {}
|
|
|
|
|
|
def _state_accessor_graph(agent_factory: Any, assistant_id: str | None, mode: str, config: dict[str, Any]) -> Any:
|
|
app_config = (config.get("context") or {}).get("app_config")
|
|
key = (assistant_id, mode)
|
|
cached = _state_accessor_graph_cache.get(key)
|
|
if cached is not None and cached[0] is agent_factory and cached[1] is app_config:
|
|
return cached[2]
|
|
if len(_state_accessor_graph_cache) >= _STATE_ACCESSOR_GRAPH_CACHE_MAX:
|
|
_state_accessor_graph_cache.clear()
|
|
graph = agent_factory(config=config)
|
|
_state_accessor_graph_cache[key] = (agent_factory, app_config, graph)
|
|
return graph
|
|
|
|
|
|
class _RawCheckpointSnapshot:
|
|
"""StateSnapshot-shaped view over a raw checkpoint tuple (full mode only).
|
|
|
|
``next``/``tasks`` are not derivable without the compiled graph and
|
|
degrade to empty; everything the read endpoints serialize (values,
|
|
metadata, config ancestry, created_at) comes straight from the tuple.
|
|
"""
|
|
|
|
__slots__ = ("checkpoint_exists", "config", "values", "metadata", "parent_config", "created_at", "tasks", "tasks_known", "next")
|
|
|
|
def __init__(self, config: dict[str, Any], tup: Any | None) -> None:
|
|
self.checkpoint_exists = tup is not None
|
|
self.config = getattr(tup, "config", None) or config
|
|
checkpoint = getattr(tup, "checkpoint", None) or {}
|
|
self.values = dict(checkpoint.get("channel_values") or {})
|
|
self.metadata = dict(getattr(tup, "metadata", None) or {})
|
|
self.parent_config = getattr(tup, "parent_config", None)
|
|
self.created_at = checkpoint.get("ts") or self.metadata.get("created_at", "")
|
|
self.tasks: tuple = ()
|
|
self.tasks_known = False
|
|
self.next: tuple = ()
|
|
|
|
|
|
class _RawCheckpointReadAccessor:
|
|
"""Degraded full-mode read accessor for when the agent factory is down.
|
|
|
|
Full-mode checkpoints persist complete ``channel_values``, so reads do not
|
|
need the compiled graph. The fail-closed delta gate still applies: delta
|
|
checkpoints are rejected with :class:`CheckpointModeMismatchError` instead
|
|
of being served as partial state. Writes are unsupported — mutation paths
|
|
keep using the graph-backed accessor.
|
|
"""
|
|
|
|
def __init__(self, checkpointer: Any, mode: str) -> None:
|
|
self.checkpointer = checkpointer
|
|
self.mode = mode
|
|
|
|
@staticmethod
|
|
def _gate(tup: Any) -> None:
|
|
if checkpoint_tuple_uses_delta(tup):
|
|
raise CheckpointModeMismatchError("Thread requires delta mode; materialize and convert its checkpoints before using full mode.")
|
|
|
|
async def aget(self, config: dict[str, Any]) -> _RawCheckpointSnapshot:
|
|
tup = await self.checkpointer.aget_tuple(config)
|
|
self._gate(tup)
|
|
return _RawCheckpointSnapshot(config, tup)
|
|
|
|
async def ahistory(self, config: dict[str, Any], *, limit: int | None = None) -> list[_RawCheckpointSnapshot]:
|
|
if limit is not None and limit <= 0:
|
|
return []
|
|
result: list[_RawCheckpointSnapshot] = []
|
|
before = None
|
|
walk_config = config
|
|
if config.get("configurable", {}).get("checkpoint_id"):
|
|
# Pregel's get_state_history treats config.checkpoint_id as the
|
|
# inclusive start of the walk, while alist(before=...) is
|
|
# exclusive — fetch the anchor explicitly so the degraded path
|
|
# matches the graph path.
|
|
before = config
|
|
walk_config = {
|
|
**config,
|
|
"configurable": {k: v for k, v in config.get("configurable", {}).items() if k != "checkpoint_id"},
|
|
}
|
|
anchor = await self.checkpointer.aget_tuple(before)
|
|
self._gate(anchor)
|
|
if anchor is not None:
|
|
result.append(_RawCheckpointSnapshot(config, anchor))
|
|
if limit is None or len(result) < limit:
|
|
remaining = None if limit is None else limit - len(result)
|
|
async for tup in self.checkpointer.alist(walk_config, before=before, limit=remaining):
|
|
self._gate(tup)
|
|
result.append(_RawCheckpointSnapshot(config, tup))
|
|
if limit is not None and len(result) >= limit:
|
|
break
|
|
return result
|
|
|
|
|
|
def build_checkpoint_state_accessor(
|
|
request: Request,
|
|
*,
|
|
thread_id: str,
|
|
assistant_id: str | None = None,
|
|
checkpoint_id: str | None = None,
|
|
) -> tuple[CheckpointStateAccessor, dict[str, Any]]:
|
|
"""Build the mode-selected lead graph used for materialized checkpoint state."""
|
|
ctx = get_run_context(request)
|
|
config = build_run_config(thread_id, None, None, assistant_id=assistant_id)
|
|
configurable = config.setdefault("configurable", {})
|
|
configurable["checkpoint_ns"] = ""
|
|
if checkpoint_id is not None:
|
|
configurable["checkpoint_id"] = checkpoint_id
|
|
|
|
if ctx.app_config is not None:
|
|
config.setdefault("context", {})["app_config"] = ctx.app_config
|
|
inject_checkpoint_mode(config, ctx.checkpoint_channel_mode)
|
|
|
|
agent_factory = resolve_agent_factory(assistant_id)
|
|
try:
|
|
graph = _state_accessor_graph(agent_factory, assistant_id, ctx.checkpoint_channel_mode, config)
|
|
except Exception:
|
|
if ctx.checkpoint_channel_mode != "full":
|
|
# Delta materialization needs the graph's channel table; there is
|
|
# no degraded path. Surface the factory failure as-is.
|
|
raise
|
|
# Full-mode checkpoints carry complete channel_values: degrade to raw
|
|
# checkpointer reads so state endpoints survive a broken agent factory
|
|
# (bad model config, MCP server down, misconfigured skill).
|
|
logger.warning(
|
|
"Agent factory unavailable for thread %s; falling back to raw checkpointer reads",
|
|
thread_id,
|
|
exc_info=True,
|
|
)
|
|
return _RawCheckpointReadAccessor(ctx.checkpointer, ctx.checkpoint_channel_mode), config
|
|
accessor = CheckpointStateAccessor.bind(
|
|
graph,
|
|
ctx.checkpointer,
|
|
store=ctx.store,
|
|
mode=ctx.checkpoint_channel_mode,
|
|
)
|
|
return accessor, config
|
|
|
|
|
|
async def resolve_thread_assistant_id(
|
|
request: Request,
|
|
thread_id: str,
|
|
*,
|
|
fail_closed: bool = False,
|
|
) -> str | None:
|
|
"""Return the assistant_id recorded in thread metadata, or ``None``.
|
|
|
|
Missing records degrade to ``None`` (the default lead agent). Store
|
|
failures do the same for read callers, while mutation callers set
|
|
``fail_closed`` so they cannot compile a write graph with the wrong schema.
|
|
"""
|
|
from app.gateway.deps import get_thread_store
|
|
|
|
try:
|
|
thread_store = get_thread_store(request)
|
|
record = await thread_store.get(thread_id)
|
|
except Exception:
|
|
logger.warning("Failed to resolve assistant_id for thread %s", thread_id, exc_info=True)
|
|
if fail_closed:
|
|
raise
|
|
return None
|
|
return record.get("assistant_id") if isinstance(record, dict) else None
|
|
|
|
|
|
async def build_thread_checkpoint_state_accessor(
|
|
request: Request,
|
|
*,
|
|
thread_id: str,
|
|
checkpoint_id: str | None = None,
|
|
fail_closed: bool = False,
|
|
) -> tuple[CheckpointStateAccessor, dict[str, Any]]:
|
|
"""Single resolution boundary for state endpoints.
|
|
|
|
Thread metadata -> assistant_id -> effective assistant graph. Materializing
|
|
with the default lead schema would drop channels contributed by a custom
|
|
``AgentMiddleware.state_schema`` from the response.
|
|
"""
|
|
assistant_id = await resolve_thread_assistant_id(request, thread_id, fail_closed=fail_closed)
|
|
return build_checkpoint_state_accessor(
|
|
request,
|
|
thread_id=thread_id,
|
|
assistant_id=assistant_id,
|
|
checkpoint_id=checkpoint_id,
|
|
)
|
|
|
|
|
|
async def build_thread_checkpoint_state_mutation_accessor(
|
|
request: Request,
|
|
*,
|
|
thread_id: str,
|
|
as_node: str,
|
|
checkpoint_id: str | None = None,
|
|
) -> tuple[CheckpointStateAccessor, dict[str, Any]]:
|
|
"""Mutation accessor compiled with the thread's effective state schema.
|
|
|
|
Derives the schema through :func:`build_thread_checkpoint_state_accessor`
|
|
so writes carrying materialized state do not silently discard
|
|
extension-owned channels.
|
|
"""
|
|
read_accessor, _read_config = await build_thread_checkpoint_state_accessor(
|
|
request,
|
|
thread_id=thread_id,
|
|
checkpoint_id=checkpoint_id,
|
|
fail_closed=True,
|
|
)
|
|
state_schema = graph_state_schema(getattr(read_accessor, "graph", None))
|
|
return build_checkpoint_state_mutation_accessor(
|
|
request,
|
|
thread_id=thread_id,
|
|
as_node=as_node,
|
|
checkpoint_id=checkpoint_id,
|
|
state_schema=state_schema,
|
|
)
|
|
|
|
|
|
async def apply_checkpoint_to_run_config(
|
|
config: dict[str, Any],
|
|
*,
|
|
body: Any,
|
|
thread_id: str,
|
|
request: Request,
|
|
) -> None:
|
|
"""Validate an optional run checkpoint and attach it to RunnableConfig."""
|
|
checkpoint = getattr(body, "checkpoint", None)
|
|
checkpoint_id = getattr(body, "checkpoint_id", None)
|
|
checkpoint_ns = ""
|
|
checkpoint_map = None
|
|
|
|
if checkpoint:
|
|
if not isinstance(checkpoint, Mapping):
|
|
raise HTTPException(status_code=400, detail="checkpoint must be an object")
|
|
checkpoint_thread_id = checkpoint.get("thread_id")
|
|
if checkpoint_thread_id is not None and str(checkpoint_thread_id) != thread_id:
|
|
raise HTTPException(status_code=400, detail="checkpoint thread_id does not match request thread_id")
|
|
raw_checkpoint_id = checkpoint.get("checkpoint_id")
|
|
if raw_checkpoint_id:
|
|
checkpoint_id = str(raw_checkpoint_id)
|
|
raw_checkpoint_ns = checkpoint.get("checkpoint_ns")
|
|
if raw_checkpoint_ns is not None:
|
|
checkpoint_ns = str(raw_checkpoint_ns)
|
|
checkpoint_map = checkpoint.get("checkpoint_map")
|
|
|
|
if not checkpoint_id:
|
|
return
|
|
|
|
read_config: dict[str, Any] = {
|
|
"configurable": {
|
|
"thread_id": thread_id,
|
|
"checkpoint_ns": checkpoint_ns,
|
|
"checkpoint_id": str(checkpoint_id),
|
|
}
|
|
}
|
|
if checkpoint_map is not None:
|
|
read_config["configurable"]["checkpoint_map"] = checkpoint_map
|
|
|
|
checkpointer = get_checkpointer(request)
|
|
try:
|
|
checkpoint_tuple = await checkpointer.aget_tuple(read_config)
|
|
except Exception as exc:
|
|
logger.exception("Failed to validate checkpoint %s for thread %s", checkpoint_id, sanitize_log_param(thread_id))
|
|
raise HTTPException(status_code=500, detail="Failed to validate checkpoint") from exc
|
|
if checkpoint_tuple is None:
|
|
raise HTTPException(status_code=404, detail=f"Checkpoint {checkpoint_id} not found")
|
|
|
|
configurable = config.setdefault("configurable", {})
|
|
if not isinstance(configurable, dict):
|
|
raise HTTPException(status_code=400, detail="request config configurable must be an object")
|
|
configurable["thread_id"] = thread_id
|
|
configurable["checkpoint_ns"] = checkpoint_ns
|
|
configurable["checkpoint_id"] = str(checkpoint_id)
|
|
if checkpoint_map is not None:
|
|
configurable["checkpoint_map"] = checkpoint_map
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Run lifecycle
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
async def start_run(
|
|
body: RunCreateRequest,
|
|
thread_id: str,
|
|
request: Request,
|
|
) -> RunRecord:
|
|
"""Create a RunRecord and launch the background agent task.
|
|
|
|
Parameters
|
|
----------
|
|
body : RunCreateRequest
|
|
The validated request body shared by HTTP and internal launch paths.
|
|
thread_id : str
|
|
Target thread.
|
|
request : Request
|
|
FastAPI request — used to retrieve singletons from ``app.state``.
|
|
"""
|
|
stream_modes = normalize_stream_modes(body.stream_mode)
|
|
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_
|
|
|
|
body_context = getattr(body, "context", None) or {}
|
|
model_name = body_context.get("model_name")
|
|
# Coerce non-string model_name values to str before truncation.
|
|
if model_name is not None and not isinstance(model_name, str):
|
|
model_name = str(model_name)
|
|
|
|
# Validate model against the allowlist when a model_name is provided.
|
|
if model_name:
|
|
app_config = get_app_config()
|
|
resolved = app_config.get_model_config(model_name)
|
|
if resolved is None:
|
|
raise HTTPException(
|
|
status_code=400,
|
|
detail=f"Model {model_name!r} is not in the configured model allowlist",
|
|
)
|
|
|
|
owner_user_id = get_trusted_internal_owner_user_id(request)
|
|
# Stateless run endpoints carry thread_id in the request *body*, so the
|
|
# @require_permission(owner_check=True) decorator -- which resolves ownership
|
|
# from the path param -- cannot protect them. Enforce thread ownership here,
|
|
# before any run is created, so one user cannot start runs on (or read /wait
|
|
# checkpoint state from) another user's thread. Missing rows (auto-created
|
|
# temp threads) and NULL-owner rows (shared / pre-auth data) stay accessible
|
|
# via check_access; only a thread already owned by another user is rejected
|
|
# with 404, matching thread_runs.py's anti-enumeration behaviour. Internal
|
|
# channel runs act on behalf of the connection owner carried in
|
|
# X-DeerFlow-Owner-User-Id, so they are scoped to that owner instead of
|
|
# bypassing the check -- a leaked internal token must not grant cross-user
|
|
# thread access.
|
|
user = getattr(request.state, "user", None)
|
|
if user is not None:
|
|
allowed = await run_ctx.thread_store.check_access(thread_id, str(user.id))
|
|
if not allowed and owner_user_id and getattr(user, "system_role", None) == INTERNAL_SYSTEM_ROLE:
|
|
# Channel workers may also act for the connection owner named in
|
|
# the trusted header (e.g. claiming a legacy default-owned channel
|
|
# thread for its real owner).
|
|
allowed = await run_ctx.thread_store.check_access(thread_id, owner_user_id)
|
|
if not allowed:
|
|
raise HTTPException(status_code=404, detail=f"Thread {thread_id} not found")
|
|
|
|
owner_context_token = set_current_user(SimpleNamespace(id=owner_user_id)) if owner_user_id else None
|
|
try:
|
|
agent_factory = resolve_agent_factory(body.assistant_id)
|
|
is_internal_caller = getattr(getattr(request, "state", None), "auth_source", None) == AUTH_SOURCE_INTERNAL
|
|
command = getattr(body, "command", None)
|
|
if command and command.get("resume") is not None:
|
|
graph_input = Command(resume=command["resume"])
|
|
else:
|
|
graph_input = normalize_input(body.input, trusted_internal=is_internal_caller)
|
|
config = build_run_config(thread_id, body.config, body.metadata, assistant_id=body.assistant_id)
|
|
await apply_checkpoint_to_run_config(config, body=body, thread_id=thread_id, request=request)
|
|
|
|
# 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.
|
|
merge_run_context_overrides(config, getattr(body, "context", None), internal=is_internal_caller)
|
|
if not is_internal_caller:
|
|
# ``body.config`` is free-form and copied verbatim by
|
|
# ``build_run_config``; scrub internal-only keys smuggled there.
|
|
strip_internal_context_keys(config)
|
|
internal_owner_user = await resolve_trusted_internal_owner_for_attribution(request, owner_user_id)
|
|
inject_authenticated_user_context(
|
|
config,
|
|
request,
|
|
internal_owner_user=internal_owner_user,
|
|
request_context=getattr(body, "context", None),
|
|
)
|
|
|
|
async def run_after_metadata(record: RunRecord) -> None:
|
|
metadata_task = asyncio.create_task(
|
|
_ensure_thread_metadata(
|
|
run_ctx,
|
|
record,
|
|
owner_user_id=owner_user_id,
|
|
)
|
|
)
|
|
abort_task = asyncio.create_task(record.abort_event.wait())
|
|
metadata_failure_logged = False
|
|
try:
|
|
done, _ = await asyncio.wait(
|
|
(metadata_task, abort_task),
|
|
timeout=_THREAD_METADATA_SETUP_TIMEOUT_SECONDS,
|
|
return_when=asyncio.FIRST_COMPLETED,
|
|
)
|
|
if metadata_task in done:
|
|
try:
|
|
metadata_task.result()
|
|
except asyncio.CancelledError:
|
|
pass
|
|
except Exception:
|
|
metadata_failure_logged = True
|
|
logger.warning(
|
|
"Failed to ensure thread_meta for %s (non-fatal)",
|
|
sanitize_log_param(thread_id),
|
|
exc_info=True,
|
|
)
|
|
elif abort_task not in done:
|
|
logger.warning(
|
|
"Timed out ensuring thread_meta for %s after %.1fs",
|
|
sanitize_log_param(thread_id),
|
|
_THREAD_METADATA_SETUP_TIMEOUT_SECONDS,
|
|
)
|
|
finally:
|
|
if metadata_task.done():
|
|
if not metadata_failure_logged:
|
|
_log_thread_metadata_task_result(metadata_task, thread_id=thread_id)
|
|
else:
|
|
metadata_task.cancel()
|
|
metadata_task.add_done_callback(
|
|
lambda task: _log_thread_metadata_task_result(
|
|
task,
|
|
thread_id=thread_id,
|
|
)
|
|
)
|
|
if not abort_task.done():
|
|
abort_task.cancel()
|
|
abort_task.add_done_callback(_consume_task_result)
|
|
# Continue through run_agent even after metadata abort/timeout:
|
|
# its startup barrier is the single path that turns pending
|
|
# cancellation into no-agent-construction plus publish_end.
|
|
await 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,
|
|
)
|
|
|
|
try:
|
|
async with goal_thread_lock(thread_id):
|
|
record = await run_mgr.create_or_reject(
|
|
thread_id,
|
|
body.assistant_id,
|
|
on_disconnect=disconnect,
|
|
metadata=body.metadata or {},
|
|
# Persist a secret-redacted copy of the config: the run record is
|
|
# written to runs.kwargs_json and echoed by the run API, so a
|
|
# request-scoped secret (#3861) must not ride along. The live
|
|
# config built above keeps the secrets for the actual run.
|
|
kwargs={"input": body.input, "config": redact_config_secrets(body.config)},
|
|
multitask_strategy=body.multitask_strategy,
|
|
model_name=model_name,
|
|
user_id=owner_user_id,
|
|
)
|
|
|
|
worker = run_after_metadata(record)
|
|
try:
|
|
# No await is allowed between durable admission and task
|
|
# attachment. Metadata setup runs inside the attached
|
|
# worker so a pending cancellation can bypass stalled
|
|
# thread-store IO and still reach run_agent's startup
|
|
# barrier / stream finalization.
|
|
record.task = asyncio.create_task(worker)
|
|
except Exception as exc:
|
|
worker.close()
|
|
await run_mgr.fail_start_if_pending(
|
|
record.run_id,
|
|
error=f"Failed to attach run worker: {exc}",
|
|
)
|
|
raise
|
|
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
|
|
|
|
# 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
|
|
finally:
|
|
if owner_context_token is not None:
|
|
reset_current_user(owner_context_token)
|
|
|
|
|
|
async def launch_scheduled_thread_run(
|
|
*,
|
|
thread_id: str,
|
|
assistant_id: str | None,
|
|
prompt: str,
|
|
request: Request | None = None,
|
|
app: Any | None = None,
|
|
owner_user_id: str | None = None,
|
|
metadata: dict[str, Any] | None = None,
|
|
) -> dict[str, Any]:
|
|
if request is None:
|
|
if app is None:
|
|
raise ValueError("launch_scheduled_thread_run requires request or app")
|
|
request = SimpleNamespace(
|
|
app=app,
|
|
headers=({INTERNAL_OWNER_USER_ID_HEADER_NAME: owner_user_id} if owner_user_id else {}),
|
|
state=SimpleNamespace(
|
|
user=get_internal_user(),
|
|
auth_source=AUTH_SOURCE_INTERNAL,
|
|
),
|
|
cookies={},
|
|
)
|
|
body = RunCreateRequest(
|
|
assistant_id=assistant_id,
|
|
input={"messages": [{"role": "user", "content": prompt}]},
|
|
command=None,
|
|
metadata=metadata or {},
|
|
config=None,
|
|
# ``user_id`` mirrors what IM channels put in ``body.context`` so
|
|
# runtime-context consumers without a ContextVar fallback (e.g.
|
|
# user-scoped GuardrailMiddleware providers) see the owning user;
|
|
# ``inject_authenticated_user_context`` skips the internal user.
|
|
context=({"non_interactive": True, "user_id": owner_user_id} if owner_user_id else {"non_interactive": True}),
|
|
webhook=None,
|
|
checkpoint_id=None,
|
|
checkpoint=None,
|
|
interrupt_before=None,
|
|
interrupt_after=None,
|
|
stream_mode=None,
|
|
stream_subgraphs=False,
|
|
stream_resumable=None,
|
|
on_disconnect="continue",
|
|
on_completion=None,
|
|
multitask_strategy="reject",
|
|
after_seconds=None,
|
|
if_not_exists="create",
|
|
feedback_keys=None,
|
|
)
|
|
record = await start_run(body, thread_id, request)
|
|
return {"run_id": record.run_id, "thread_id": record.thread_id}
|
|
|
|
|
|
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")
|
|
if await _terminal_record_stream_missing(bridge, record):
|
|
yield format_sse("end", None)
|
|
return
|
|
|
|
gap_emitted = False
|
|
try:
|
|
async for entry in bridge.subscribe(record.run_id, last_event_id=last_event_id):
|
|
if await request.is_disconnected():
|
|
break
|
|
|
|
if isinstance(entry, StreamGap):
|
|
gap_emitted = True
|
|
yield format_sse(
|
|
"gap",
|
|
{
|
|
"code": "stream_replay_gap",
|
|
"run_id": record.run_id,
|
|
"requested_event_id": entry.requested_event_id,
|
|
"earliest_available_event_id": entry.earliest_available_event_id,
|
|
"latest_available_event_id": entry.latest_available_event_id,
|
|
"recovery": "reload_durable_state",
|
|
},
|
|
)
|
|
return
|
|
|
|
if entry is HEARTBEAT_SENTINEL:
|
|
if await _orphan_recovery_observed_after_heartbeat(record, run_mgr):
|
|
yield format_sse("end", None)
|
|
return
|
|
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:
|
|
# store_only records are cross-worker runs hydrated from the RunStore; this
|
|
# worker holds no in-memory task/abort state for them, so run_mgr.cancel()
|
|
# cannot stop the task (it would 409). Skip on_disconnect cancellation for
|
|
# those and only act on runs this worker actually owns.
|
|
if not gap_emitted and not record.store_only and record.status in (RunStatus.pending, RunStatus.running):
|
|
if record.on_disconnect == DisconnectMode.cancel:
|
|
await run_mgr.cancel(record.run_id)
|
|
|
|
|
|
async def wait_for_run_completion(
|
|
bridge: StreamBridge,
|
|
record: RunRecord,
|
|
request: Request,
|
|
run_mgr: RunManager,
|
|
) -> bool:
|
|
"""Block until the run publishes ``END_SENTINEL``, honouring on_disconnect.
|
|
|
|
The non-streaming ``/wait`` endpoints used to ``await record.task``
|
|
directly with no disconnect handling. When the client (or an
|
|
intermediate HTTP proxy) timed out during a long tool call such as
|
|
``pip install``, the handler would swallow ``CancelledError`` and
|
|
serialize whatever checkpoint happened to exist — masking a half-finished
|
|
run as a normal completion (issue #3265).
|
|
|
|
This helper consumes the same bridge that ``sse_consumer`` does so the
|
|
wait path shares its disconnect semantics: each wake-up polls
|
|
``request.is_disconnected()``; on a real disconnect it cancels the
|
|
background run when ``record.on_disconnect`` is ``cancel``. The bridge's
|
|
heartbeat sentinels guarantee at least one wake-up per
|
|
``heartbeat_interval`` even when the agent emits no events for a while.
|
|
|
|
Returns:
|
|
``True`` when ``END_SENTINEL`` was observed (run reached a terminal
|
|
state), ``False`` when the loop exited because the client
|
|
disconnected. Callers must skip checkpoint serialization on
|
|
``False`` so a partial checkpoint is not returned as a normal
|
|
response.
|
|
"""
|
|
completed = False
|
|
if await _terminal_record_stream_missing(bridge, record):
|
|
return True
|
|
|
|
resume_from_event_id: str | None = None
|
|
try:
|
|
while True:
|
|
gap_seen = False
|
|
async for entry in bridge.subscribe(record.run_id, last_event_id=resume_from_event_id):
|
|
# END_SENTINEL means the run reached a terminal state; honour it
|
|
# even if the client just disconnected so the caller still serializes
|
|
# the real final checkpoint.
|
|
if entry is END_SENTINEL:
|
|
completed = True
|
|
return True
|
|
if isinstance(entry, StreamGap):
|
|
# The wait API only needs terminal completion, not a complete
|
|
# event replay. Resume at the retained tail rather than
|
|
# treating a bridge gap as a client disconnect.
|
|
resume_from_event_id = entry.latest_available_event_id
|
|
gap_seen = True
|
|
break
|
|
if entry is HEARTBEAT_SENTINEL and await _orphan_recovery_observed_after_heartbeat(record, run_mgr):
|
|
completed = True
|
|
return True
|
|
if await request.is_disconnected():
|
|
return False
|
|
# Heartbeats and regular events: keep waiting for END_SENTINEL.
|
|
if not gap_seen:
|
|
return completed
|
|
finally:
|
|
if not completed and record.status in (RunStatus.pending, RunStatus.running):
|
|
if record.on_disconnect == DisconnectMode.cancel:
|
|
await run_mgr.cancel(record.run_id)
|