From beaae220882ed2ba585d86a4c3952251e3f2b78c Mon Sep 17 00:00:00 2001 From: Wenchao An Date: Sat, 3 Oct 2026 21:16:01 +0800 Subject: [PATCH] feat(extensions): add host-bound Agent run control (#6190) * feat(extensions): add host-bound Agent run control * fix(extensions): preserve internal launches and scope run control * docs(extensions): use a plugin-local run operation key --- backend/app/gateway/app.py | 8 + backend/app/gateway/deps.py | 11 + backend/app/gateway/extension_agent_runs.py | 182 +++++++++ backend/app/gateway/routers/plugins.py | 6 +- backend/docs/extension-agent-runs.md | 108 ++++++ .../deerflow_extension_api/__init__.py | 10 +- .../deerflow_extension_api/agent_runs.py | 102 +++++ .../deerflow_extension_api/plugins.py | 4 +- backend/packages/extension-api/pyproject.toml | 2 +- .../harness/deerflow/extensions/AGENTS.md | 10 + .../deerflow/extensions/plugin_tools.py | 3 + .../harness/deerflow/runtime/runs/worker.py | 10 + .../deerflow/runtime/secret_context.py | 3 + backend/packages/harness/pyproject.toml | 2 +- backend/tests/test_extension_agent_runs.py | 366 ++++++++++++++++++ backend/tests/test_extension_api_contracts.py | 2 +- backend/tests/test_plugin_tools.py | 22 ++ backend/uv.lock | 4 +- docs/full-stack-plugins.md | 7 +- 19 files changed, 853 insertions(+), 9 deletions(-) create mode 100644 backend/app/gateway/extension_agent_runs.py create mode 100644 backend/docs/extension-agent-runs.md create mode 100644 backend/packages/extension-api/deerflow_extension_api/agent_runs.py create mode 100644 backend/tests/test_extension_agent_runs.py diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index ce6e48f576..d6a5e44020 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -1080,6 +1080,14 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for setattr(app.state, RUN_EVIDENCE_READER_RESOLVER_KEY, _resolve_extension_run_evidence_reader) + from deerflow_extension_api.agent_runs import AGENT_RUNS_RESOLVER_KEY + + def _resolve_extension_agent_runs(request): + host = getattr(app.state, "agent_runs_host", None) + return host.bind(request) if host is not None else None + + setattr(app.state, AGENT_RUNS_RESOLVER_KEY, _resolve_extension_agent_runs) + # CSRF: Double Submit Cookie pattern for state-changing requests app.add_middleware(CSRFMiddleware) diff --git a/backend/app/gateway/deps.py b/backend/app/gateway/deps.py index 9815841eee..bf27cc9702 100644 --- a/backend/app/gateway/deps.py +++ b/backend/app/gateway/deps.py @@ -629,12 +629,19 @@ async def langgraph_runtime(app: FastAPI, startup_config: AppConfig) -> AsyncGen recovered_runs, ) + from app.gateway.auth.repositories.sqlite import SQLiteUserRepository + from app.gateway.extension_agent_runs import GatewayAgentRunsHost + + app.state.agent_runs_host = GatewayAgentRunsHost(app, load_user=SQLiteUserRepository(sf).get_user_by_id if sf is not None else None) + stack.callback(app.state.agent_runs_host.close) + # Start the lease heartbeat if enabled (multi-worker deployments). await app.state.run_manager.start_heartbeat() try: yield finally: + app.state.agent_runs_host.close() # Drain in-flight run tasks BEFORE the AsyncExitStack tears down the # checkpointer (and its connection pool). A run still mid-graph would # otherwise leak into asyncio.run() shutdown, where langgraph's @@ -759,7 +766,11 @@ def get_run_context(request: Request) -> RunContext: captured in :func:`langgraph_runtime` so callers never see a store bound to one backend paired with a config pointing at another. """ + host = getattr(request.app.state, "agent_runs_host", None) + # Internal/channel and PAT runs deliberately receive no retained delegation. + agent_runs = host.bind(request) if host is not None else None return RunContext( + agent_runs=agent_runs, checkpointer=get_checkpointer(request), store=get_store(request), event_store=get_run_event_store(request), diff --git a/backend/app/gateway/extension_agent_runs.py b/backend/app/gateway/extension_agent_runs.py new file mode 100644 index 0000000000..3bcaea74c2 --- /dev/null +++ b/backend/app/gateway/extension_agent_runs.py @@ -0,0 +1,182 @@ +"""Adapt the normal Gateway run routes to a revocable extension capability.""" + +from __future__ import annotations + +import asyncio +import json +import math +from collections.abc import Awaitable, Callable, Mapping +from contextlib import asynccontextmanager +from typing import Any + +from deerflow_extension_api.agent_runs import AgentRun, AgentRunError +from fastapi import HTTPException, Request +from pydantic import TypeAdapter, ValidationError + +from app.gateway.auth_disabled import AUTH_SOURCE_AUTH_DISABLED, AUTH_SOURCE_SESSION, get_auth_disabled_user, is_auth_disabled +from app.gateway.authz import AuthContext, resolve_route_permissions +from deerflow.runtime.user_context import reset_current_user, set_current_user +from deerflow.utils.thread_id import ThreadId + +_THREAD_ID = TypeAdapter(ThreadId) + + +def _json_object(value: Mapping[str, Any] | None) -> dict[str, Any] | None: + if value is None: + return None + try: + raw = json.dumps(dict(value), allow_nan=False) + if len(raw.encode()) > 256 * 1024: + raise ValueError("payload too large") + return json.loads(raw) + except (TypeError, ValueError) as exc: + raise AgentRunError(422, "Agent run input must be a JSON object of at most 256 KiB") from exc + + +def _view(response) -> AgentRun: + return AgentRun(response.thread_id, response.run_id, response.status, response.assistant_id, response.stop_reason) + + +class GatewayAgentRunsHost: + """One app lifetime; never retain an HTTP Request or its credentials.""" + + def __init__(self, app, *, load_user: Callable[[str], Awaitable[Any]] | None): + self.app = app + self.load_user = load_user + self.active = True + + def close(self) -> None: + self.active = False + + def bind(self, request: Request): + user = getattr(request.state, "user", None) + auth = getattr(request.state, "auth", None) + if user is None or auth is None or not auth.is_authenticated: + return None + if str(auth.user.id) != str(user.id): + raise PermissionError("Agent run control requires an authenticated user") + source = getattr(request.state, "auth_source", None) + # PAT revocation and internal-owner delegation need separate durable + # grants. Never turn either into an unrestricted session capability. + if source not in (AUTH_SOURCE_SESSION, AUTH_SOURCE_AUTH_DISABLED): + return None + return _BoundAgentRuns(self, str(user.id), user.token_version, source, frozenset(auth.permissions)) + + +class _BoundAgentRuns: + def __init__(self, host, user_id: str, token_version: int, source: str, permissions: frozenset[str], plugin_namespace: str | None = None): + self._host = host + self._user_id = user_id + self._token_version = token_version + self._source = source + self._permissions = permissions + self._plugin_namespace = plugin_namespace + + def for_plugin(self, namespace: str): + if not isinstance(namespace, str) or not namespace or len(namespace) > 96 or ":" in namespace: + raise AgentRunError(422, "Invalid plugin namespace") + if self._plugin_namespace is not None: + if namespace != self._plugin_namespace: + raise AgentRunError(422, "Agent run handle is already bound to a plugin") + return self + return _BoundAgentRuns(self._host, self._user_id, self._token_version, self._source, self._permissions, namespace) + + @asynccontextmanager + async def _request(self, permission: str, thread_id: str | None = None): + if not self._host.active: + raise AgentRunError(503, "Agent run host unavailable") + if self._source == AUTH_SOURCE_AUTH_DISABLED: + user = get_auth_disabled_user() if is_auth_disabled() else None + else: + user = await self._host.load_user(self._user_id) if self._host.load_user is not None else None + if user is None or str(user.id) != self._user_id or user.token_version != self._token_version or user.needs_setup: + raise AgentRunError(403, "Agent run authorization revoked") + permissions = self._permissions.intersection(await resolve_route_permissions(user, is_internal=False)) + if permission not in permissions: + raise AgentRunError(403, f"Permission denied: {permission}") + if not self._host.active: + raise AgentRunError(503, "Agent run host unavailable") + request = Request({"type": "http", "app": self._host.app, "method": "POST", "path": "/api/threads", "headers": [], "query_string": b"", "state": {}}) + request.state.user = user + request.state.auth_source = self._source + request.state.auth = AuthContext(user, list(permissions)) + token = set_current_user(user) + try: + if thread_id is not None: + _THREAD_ID.validate_python(thread_id) + # Read routes historically accept untracked legacy threads. + # The delegated handle requires a live owned/shared row too. + store = self._host.app.state.thread_store + if not await store.check_access(thread_id, self._user_id, require_existing=True): + raise AgentRunError(404, "Thread not found") + yield request + except HTTPException as exc: + raise AgentRunError(exc.status_code, str(exc.detail)) from exc + except ValidationError as exc: + raise AgentRunError(422, "Invalid Agent run request") from exc + finally: + reset_current_user(token) + + async def create_thread(self, *, assistant_id="lead_agent", thread_id=None, metadata=None) -> str: + from app.gateway.routers.threads import ThreadCreateRequest, create_thread + + async with self._request("threads:write") as request: + response = await create_thread(ThreadCreateRequest(assistant_id=assistant_id, thread_id=thread_id, metadata=_json_object(metadata) or {}), request) + return response.thread_id + + async def _start(self, *, thread_id, input=None, resume=None, context=None, idempotency_key=None): + from app.gateway.routers.thread_runs import create_run + from app.gateway.run_models import RunCreateRequest + from app.gateway.services import resolve_thread_assistant_id + + if idempotency_key is not None and (not isinstance(idempotency_key, str) or not idempotency_key.strip() or len(idempotency_key) > 200): + raise AgentRunError(422, "Idempotency key must contain 1 to 200 characters") + # Copy inputs before the first await; plugin tasks cannot mutate admitted data. + run_input = _json_object(input) + run_context = _json_object(context) + command = _json_object({"resume": resume}) if resume is not None else None + if run_context is not None and any(key in run_context for key in ("agent_name", "is_bootstrap")): + raise AgentRunError(422, "Agent identity is bound to the thread") + async with self._request("runs:create", thread_id) as request: + assistant_id = await resolve_thread_assistant_id(request, thread_id, fail_closed=True) + body = RunCreateRequest(assistant_id=assistant_id, input=run_input, command=command, context=run_context, on_disconnect="continue") + response = await create_run(thread_id, body, request, idempotency_key=f"extension:{self._plugin_namespace or ''}:{idempotency_key}" if idempotency_key is not None else None) + return _view(response) + + async def start(self, *, thread_id, input, context=None, idempotency_key=None) -> AgentRun: + return await self._start(thread_id=thread_id, input=input, context=context, idempotency_key=idempotency_key) + + async def resume(self, *, thread_id, resume, idempotency_key=None) -> AgentRun: + if resume is None: + raise AgentRunError(422, "An explicit interrupt response is required") + return await self._start(thread_id=thread_id, resume=resume, idempotency_key=idempotency_key) + + async def get(self, *, thread_id, run_id) -> AgentRun: + from app.gateway.routers.thread_runs import get_run + + async with self._request("runs:read", thread_id) as request: + return _view(await get_run(thread_id, run_id, request)) + + async def get_state(self, *, thread_id) -> dict[str, Any]: + from app.gateway.routers.threads import get_thread_state + + async with self._request("threads:read", thread_id) as request: + return (await get_thread_state(thread_id, request)).model_dump(mode="json") + + async def cancel(self, *, thread_id, run_id) -> None: + from app.gateway.routers.thread_runs import cancel_run + + async with self._request("runs:cancel", thread_id) as request: + await cancel_run(thread_id, run_id, request, wait=False, action="interrupt") + + async def wait(self, *, thread_id, run_id, timeout=60) -> AgentRun: + if not isinstance(timeout, (int, float)) or not math.isfinite(timeout) or not 0 < timeout <= 3600: + raise AgentRunError(422, "Wait timeout must be between 0 and 3600 seconds") + async with asyncio.timeout(timeout): + delay = 0.25 + while True: + run = await self.get(thread_id=thread_id, run_id=run_id) + if run.status not in ("pending", "running"): + return run + await asyncio.sleep(delay) + delay = min(delay * 2, 4) diff --git a/backend/app/gateway/routers/plugins.py b/backend/app/gateway/routers/plugins.py index 2e63b0c723..5b6f1be4c6 100644 --- a/backend/app/gateway/routers/plugins.py +++ b/backend/app/gateway/routers/plugins.py @@ -93,6 +93,7 @@ async def plugin_asset(request: Request, namespace: str, revision: str, path: st @router.post("/{namespace}/actions/{action_name}") async def invoke_plugin_action(request: Request, namespace: str, action_name: str): """Invoke an installed action with the authenticated viewer and deployment settings.""" + from deerflow_extension_api.agent_runs import AgentRunError, resolve_agent_runs from deerflow_extension_api.auth import resolve_principal from deerflow_extension_api.plugins import ActionContext @@ -136,7 +137,10 @@ async def invoke_plugin_action(request: Request, namespace: str, action_name: st raise HTTPException(422, "Plugin action requires a JSON object.") from exc try: async with asyncio.timeout(30): - return await action.handler(MappingProxyType(payload), ActionContext(principal, MappingProxyType(settings))) + runs = resolve_agent_runs(request) + return await action.handler(MappingProxyType(payload), ActionContext(principal, MappingProxyType(settings), agent_runs=runs.for_plugin(namespace) if runs is not None else None)) + except AgentRunError as exc: + raise HTTPException(exc.status_code, str(exc)) from exc except TimeoutError as exc: raise HTTPException(504, "Plugin action timed out.") from exc except ValueError as exc: diff --git a/backend/docs/extension-agent-runs.md b/backend/docs/extension-agent-runs.md new file mode 100644 index 0000000000..e3c485fa74 --- /dev/null +++ b/backend/docs/extension-agent-runs.md @@ -0,0 +1,108 @@ +# Agent run control for extensions + +`deerflow_extension_api.AgentRuns` is an optional, host-bound capability for +creating and continuing **full Agent runs**. Runs use Gateway's normal admission, +Agent assembly, tools, middleware, checkpoints, run records, and cancellation. +It is distinct from `ModelInvoker`, which makes one bounded model invocation. +No workflow, team roster, delegation policy, or handoff scheduler is built in. + +## Acquire the capability + +Unified backend actions and model tools receive `context.agent_runs`. Contributed +Gateway routes can call `require_agent_runs(request)` without importing private +host modules. An unsupported host returns `None` from `resolve_agent_runs` and +raises `NotImplementedError` from the required variant. + +```python +from deerflow_extension_api import ActionContext + +async def begin(payload, context: ActionContext): + runs = context.agent_runs + if runs is None: + raise RuntimeError("This host does not support Agent run control") + thread_id = await runs.create_thread(assistant_id="researcher") + run = await runs.start( + thread_id=thread_id, + input={"messages": [{"role": "user", "content": payload["prompt"]}]}, + idempotency_key=payload["operation_id"], + ) + return {"thread_id": run.thread_id, "run_id": run.run_id} +``` + +Gateway binds the handle from authenticated request state. Plugins cannot pass a +user ID to select another principal. Session credentials and local auth-disabled +mode are supported. PAT and internal/channel credentials deliberately do not +receive retained run control: their revocation and owner delegation require a +separate grant contract. Embedded hosts may implement the protocol themselves. +The Gateway worker carries the capability only in its reserved runtime context; +untrusted request input cannot replace it. Direct embedded or subagent execution +does not acquire it automatically. + +## Operations + +- `create_thread(assistant_id="lead_agent", thread_id=None, metadata=None)` creates + an owned thread. A caller-chosen ID is idempotent under the ordinary thread + creation rules. Use a stable ID when recovering uncertain creation. +- `start(thread_id, input, context=None, idempotency_key=None)` starts or continues + the Agent associated with that existing thread. Select a Custom Agent when + creating the thread; changing `agent_name` or bootstrapping through `context` + is rejected. Model/context options follow normal host validation. External + system/developer messages are rejected. Concurrent starts conflict rather than + implicitly interrupting each other. +- `get(thread_id, run_id)` returns an `AgentRun` with IDs, status, originating + assistant, and stop reason. It follows the normal owner-scoped run API. +- `wait(thread_id, run_id, timeout=60)` polls until the run leaves pending/running. + The timeout must be positive and at most 3600 seconds. Timeout or cancellation + of the waiting task leaves the Agent run active; call `cancel` explicitly. +- `get_state(thread_id)` returns the latest normal thread-state projection, + including values, next nodes and interrupt information. This is **mutable + thread state**, not a snapshot permanently associated with a run ID. +- `resume(thread_id, resume, idempotency_key=None)` submits an explicit LangGraph + interrupt response through ordinary admission. It never decides whether a + pending action should be approved. +- `cancel(thread_id, run_id)` requests the normal interrupt action and preserves + checkpoints. A remote worker may acknowledge before cancellation completes; + use `wait`/`get` to observe the final status. + +Run operations use keyword-only arguments. Input and context are detached JSON objects, +limited to 256 KiB each. Idempotency keys are 1–200 characters. Action/tool handles +automatically scope keys to the registered plugin namespace; contributed routes +use `require_agent_runs(request).for_plugin("example.team")` for the same isolation. +A scoped handle cannot be rebound to another plugin. Gateway additionally scopes +keys to the user and thread. Callers may use colons in their local keys. +Reusing a key with different input is a conflict under existing admission rules. +`AgentRunError.status_code` exposes normal HTTP-equivalent rejection categories, +including denied permissions, missing resources, conflicts, and invalid input. +Unified backend actions preserve these statuses in their HTTP response. + +`wait` polls immediately, then backs off from 0.25 seconds to a four-second cap. +The caller's timeout bounds the entire wait, including authorization and polling. + +A completed graph invocation can have run status `success` while paused at a +LangGraph interrupt. This matches existing Gateway semantics: inspect the returned +thread state's `next`/`interrupts` before treating success as completed work. +Do not automatically resume or approve merely because a wait returned. + +## Service-owned handoffs and lifetime + +A trusted plugin service may retain a handle handed to it by an authenticated +action/tool and use it after that handler returns. For an A-to-B handoff, wait +for A, inspect its state and pending interrupts, apply the plugin's policy, then +create B's thread and pass the selected result as B's human input. The plugin +owns its task lifecycle, supervision, concurrency limits, retry policy, and any +durable workflow state. Action/tool handlers retain their existing 30-second +execution limit; long work belongs in a supervised plugin service. + +Handles contain no cookies or bearer tokens. They delegate the original user's +permission ceiling for this process lifetime rather than storing the request. +Every operation reloads the session user, rejects deletion/password-reset token +version changes or incomplete setup, intersects current route permissions with +the original grant, and checks thread ownership. Agent/model admission uses that +fresh user. Local auth-disabled grants stop working when that mode is disabled. +Gateway revokes all handles before draining runs at shutdown. + +The handle is not serializable and cannot survive a restart. Persist thread/run +IDs and workflow progress, then reacquire a capability from a newly authenticated +request to continue. Existing run/checkpoint durability and orphan recovery remain +the host's responsibility. The original route-permission ceiling cannot widen; Agent/model/tool policy +still evaluates the current user. Do not expose a retained handle to other users. diff --git a/backend/packages/extension-api/deerflow_extension_api/__init__.py b/backend/packages/extension-api/deerflow_extension_api/__init__.py index a6555fbb40..5b78058f1e 100644 --- a/backend/packages/extension-api/deerflow_extension_api/__init__.py +++ b/backend/packages/extension-api/deerflow_extension_api/__init__.py @@ -7,6 +7,7 @@ extensions can therefore be released independently of the host. from __future__ import annotations +from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY, AGENT_RUNS_RESOLVER_KEY, AgentRun, AgentRunError, AgentRuns, require_agent_runs, resolve_agent_runs from deerflow_extension_api.assembly import ( AgentAssemblyDescriptor, AgentAssemblyObserver, @@ -98,9 +99,16 @@ from deerflow_extension_api.state import ExtensionData #: Contract version. Before 1.0, minors may break and patches are additive. #: From 1.0 on, bump the major for breaking changes. -API_VERSION = "0.2.4" +API_VERSION = "0.2.5" __all__ = [ + "AGENT_RUNS_CONTEXT_KEY", + "AGENT_RUNS_RESOLVER_KEY", + "AgentRun", + "AgentRunError", + "AgentRuns", + "require_agent_runs", + "resolve_agent_runs", "ModelInvocationError", "ModelInvocationFailed", "ModelInvocationRequest", diff --git a/backend/packages/extension-api/deerflow_extension_api/agent_runs.py b/backend/packages/extension-api/deerflow_extension_api/agent_runs.py new file mode 100644 index 0000000000..01754ab879 --- /dev/null +++ b/backend/packages/extension-api/deerflow_extension_api/agent_runs.py @@ -0,0 +1,102 @@ +"""Optional host-bound control of full Agent runs, independent of Gateway imports.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any, Protocol + +AGENT_RUNS_RESOLVER_KEY = "deerflow_agent_runs_resolver" +AGENT_RUNS_CONTEXT_KEY = "__deerflow_extension_agent_runs" + + +class AgentRunError(Exception): + """Host rejection with an HTTP-equivalent status (403, 404, 409, 422, 503).""" + + def __init__(self, status_code: int, message: str): + super().__init__(message) + self.status_code = status_code + + +@dataclass(frozen=True) +class AgentRun: + thread_id: str + run_id: str + status: str + assistant_id: str | None = None + stop_reason: str | None = None + + +class AgentRuns(Protocol): + """A process-local delegated capability bound by the host to one user. + + The host rechecks permissions and ownership on every operation. Retaining + this handle after a handler returns permits a service to orchestrate runs; + it is not a credential to serialize or reuse after host shutdown/restart. + """ + + def for_plugin(self, namespace: str) -> AgentRuns: + """Partition idempotency keys without widening this user's permissions. + + Action/tool dispatch scopes handles to the registered plugin namespace. + Contributed routes scope request-resolved handles explicitly. A scoped + handle cannot be rebound to a different plugin. + """ + ... + + async def create_thread(self, *, assistant_id: str = "lead_agent", thread_id: str | None = None, metadata: Mapping[str, Any] | None = None) -> str: + """Create an owned thread, or return the caller's existing thread ID.""" + ... + + async def start(self, *, thread_id: str, input: Mapping[str, Any], context: Mapping[str, Any] | None = None, idempotency_key: str | None = None) -> AgentRun: + """Start or continue the thread's Agent with ordinary run admission. + + Scoped handles namespace local idempotency keys automatically. An active + run conflicts; start never implicitly cancels it. + """ + ... + + async def resume(self, *, thread_id: str, resume: Any, idempotency_key: str | None = None) -> AgentRun: + """Submit an explicit interrupt response; never approve automatically.""" + ... + + async def get(self, *, thread_id: str, run_id: str) -> AgentRun: + """Read current status, including on a different Gateway worker.""" + ... + + async def get_state(self, *, thread_id: str) -> dict[str, Any]: + """Read the thread's latest API state (values, next, interrupts, etc.). + + This is thread state, not an immutable run result. Serialize follow-up + admissions if a plugin needs to associate output with one specific run. + """ + ... + + async def wait(self, *, thread_id: str, run_id: str, timeout: float = 60) -> AgentRun: + """Wait for a terminal/interrupt status, or raise TimeoutError. + + Timeout or cancellation of the caller does not cancel the Agent run. + """ + ... + + async def cancel(self, *, thread_id: str, run_id: str) -> None: + """Request interrupt, retaining checkpoints. Completion may be async.""" + ... + + +def resolve_agent_runs(request: object) -> AgentRuns | None: + """Resolve from host-authenticated request state, never a supplied user ID. + + Unsupported hosts, credentials or unstamped requests return None. Inconsistent + authenticated identities raise PermissionError. No ambient user fallback exists. + """ + state = getattr(getattr(request, "app", None), "state", None) + resolver = getattr(state, AGENT_RUNS_RESOLVER_KEY, None) + return resolver(request) if callable(resolver) else None + + +def require_agent_runs(request: object) -> AgentRuns: + runs = resolve_agent_runs(request) + if runs is None: + raise NotImplementedError("host-bound Agent run control is unavailable") + return runs diff --git a/backend/packages/extension-api/deerflow_extension_api/plugins.py b/backend/packages/extension-api/deerflow_extension_api/plugins.py index 69abbdb8dd..c745e0ba85 100644 --- a/backend/packages/extension-api/deerflow_extension_api/plugins.py +++ b/backend/packages/extension-api/deerflow_extension_api/plugins.py @@ -7,10 +7,11 @@ and an authenticated principal for each admitted call; this is not a sandbox. from __future__ import annotations from collections.abc import Awaitable, Callable, Mapping -from dataclasses import dataclass +from dataclasses import dataclass, field from pathlib import Path from typing import Any +from deerflow_extension_api.agent_runs import AgentRuns from deerflow_extension_api.auth import ExtensionPrincipal from deerflow_extension_api.settings import FrontendBinding, SettingsContribution, SettingsField, SettingValue @@ -19,6 +20,7 @@ from deerflow_extension_api.settings import FrontendBinding, SettingsContributio class ActionContext: principal: ExtensionPrincipal settings: Mapping[str, SettingValue] + agent_runs: AgentRuns | None = field(default=None, kw_only=True) @dataclass(frozen=True) diff --git a/backend/packages/extension-api/pyproject.toml b/backend/packages/extension-api/pyproject.toml index 8b2c231e5e..2423dab995 100644 --- a/backend/packages/extension-api/pyproject.toml +++ b/backend/packages/extension-api/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "deerflow-extension-api" -version = "0.2.4" +version = "0.2.5" description = "Public contracts for DeerFlow extensions" requires-python = ">=3.12" # Keep the contract package import-light and independent from the host. Public diff --git a/backend/packages/harness/deerflow/extensions/AGENTS.md b/backend/packages/harness/deerflow/extensions/AGENTS.md index ab282f603e..97b0db7383 100644 --- a/backend/packages/harness/deerflow/extensions/AGENTS.md +++ b/backend/packages/harness/deerflow/extensions/AGENTS.md @@ -422,3 +422,13 @@ No online settings write API is added. `plugin_tools.py` joins normal tool assem the run's extension snapshot; task delegation passes that snapshot explicitly. Browser public-field projection is an allowlist. Package code is trusted, not sandboxed. See `docs/full-stack-plugins.md` and the independently packaged bookmark example. + +Full Agent run control is an optional `deerflow_extension_api.AgentRuns` handle +on action/tool contexts and the request resolver. Gateway owns principal binding, +revocation and ordinary route admission in `app/gateway/extension_agent_runs.py`. +Never replace this with `ModelInvoker`, raw global RunManager access, or a +caller-supplied user ID. Service-held handles are process-local, permission-capped +and revoked before host shutdown; PAT/internal grants remain unsupported. +Unstamped internal launches receive no handle and must still start normally. +Action/tool dispatch scopes handles to each registered plugin namespace for +idempotency isolation; request-resolved handles use `for_plugin` explicitly. diff --git a/backend/packages/harness/deerflow/extensions/plugin_tools.py b/backend/packages/harness/deerflow/extensions/plugin_tools.py index 16727a71be..0e44e6877a 100644 --- a/backend/packages/harness/deerflow/extensions/plugin_tools.py +++ b/backend/packages/harness/deerflow/extensions/plugin_tools.py @@ -11,6 +11,7 @@ from collections.abc import Mapping from copy import deepcopy from types import MappingProxyType +from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY from deerflow_extension_api.auth import ExtensionPrincipal from deerflow_extension_api.plugins import ToolContext from jsonschema import Draft202012Validator @@ -71,10 +72,12 @@ def _build_tool(source, plugin, declaration): if len(json.dumps(payload, allow_nan=False).encode()) > 256 * 1024 or not validator.is_valid(payload): raise ToolException("Invalid plugin tool input.") context = runtime.context if isinstance(runtime.context, Mapping) else {} + runs = context.get(AGENT_RUNS_CONTEXT_KEY) tool_context = ToolContext( ExtensionPrincipal(resolve_runtime_user_id(runtime)), MappingProxyType(settings), context.get("thread_id"), + agent_runs=runs.for_plugin(plugin.namespace) if runs is not None else None, ) async with asyncio.timeout(30): result = await declaration.handler(MappingProxyType(payload), tool_context) diff --git a/backend/packages/harness/deerflow/runtime/runs/worker.py b/backend/packages/harness/deerflow/runtime/runs/worker.py index fe44e4b4e9..69d4458bc5 100644 --- a/backend/packages/harness/deerflow/runtime/runs/worker.py +++ b/backend/packages/harness/deerflow/runtime/runs/worker.py @@ -33,6 +33,7 @@ from datetime import datetime from functools import lru_cache from typing import Any, Final, Literal, cast +from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY from langgraph.checkpoint.base import empty_checkpoint from langgraph.types import Overwrite @@ -218,6 +219,7 @@ def _release_run_scoped_references( "__run_journal", CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY, CONVERSATION_READER_CONTEXT_KEY, + AGENT_RUNS_CONTEXT_KEY, } try: from deerflow.extensions import EXTENSION_SNAPSHOT_CONTEXT_KEY @@ -242,6 +244,7 @@ def _release_run_scoped_references( if isinstance(configurable, dict): configurable.pop("__pregel_runtime", None) configurable.pop(CONVERSATION_READER_CONTEXT_KEY, None) + configurable.pop(AGENT_RUNS_CONTEXT_KEY, None) context = runnable_config.get("context") if isinstance(context, dict): for key in internal_context_keys: @@ -538,6 +541,7 @@ _SERVER_OWNED_RUNTIME_CONTEXT_KEYS: Final[frozenset[str]] = ( CURRENT_RUN_PRE_EXISTING_MESSAGE_IDS_KEY, DEERFLOW_TRACE_METADATA_KEY, CONVERSATION_READER_CONTEXT_KEY, + AGENT_RUNS_CONTEXT_KEY, THREAD_INCARNATION_CONTEXT_KEY, THREAD_INCARNATION_METADATA_GUARD_KEY, "is_subagent", @@ -564,6 +568,7 @@ def _build_runtime_context( task_store: Any | None = None, extensions: Any | None = None, conversation_reader: Any | None = None, + agent_runs: Any | None = None, *, thread_incarnation: str | None | object = _THREAD_INCARNATION_UNSET, ) -> dict[str, Any]: @@ -589,6 +594,8 @@ def _build_runtime_context( runtime_ctx.setdefault(key, value) if app_config is not None: runtime_ctx["app_config"] = app_config + if agent_runs is not None: + runtime_ctx[AGENT_RUNS_CONTEXT_KEY] = agent_runs if conversation_reader is not None: runtime_ctx[CONVERSATION_READER_CONTEXT_KEY] = conversation_reader if task_store is not None: @@ -650,6 +657,7 @@ class RunContext: on_run_completed: Any | None = field(default=None) # The host binds this capability to one run's authenticated reader and references. conversation_reader: Any | None = field(default=None) + agent_runs: Any | None = field(default=None) def _install_runtime_context(config: dict, runtime_context: dict[str, Any]) -> None: @@ -658,6 +666,7 @@ def _install_runtime_context(config: dict, runtime_context: dict[str, Any]) -> N configurable = config.get("configurable") if isinstance(configurable, dict): configurable.pop(CONVERSATION_READER_CONTEXT_KEY, None) + configurable.pop(AGENT_RUNS_CONTEXT_KEY, None) existing_context = config.get("context") if isinstance(existing_context, dict): existing_context.setdefault("thread_id", runtime_context["thread_id"]) @@ -1148,6 +1157,7 @@ async def run_agent( task_store, extensions, ctx.conversation_reader, + ctx.agent_runs, thread_incarnation=thread_incarnation, ) # Bind every checkpoint produced by this run to the effective agent diff --git a/backend/packages/harness/deerflow/runtime/secret_context.py b/backend/packages/harness/deerflow/runtime/secret_context.py index 32e82ec79d..93aee30e09 100644 --- a/backend/packages/harness/deerflow/runtime/secret_context.py +++ b/backend/packages/harness/deerflow/runtime/secret_context.py @@ -16,6 +16,8 @@ from __future__ import annotations from typing import Any +from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY + from deerflow.trace_context import DEERFLOW_TRACE_METADATA_KEY # Reserved sub-key of the run context that holds request-scoped secrets supplied @@ -186,6 +188,7 @@ _SLASH_SKILL_ACTIVATION_RUN_KEY = "__slash_skill_activation_run" REDACTED_CONTEXT_KEYS = frozenset( { SECRETS_CONTEXT_KEY, + AGENT_RUNS_CONTEXT_KEY, ACTIVE_SECRETS_CONTEXT_KEY, _SLASH_SECRET_SOURCE_KEY, _SECRETS_BINDING_AUDIT_KEY, diff --git a/backend/packages/harness/pyproject.toml b/backend/packages/harness/pyproject.toml index 242ff903a3..bd3ef76a0e 100644 --- a/backend/packages/harness/pyproject.toml +++ b/backend/packages/harness/pyproject.toml @@ -13,7 +13,7 @@ dependencies = [ # the contract version it implements, extensions declare ranges. A range # here would let pip resolve a newer contract package than this harness # implements, making newer extensions look supported at runtime. - "deerflow-extension-api==0.2.4", + "deerflow-extension-api==0.2.5", "dotenv>=0.9.9", "exa-py>=1.0.0", "httpx>=0.28.0", diff --git a/backend/tests/test_extension_agent_runs.py b/backend/tests/test_extension_agent_runs.py new file mode 100644 index 0000000000..761314d46f --- /dev/null +++ b/backend/tests/test_extension_agent_runs.py @@ -0,0 +1,366 @@ +"""Host-bound run control: authorization, lifecycle and ordinary admission.""" + +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import pytest_asyncio +from deerflow_extension_api.agent_runs import AgentRunError, require_agent_runs, resolve_agent_runs +from fastapi import FastAPI, Request + +from app.gateway.auth.models import User +from app.gateway.authz import AuthContext +from app.gateway.extension_agent_runs import GatewayAgentRunsHost + + +@pytest.fixture +def host(monkeypatch): + user = User(email="alice@example.com") + app = FastAPI() + load_user = AsyncMock(return_value=user) + control = GatewayAgentRunsHost(app, load_user=load_user) + request = Request({"type": "http", "app": app, "headers": [], "state": {}}) + request.state.user = user + request.state.auth_source = "session" + request.state.auth = AuthContext(user, ["threads:write", "threads:read", "runs:create", "runs:read", "runs:cancel"]) + monkeypatch.setattr("app.gateway.extension_agent_runs.resolve_route_permissions", AsyncMock(return_value=list(request.state.auth.permissions))) + return control, request, load_user + + +def test_unsupported_host_and_credentials_fail_closed(host): + control, request, _ = host + assert resolve_agent_runs(request) is None + with pytest.raises(NotImplementedError): + require_agent_runs(request) + request.state.auth_source = "pat" + assert control.bind(request) is None + request.state.auth_source = "internal" + assert control.bind(request) is None + + +def test_unstamped_requests_receive_no_delegated_capability(host): + control, request, _ = host + assert control.bind(SimpleNamespace(state=SimpleNamespace())) is None + assert control.bind(SimpleNamespace(state=SimpleNamespace(user=request.state.user, auth_source="internal"))) is None + request.state.auth = SimpleNamespace(is_authenticated=False) + assert control.bind(request) is None + + +def test_binding_rejects_inconsistent_authenticated_identity(host): + control, request, _ = host + request.state.user = User(email="other@example.com") + with pytest.raises(PermissionError): + control.bind(request) + + +@pytest.mark.asyncio +async def test_handle_rechecks_user_and_shutdown_before_dispatch(host, monkeypatch): + control, request, load_user = host + runs = control.bind(request) + handler = AsyncMock(return_value=SimpleNamespace(thread_id="thread-a")) + monkeypatch.setattr("app.gateway.routers.threads.create_thread", handler) + assert await runs.create_thread() == "thread-a" + load_user.return_value = request.state.user.model_copy(update={"token_version": 1}) + with pytest.raises(AgentRunError, match="revoked"): + await runs.create_thread() + load_user.return_value = request.state.user + control.close() + with pytest.raises(AgentRunError, match="unavailable"): + await runs.create_thread() + assert handler.await_count == 1 + + +@pytest.mark.asyncio +async def test_permissions_are_intersected_with_original_grant_and_current_policy(host, monkeypatch): + control, request, _ = host + request.state.auth.permissions = ["runs:read"] + runs = control.bind(request) + request.state.auth.permissions.append("threads:write") + monkeypatch.setattr("app.gateway.extension_agent_runs.resolve_route_permissions", AsyncMock(return_value=["threads:write", "runs:read"])) + with pytest.raises(AgentRunError) as denied: + await runs.create_thread() + assert denied.value.status_code == 403 + monkeypatch.setattr("app.gateway.extension_agent_runs.resolve_route_permissions", AsyncMock(return_value=[])) + with pytest.raises(AgentRunError) as revoked: + await runs.get(thread_id="thread-a", run_id="run-a") + assert revoked.value.status_code == 403 + + +def test_runtime_handle_is_host_owned_redacted_and_released(): + from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY + + from deerflow.runtime.runs.worker import _build_runtime_context, _install_runtime_context, _release_run_scoped_references + from deerflow.runtime.secret_context import redact_config_secrets + + key = AGENT_RUNS_CONTEXT_KEY + assert key not in _build_runtime_context("thread", "run", {key: "forged"}) + handle = object() + runtime = _build_runtime_context("thread", "run", {key: "forged"}, agent_runs=handle) + config = {"context": {key: "forged"}, "configurable": {key: "forged"}} + _install_runtime_context(config, runtime) + assert config["context"][key] is handle + assert key not in config["configurable"] + assert key not in redact_config_secrets(config)["context"] + _release_run_scoped_references([config], runtime, None) + assert key not in runtime + assert key not in config["context"] + + +@pytest_asyncio.fixture +async def runtime(host, monkeypatch, tmp_path): + import asyncio + + from deerflow_extension_api.agent_runs import AGENT_RUNS_CONTEXT_KEY + from langchain_core.messages import AIMessage + from langgraph.checkpoint.memory import InMemorySaver + from langgraph.graph import MessagesState, StateGraph + from langgraph.runtime import Runtime + from langgraph.store.memory import InMemoryStore + from langgraph.types import interrupt + + from deerflow.config.app_config import AppConfig, reset_app_config, set_app_config + from deerflow.persistence.thread_meta.memory import MemoryThreadMetaStore + from deerflow.runtime import RunManager + from deerflow.runtime.events.store.memory import MemoryRunEventStore + from deerflow.runtime.runs.store.memory import MemoryRunStore + from deerflow.runtime.stream_bridge.memory import MemoryStreamBridge + + control, request, _ = host + state = request.app.state + state.agent_runs_host = control + state.checkpointer = InMemorySaver() + state.store = InMemoryStore() + state.thread_store = MemoryThreadMetaStore(state.store) + state.run_store = MemoryRunStore() + state.run_manager = RunManager(store=state.run_store) + state.run_event_store = MemoryRunEventStore() + state.stream_bridge = MemoryStreamBridge() + monkeypatch.setenv("DEER_FLOW_HOME", str(tmp_path)) + set_app_config(AppConfig.model_validate({"sandbox": {"use": "deerflow.sandbox.local:LocalSandboxProvider"}})) + entered = asyncio.Event() + handles = [] + + async def answer(state: MessagesState, runtime: Runtime): + handles.append(runtime.context.get(AGENT_RUNS_CONTEXT_KEY)) + prompt = state["messages"][-1].content + if prompt == "wait": + entered.set() + await asyncio.Event().wait() + if prompt == "approve": + prompt = interrupt({"question": "Continue?"}) + return {"messages": [AIMessage(content=f"answer:{prompt}")]} + + def factory(*, config): + builder = StateGraph(MessagesState) + builder.add_node("answer", answer) + builder.set_entry_point("answer") + builder.set_finish_point("answer") + return builder.compile() + + monkeypatch.setattr("app.gateway.services.resolve_agent_factory", lambda _: factory) + try: + yield control.bind(request), state, request, entered, handles + finally: + control.close() + await state.run_manager.shutdown() + reset_app_config() + + +@pytest.mark.asyncio +async def test_real_run_lifecycle_and_automatic_followup(runtime): + runs, state, request, entered, handles = runtime + thread_a = await runs.create_thread(thread_id="extension-a") + first = await runs.start(thread_id=thread_a, input={"messages": [{"role": "user", "content": "first"}]}, idempotency_key="team:first") + assert (await runs.wait(thread_id=thread_a, run_id=first.run_id, timeout=5)).status == "success" + retry = await runs.start(thread_id=thread_a, input={"messages": [{"role": "user", "content": "first"}]}, idempotency_key="team:first") + assert retry.run_id == first.run_id + output_a = await runs.get_state(thread_id=thread_a) + assert output_a["values"]["messages"][-1]["content"] == "answer:first" + assert handles and all(handle is not None for handle in handles) + # A service holding only the original capability can hand A's result to B. + thread_b = await runs.create_thread(thread_id="extension-b") + second = await runs.start(thread_id=thread_b, input={"messages": [{"role": "user", "content": output_a["values"]["messages"][-1]["content"]}]}) + assert (await runs.wait(thread_id=thread_b, run_id=second.run_id, timeout=5)).status == "success" + continued = await runs.start(thread_id=thread_a, input={"messages": [{"role": "user", "content": "next"}]}) + assert (await runs.wait(thread_id=thread_a, run_id=continued.run_id, timeout=5)).status == "success" + snapshot = await runs.get_state(thread_id=thread_a) + assert [message["content"] for message in snapshot["values"]["messages"]] == ["first", "answer:first", "next", "answer:next"] + stored = await state.thread_store.get(thread_a, user_id=str(request.state.user.id)) + assert stored["user_id"] == str(request.state.user.id) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("launcher", ["scheduled", "mcp"]) +async def test_internal_launchers_work_with_installed_host_without_delegation(runtime, launcher, monkeypatch): + from app.gateway.services import launch_mcp_task_notification_run, launch_scheduled_thread_run + + runs, state, request, _, handles = runtime + monkeypatch.setattr("app.gateway.services.get_local_provider", lambda: SimpleNamespace(get_user=AsyncMock(return_value=request.state.user))) + thread = await runs.create_thread(thread_id="internal-launch") + kwargs = dict(app=request.app, thread_id=thread, assistant_id="lead_agent", owner_user_id=str(request.state.user.id)) + if launcher == "scheduled": + result = await launch_scheduled_thread_run(**kwargs, prompt="scheduled", metadata={"scheduled_task_run_id": "occurrence-1"}) + else: + result = await launch_mcp_task_notification_run(**kwargs, task_id="task-1", dispatch_version=1, dispatch_attempt=1, event={"status": "completed"}) + assert (await runs.wait(thread_id=thread, run_id=result["run_id"], timeout=5)).status == "success" + assert handles == [None] + assert (await state.thread_store.get(thread, user_id=str(request.state.user.id)))["user_id"] == str(request.state.user.id) + + +@pytest.mark.asyncio +async def test_plugin_actions_isolate_same_local_idempotency_key(runtime): + import json + + from deerflow_extension_api.agent_runs import AGENT_RUNS_RESOLVER_KEY + from deerflow_extension_api.auth import EXTENSION_PRINCIPAL_RESOLVER_KEY, ExtensionPrincipal + from deerflow_extension_api.plugins import BackendAction, PluginContribution + + from app.gateway.routers.plugins import invoke_plugin_action + from deerflow.extensions.registry import ExtensionRegistry + + runs, state, request, _, _ = runtime + thread = await runs.create_thread(thread_id="plugin-keys") + + async def start(payload, context): + run = await context.agent_runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "same input"}]}, idempotency_key=payload["key"]) + return {"run_id": run.run_id} + + registry = ExtensionRegistry() + for namespace in ("community.first", "community.second"): + with registry.attributed_to(namespace): + registry.plugin(PluginContribution(namespace=namespace, title=namespace, enabled=True, backend=(BackendAction("start", start),))) + state.extensions = registry.build() + setattr(state, AGENT_RUNS_RESOLVER_KEY, state.agent_runs_host.bind) + setattr(state, EXTENSION_PRINCIPAL_RESOLVER_KEY, lambda _: ExtensionPrincipal(str(request.state.user.id))) + + async def invoke(namespace): + async def receive(): + return {"type": "http.request", "body": json.dumps({"key": "shared:operation"}).encode(), "more_body": False} + + admitted = Request({"type": "http", "app": request.app, "headers": [], "state": {"user": request.state.user, "auth": request.state.auth, "auth_source": "session"}}, receive) + return await invoke_plugin_action(admitted, namespace, "start") + + first = await invoke("community.first") + await runs.wait(thread_id=thread, run_id=first["run_id"], timeout=5) + second = await invoke("community.second") + assert second["run_id"] != first["run_id"] + await runs.wait(thread_id=thread, run_id=second["run_id"], timeout=5) + assert (await invoke("community.first"))["run_id"] == first["run_id"] + + +@pytest.mark.asyncio +async def test_plugin_scoping_preserves_key_validation_and_authorization(host): + control, request, load_user = host + scoped = control.bind(request).for_plugin("community.test") + assert scoped.for_plugin("community.test") is scoped + with pytest.raises(AgentRunError): + scoped.for_plugin("community.other") + for key in ("", " ", "x" * 201, 1): + with pytest.raises(AgentRunError) as invalid: + await scoped.start(thread_id="thread", input={}, idempotency_key=key) + assert invalid.value.status_code == 422 + load_user.return_value = None + with pytest.raises(AgentRunError) as revoked: + await scoped.get(thread_id="thread", run_id="run") + assert revoked.value.status_code == 403 + + +@pytest.mark.asyncio +async def test_wait_uses_capped_backoff_and_returns_terminal_status(host, monkeypatch): + from deerflow_extension_api.agent_runs import AgentRun + + control, request, _ = host + runs = control.bind(request) + runs.get = AsyncMock(side_effect=[AgentRun("thread", "run", "running")] * 6 + [AgentRun("thread", "run", "interrupted")]) + sleep = AsyncMock() + monkeypatch.setattr("app.gateway.extension_agent_runs.asyncio.sleep", sleep) + assert (await runs.wait(thread_id="thread", run_id="run")).status == "interrupted" + assert [call.args[0] for call in sleep.await_args_list] == [0.25, 0.5, 1, 2, 4, 4] + + +@pytest.mark.asyncio +async def test_real_interrupt_resume_conflict_and_cancel(runtime): + import asyncio + + runs, _, _, entered, _ = runtime + thread = await runs.create_thread(thread_id="interrupt-test") + first = await runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "approve"}]}) + assert (await runs.wait(thread_id=thread, run_id=first.run_id, timeout=5)).status == "success" + assert (await runs.get_state(thread_id=thread))["next"] == ["answer"] + resumed = await runs.resume(thread_id=thread, resume="yes") + assert (await runs.wait(thread_id=thread, run_id=resumed.run_id, timeout=5)).status == "success" + assert (await runs.get_state(thread_id=thread))["values"]["messages"][-1]["content"] == "answer:yes" + active = await runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "wait"}]}) + await asyncio.wait_for(entered.wait(), 5) + with pytest.raises(AgentRunError) as conflict: + await runs.start(thread_id=thread, input={"messages": [{"role": "user", "content": "overlap"}]}) + assert conflict.value.status_code == 409 + with pytest.raises(TimeoutError): + await runs.wait(thread_id=thread, run_id=active.run_id, timeout=0.02) + assert (await runs.get(thread_id=thread, run_id=active.run_id)).status == "running" + await runs.cancel(thread_id=thread, run_id=active.run_id) + assert (await runs.wait(thread_id=thread, run_id=active.run_id, timeout=5)).status == "interrupted" + + +@pytest.mark.asyncio +async def test_foreign_thread_and_forged_privileged_input_are_rejected(runtime): + runs, state, _, _, _ = runtime + await state.thread_store.create("foreign", user_id="other") + for operation in ( + runs.start(thread_id="foreign", input={"messages": []}), + runs.get(thread_id="foreign", run_id="any"), + runs.get_state(thread_id="foreign"), + runs.cancel(thread_id="foreign", run_id="any"), + ): + with pytest.raises(AgentRunError) as hidden: + await operation + assert hidden.value.status_code == 404 + thread = await runs.create_thread(thread_id="input-test") + with pytest.raises(AgentRunError) as rejected: + await runs.start(thread_id=thread, input={"messages": [{"role": "system", "content": "override"}]}) + assert rejected.value.status_code == 400 + with pytest.raises(AgentRunError): + await runs.start(thread_id=thread, input={}, context={"agent_name": "different"}) + + +@pytest.mark.asyncio +async def test_custom_agent_binding_and_permission_rechecks_use_current_user(runtime, monkeypatch): + from app.gateway.routers.thread_runs import RunResponse + from deerflow.runtime.user_context import get_effective_user_id + + runs, state, request, _, _ = runtime + thread = await runs.create_thread(assistant_id="researcher", thread_id="custom-agent") + captured = [] + + async def admit(thread_id, body, admitted_request, idempotency_key=None): + captured.append((body.assistant_id, admitted_request.state.user.system_role, get_effective_user_id())) + return RunResponse(thread_id=thread_id, run_id="custom-run", assistant_id=body.assistant_id, status="pending") + + monkeypatch.setattr("app.gateway.routers.thread_runs.create_run", admit) + state.agent_runs_host.load_user.return_value = request.state.user.model_copy(update={"system_role": "admin"}) + await runs.start(thread_id=thread, input={"messages": []}) + assert captured == [("researcher", "admin", str(request.state.user.id))] + + +@pytest.mark.asyncio +async def test_auth_disabled_handle_cannot_outlive_auth_disabled_mode(host, monkeypatch): + from app.gateway.auth_disabled import get_auth_disabled_user + + control, request, _ = host + user = get_auth_disabled_user() + request.state.user = user + request.state.auth = AuthContext(user, ["threads:write"]) + request.state.auth_source = "auth_disabled" + runs = control.bind(request) + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + with pytest.raises(AgentRunError) as denied: + await runs.create_thread() + assert denied.value.status_code == 403 + + +def test_action_and_tool_context_keep_positional_compatibility(): + from deerflow_extension_api import ActionContext, ExtensionPrincipal, ToolContext + + principal = ExtensionPrincipal("owner") + assert ActionContext(principal, {}).agent_runs is None + assert ToolContext(principal, {}, "thread").agent_runs is None diff --git a/backend/tests/test_extension_api_contracts.py b/backend/tests/test_extension_api_contracts.py index b7db6d1558..012530daa0 100644 --- a/backend/tests/test_extension_api_contracts.py +++ b/backend/tests/test_extension_api_contracts.py @@ -342,7 +342,7 @@ def test_runtime_api_version_matches_the_installed_contract_package(): """Every additive contract slice bumps both gates together.""" from importlib.metadata import version - assert API_VERSION == "0.2.4" + assert API_VERSION == "0.2.5" assert API_VERSION == version("deerflow-extension-api") diff --git a/backend/tests/test_plugin_tools.py b/backend/tests/test_plugin_tools.py index e44b584067..6768632356 100644 --- a/backend/tests/test_plugin_tools.py +++ b/backend/tests/test_plugin_tools.py @@ -142,3 +142,25 @@ def test_assembly_keeps_ordinary_and_unaffected_plugin_tools_on_collision(instal duplicate_snapshot = replace(loaded, plugins=loaded.plugins + loaded.plugins) with pytest.raises(ValueError, match="collision"): assembly.get_available_tools(app_config=config, extensions=duplicate_snapshot, include_mcp=False, include_upload_tool=False) + + +@pytest.mark.asyncio +async def test_model_tool_receives_run_control_from_host_runtime(installed): + from unittest.mock import Mock + + from deerflow_extension_api import AGENT_RUNS_CONTEXT_KEY + + loaded, plugin, calls = installed + (tool,) = build_plugin_tools(loaded) + graph = StateGraph(MessagesState) + graph.add_node("tools", ToolNode([tool])) + graph.add_edge(START, "tools") + graph.add_edge("tools", END) + scoped = object() + handle = SimpleNamespace(for_plugin=Mock(return_value=scoped)) + await graph.compile().ainvoke( + {"messages": [AIMessage(content="", tool_calls=[{"id": "call", "name": tool.name, "args": {"query": "hello"}}])]}, + context={"user_id": "alice", "thread_id": "thread", AGENT_RUNS_CONTEXT_KEY: handle}, + ) + handle.for_plugin.assert_called_once_with(plugin.namespace) + assert calls[0].agent_runs is scoped diff --git a/backend/uv.lock b/backend/uv.lock index 2fb22e799c..c78345270a 100644 --- a/backend/uv.lock +++ b/backend/uv.lock @@ -803,7 +803,7 @@ wheels = [ [[package]] name = "deer-flow" -version = "2.2.0dev" +version = "2.2.0.dev0" source = { virtual = "." } dependencies = [ { name = "bcrypt" }, @@ -920,7 +920,7 @@ extensions = [] [[package]] name = "deerflow-extension-api" -version = "0.2.4" +version = "0.2.5" source = { editable = "packages/extension-api" } [[package]] diff --git a/docs/full-stack-plugins.md b/docs/full-stack-plugins.md index 694c8752f4..e9c6cf1cc1 100644 --- a/docs/full-stack-plugins.md +++ b/docs/full-stack-plugins.md @@ -3,7 +3,12 @@ A deployment-installed Python extension can register a `PluginContribution` with optional browser code, authenticated backend actions and model tools. This extends the existing `install(registry, config)` workflow. MCP and Skills keep their existing -APIs and lifecycles. Public contracts live in `deerflow_extension_api` (0.2.4). +APIs and lifecycles. Public contracts live in `deerflow_extension_api` (0.2.5). + +Backend actions and model tools may use the optional host-bound +`context.agent_runs` capability to create, continue, inspect, resume, and cancel +full Agent runs. See [Agent run control](../backend/docs/extension-agent-runs.md) +for authorization, service-held handle lifetime, and handoff examples. The browser contribution API is experimental. `BrowserModule(code=...)` remains the self-contained transport; `BrowserAssets(root=...)` adds manifest-listed resources