董宏凯andCursor
f82363d81f
- Backend: action_coordinator, capability_snapshot, event_normalizer, tool_executor updates - Backend: approved_tool_executor with write guard, action leases, migration script - Frontend: maf-roundtable module (API client, event timeline, approval cards, fact flow panel) - Frontend: MafRoutes, MafSessionsPage, MafRunDetailPage - Tests: action_leases, migration_phase5, phase5_entry_e2e, real_write_e2e/http_e2e - Docs: update implementation plan Co-authored-by: Cursor <cursoragent@cursor.com>
338 lines
13 KiB
Python
338 lines
13 KiB
Python
"""Controlled DeerFlow runtime invocation client for MAF roundtable.
|
|
|
|
All invocations require an explicit ``ExecutionPrincipal`` — the server-generated
|
|
identity derived ONLY from the authenticated MAF session owner. No raw tokens,
|
|
cookies, or credentials ever pass into events, checkpoints, or prompts.
|
|
|
|
Two invocation modes (the adapter uses mode 2):
|
|
1. Direct checkpointer access for thread creation and message reads.
|
|
2. In-process ``start_run`` via a typed internal request that carries the
|
|
principal's user identity (never ``user=None``).
|
|
|
|
Security invariants:
|
|
- ExecutionPrincipal is constructed ONLY from the MAF session's user_id.
|
|
- The internal request object explicitly carries the principal's user so
|
|
downstream code sees a real identity, not a None placeholder.
|
|
- Agent access is enforced server-side (enforce_access=True).
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import uuid
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from langgraph.checkpoint.base import empty_checkpoint
|
|
|
|
from deerflow.runtime.user_context import reset_current_user, set_current_user
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _serialize_tool_result(result: Any) -> Any:
|
|
"""Convert a tool result to a JSON-safe dict/list/str/int/float/bool/None."""
|
|
if result is None:
|
|
return None
|
|
if isinstance(result, (str, int, float, bool)):
|
|
return result
|
|
if isinstance(result, (list, tuple)):
|
|
return [_serialize_tool_result(v) for v in result]
|
|
if isinstance(result, dict):
|
|
return {str(k): _serialize_tool_result(v) for k, v in result.items()}
|
|
return str(result)[:1000]
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExecutionPrincipal:
|
|
"""Server-generated execution identity for a MAF seat run.
|
|
|
|
Constructed ONLY from the authenticated MAF session owner. Never from
|
|
request body, query params, or client-supplied data.
|
|
"""
|
|
|
|
user_id: str
|
|
session_id: str
|
|
run_id: str
|
|
|
|
def __post_init__(self) -> None:
|
|
if not self.user_id or not self.user_id.strip():
|
|
raise ValueError("ExecutionPrincipal requires a non-empty user_id")
|
|
|
|
|
|
class _PrincipalUser:
|
|
"""Minimal CurrentUser protocol implementation backed by ExecutionPrincipal."""
|
|
|
|
def __init__(self, principal: ExecutionPrincipal) -> None:
|
|
self.id = principal.user_id
|
|
|
|
|
|
class _InternalState:
|
|
"""Typed state container for the internal run request.
|
|
|
|
Carries the principal's user identity explicitly — never None — so
|
|
downstream access checks and run metadata attribution work correctly.
|
|
"""
|
|
|
|
def __init__(self, principal: ExecutionPrincipal) -> None:
|
|
self.user = _PrincipalUser(principal)
|
|
self.auth = None
|
|
# This marker is set only on the server-created request used by the
|
|
# MAF dispatcher. Gateway code uses it to distinguish trusted MAF
|
|
# execution context from user-supplied run configuration.
|
|
self.maf_internal = True
|
|
|
|
|
|
@dataclass
|
|
class _InternalRunRequest:
|
|
"""Typed internal request for in-process agent run invocation.
|
|
|
|
Replaces the former ``SimpleNamespace(app=..., state=SimpleNamespace(user=None))``
|
|
pattern. The state explicitly carries the principal's user identity.
|
|
"""
|
|
|
|
app: Any
|
|
state: _InternalState
|
|
|
|
@classmethod
|
|
def from_principal(cls, app: Any, principal: ExecutionPrincipal) -> _InternalRunRequest:
|
|
return cls(app=app, state=_InternalState(principal))
|
|
|
|
|
|
class DeerFlowRuntimeClient:
|
|
"""Controlled entry point for invoking the DeerFlow Agent Runtime from MAF.
|
|
|
|
All calls require an ExecutionPrincipal. The client:
|
|
- Sets the user ContextVar before each invocation.
|
|
- Creates threads with proper user attribution.
|
|
- Invokes start_run with server-generated identity.
|
|
- Never passes raw tokens or credentials downstream.
|
|
"""
|
|
|
|
def __init__(self, app: Any) -> None:
|
|
self._app = app
|
|
|
|
@property
|
|
def app(self) -> Any:
|
|
return self._app
|
|
|
|
async def create_thread(self, principal: ExecutionPrincipal) -> str:
|
|
"""Create a DeerFlow thread owned by the principal's user.
|
|
|
|
The user ContextVar is set so thread_store.create records the correct
|
|
user_id in thread_meta.
|
|
"""
|
|
token = set_current_user(_PrincipalUser(principal))
|
|
try:
|
|
checkpointer = self._app.state.checkpointer
|
|
thread_store = self._app.state.thread_store
|
|
thread_id = str(uuid.uuid4())
|
|
await thread_store.create(
|
|
thread_id,
|
|
assistant_id=None,
|
|
metadata={
|
|
"thread_type": "maf_roundtable",
|
|
"system": True,
|
|
"maf_session_id": principal.session_id,
|
|
"maf_run_id": principal.run_id,
|
|
},
|
|
)
|
|
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
|
await checkpointer.aput(
|
|
config,
|
|
empty_checkpoint(),
|
|
{"step": -1, "source": "input", "writes": None, "parents": {}, "created_at": datetime.now(UTC).isoformat()},
|
|
{},
|
|
)
|
|
return thread_id
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
async def start_agent_run(
|
|
self,
|
|
principal: ExecutionPrincipal,
|
|
*,
|
|
thread_id: str,
|
|
agent_id: str,
|
|
message: str,
|
|
model_name: str | None = None,
|
|
reasoning_effort: str | None = None,
|
|
thinking_enabled: bool = True,
|
|
subagent_enabled: bool = False,
|
|
excluded_tools: list[str] | None = None,
|
|
memory_recall_disabled: bool | None = None,
|
|
memory_injection_enabled: bool | None = None,
|
|
turn_index: int = 0,
|
|
capability_snapshot: dict[str, Any] | None = None,
|
|
write_policy: str = "confirm_high_risk",
|
|
approved_tool: str | None = None,
|
|
approved_action_id: str | None = None,
|
|
) -> Any:
|
|
"""Start a DeerFlow agent run under the principal's identity.
|
|
|
|
Returns the RunRecord (with .task and .run_id). The caller must await
|
|
the task and handle exceptions.
|
|
|
|
WriteGuard activation: when ``write_policy != "auto"`` the run context
|
|
carries ``maf_run_id`` (the activation flag the WriteGuard middleware
|
|
checks) plus the per-run data the interceptor needs to record a
|
|
pending action. ``excluded_tools`` here MUST only contain tools the
|
|
user explicitly disabled — write tools are no longer excluded (the
|
|
middleware intercepts them instead, keeping them visible to the model).
|
|
"""
|
|
from app.gateway.routers.thread_runs import RunCreateRequest
|
|
from app.gateway.services import RUN_SCOPE_BACKGROUND_CHILD, start_run
|
|
|
|
context: dict[str, Any] = {
|
|
"agent_name": agent_id,
|
|
"model_name": model_name or "",
|
|
"mode": "pro",
|
|
"reasoning_effort": reasoning_effort or "medium",
|
|
"thinking_enabled": thinking_enabled,
|
|
"is_plan_mode": False,
|
|
"subagent_enabled": subagent_enabled,
|
|
"thread_id": thread_id,
|
|
}
|
|
if memory_recall_disabled is not None:
|
|
context["memory_recall_disabled"] = memory_recall_disabled
|
|
if memory_injection_enabled is not None:
|
|
context["memory_injection_enabled"] = memory_injection_enabled
|
|
|
|
# WriteGuard activation + per-run data (server-generated identity only).
|
|
# ``maf_run_id`` is the flag the WriteGuard middleware checks to decide
|
|
# whether to intercept; the rest is the data the interceptor reads.
|
|
context["maf_run_id"] = principal.run_id
|
|
context["maf_session_id"] = principal.session_id
|
|
context["maf_user_id"] = principal.user_id
|
|
context["agent_id"] = agent_id
|
|
context["turn_index"] = turn_index
|
|
context["write_policy"] = write_policy
|
|
if capability_snapshot is not None:
|
|
context["capability_snapshot"] = capability_snapshot
|
|
# Post-approval deterministic execution: when set, the WriteGuard
|
|
# middleware passes this single tool through without interception.
|
|
# ``maf_approved_action_id`` binds the passthrough to a specific
|
|
# approved action — the middleware validates both tool name AND
|
|
# action_id so a model calling another tool with the same name is
|
|
# still intercepted (defense-in-depth when combined with direct
|
|
# tool execution that bypasses the LLM).
|
|
if approved_tool:
|
|
context["maf_approved_tool"] = approved_tool
|
|
if approved_action_id:
|
|
context["maf_approved_action_id"] = approved_action_id
|
|
|
|
body = RunCreateRequest(
|
|
assistant_id="lead_agent",
|
|
input={"messages": [{"type": "human", "content": [{"type": "text", "text": message}], "additional_kwargs": {}}]},
|
|
metadata={"run_scope": RUN_SCOPE_BACKGROUND_CHILD, "maf_run_id": principal.run_id},
|
|
config={"recursion_limit": 1000},
|
|
context=context,
|
|
excluded_tools=excluded_tools or [],
|
|
on_disconnect="continue",
|
|
multitask_strategy="reject",
|
|
)
|
|
# Typed internal request carrying the principal's identity — never user=None.
|
|
request = _InternalRunRequest.from_principal(self._app, principal)
|
|
|
|
token = set_current_user(_PrincipalUser(principal))
|
|
try:
|
|
record = await start_run(
|
|
body,
|
|
thread_id,
|
|
request,
|
|
run_scope=RUN_SCOPE_BACKGROUND_CHILD,
|
|
)
|
|
return record
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
async def read_thread_messages(self, principal: ExecutionPrincipal, thread_id: str) -> list[dict[str, Any]]:
|
|
"""Read the latest checkpoint messages from a thread."""
|
|
from deerflow.runtime.serialization import serialize_channel_values
|
|
|
|
token = set_current_user(_PrincipalUser(principal))
|
|
try:
|
|
checkpointer = self._app.state.checkpointer
|
|
config = {"configurable": {"thread_id": thread_id, "checkpoint_ns": ""}}
|
|
ct = await checkpointer.aget_tuple(config)
|
|
if ct is None:
|
|
return []
|
|
checkpoint = getattr(ct, "checkpoint", {}) or {}
|
|
values = serialize_channel_values(checkpoint.get("channel_values", {}) or {})
|
|
return values.get("messages") or []
|
|
finally:
|
|
reset_current_user(token)
|
|
|
|
async def execute_tool_directly(
|
|
self,
|
|
*,
|
|
thread_id: str,
|
|
agent_id: str,
|
|
tool_name: str,
|
|
params: dict[str, Any],
|
|
principal: ExecutionPrincipal,
|
|
turn_index: int,
|
|
capability_snapshot: dict[str, Any] | None,
|
|
write_policy: str,
|
|
approved_action_id: str,
|
|
action_record: dict[str, Any] | None,
|
|
model_name: str | None = None,
|
|
excluded_tools: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Execute a DeerFlow tool directly without LLM involvement.
|
|
|
|
For post-approval deterministic execution this delegates to the harness
|
|
``execute_approved_tool``, which runs the call through the **real**
|
|
LangGraph ``ToolNode`` / ``ToolRuntime`` / middleware chain (state
|
|
recovered from the thread checkpoint), not a faked runtime. No raw
|
|
params enter any prompt, checkpoint, or model context.
|
|
|
|
Returns ``{"status": "succeeded", "result": ...}`` or
|
|
``{"status": "failed", "error": "..."}``. A security rejection
|
|
(unauthorized / disabled / nonexistent tool, context mismatch) is a
|
|
``failed`` outcome and the tool executes 0 times.
|
|
"""
|
|
from deerflow.config import get_app_config
|
|
from deerflow.tools.approved_tool_executor import execute_approved_tool
|
|
|
|
app_config = get_app_config()
|
|
checkpointer = getattr(self._app.state, "checkpointer", None)
|
|
store = getattr(self._app.state, "store", None)
|
|
|
|
return await execute_approved_tool(
|
|
app_config=app_config,
|
|
checkpointer=checkpointer,
|
|
store=store,
|
|
thread_id=thread_id,
|
|
user_id=principal.user_id,
|
|
session_id=principal.session_id,
|
|
run_id=principal.run_id,
|
|
agent_id=agent_id,
|
|
turn_index=turn_index,
|
|
model_name=model_name,
|
|
excluded_tools=excluded_tools,
|
|
capability_snapshot=capability_snapshot,
|
|
write_policy=write_policy,
|
|
tool_name=tool_name,
|
|
params=params,
|
|
approved_action_id=approved_action_id,
|
|
action_record=action_record,
|
|
)
|
|
|
|
async def cancel_run(self, deerflow_run_id: str) -> bool:
|
|
"""Cancel a DeerFlow run by its run_id."""
|
|
run_manager = getattr(self._app.state, "run_manager", None)
|
|
if run_manager is None:
|
|
return False
|
|
try:
|
|
await run_manager.cancel(deerflow_run_id, action="interrupt")
|
|
return True
|
|
except Exception as exc: # noqa: BLE001
|
|
# Log only the exception type — never the raw message/traceback,
|
|
# which may carry params or credentials.
|
|
logger.warning(
|
|
"cancel DeerFlow run %s failed (%s)", deerflow_run_id, type(exc).__name__
|
|
)
|
|
return False
|