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

@@ -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)
 

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),

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)

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:

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.

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",

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

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)

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

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.

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)

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

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,

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",

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

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")
 
 

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

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]]

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