686 lines
30 KiB
Python
686 lines
30 KiB
Python
"""Deterministic, server-owned contracts for MAF Stage 3 dispatch.
|
||
|
||
The manager may describe work, but it does not get to choose arbitrary agent
|
||
ids, threads, tools, or capabilities. Routing is resolved from the trusted
|
||
MAF deployment configuration and the small, validated task-context allowlist.
|
||
This module is intentionally independent of the MAF SDK so the policy can be
|
||
tested with a weak/fake model and before any workflow is launched.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
from typing import Any, Literal
|
||
|
||
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
||
|
||
from deerflow.config.maf_roundtable_config import MafMode
|
||
|
||
MafIntentType = Literal[
|
||
"direct_answer",
|
||
"clarification",
|
||
"single_seat",
|
||
"parallel_consultation",
|
||
"chain_consultation",
|
||
"tool_write",
|
||
]
|
||
MafExecutionMode = Literal["manager_only", "single", "parallel", "chain", "group_chat"]
|
||
MafDispatchSource = Literal["server", "manager", "deterministic_fallback"]
|
||
|
||
|
||
class DispatchConfigurationError(RuntimeError):
|
||
"""The deployment has not supplied a complete trusted route."""
|
||
|
||
|
||
class InvalidDispatchPlanError(ValueError):
|
||
"""A proposed or constructed plan violates the dispatch contract."""
|
||
|
||
|
||
class MafDispatchCandidateSet(BaseModel):
|
||
"""Trusted route supplied to the manager before it chooses an intent."""
|
||
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
manager_agent_id: str = Field(min_length=1, max_length=64)
|
||
candidate_seat_agent_ids: list[str] = Field(default_factory=list, max_length=64)
|
||
mode: MafMode = "structured"
|
||
business_code: str | None = Field(default=None, max_length=255)
|
||
max_rounds: int = Field(ge=1, le=64)
|
||
|
||
|
||
class MafIntentDecision(BaseModel):
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
intent_type: MafIntentType
|
||
confidence: float = Field(ge=0.0, le=1.0)
|
||
reason_code: str = Field(min_length=1, max_length=64)
|
||
clarification_prompt: str | None = Field(default=None, max_length=1000)
|
||
|
||
@model_validator(mode="after")
|
||
def _clarification_shape(self) -> MafIntentDecision:
|
||
if self.intent_type == "clarification" and not self.clarification_prompt:
|
||
raise ValueError("clarification intent requires clarification_prompt")
|
||
if self.intent_type != "clarification" and self.clarification_prompt is not None:
|
||
raise ValueError("only clarification intent may include clarification_prompt")
|
||
return self
|
||
|
||
|
||
class MafDispatchAssignment(BaseModel):
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
agent_id: str = Field(min_length=1, max_length=64)
|
||
role: Literal["seat"] = "seat"
|
||
sort_order: int = Field(ge=1, le=64)
|
||
task_goal: str = Field(default="Complete the assigned analysis", min_length=1, max_length=2000)
|
||
delivery_requirements: list[str] = Field(default_factory=list, max_length=16)
|
||
|
||
|
||
class MafManagerPlanAssignment(BaseModel):
|
||
"""The only per-seat shape accepted from the manager model."""
|
||
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
agent_id: str = Field(min_length=1, max_length=64)
|
||
task_goal: str = Field(min_length=1, max_length=2000)
|
||
delivery_requirements: list[str] = Field(min_length=1, max_length=16)
|
||
order: int = Field(ge=1, le=64)
|
||
|
||
|
||
class MafManagerPlanOutput(BaseModel):
|
||
"""Strict model-facing dispatch schema.
|
||
|
||
This model deliberately contains no thread, tool, capability, or user
|
||
supplied routing fields. The server validates the IDs again after this
|
||
schema passes.
|
||
"""
|
||
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
intent_type: MafIntentType
|
||
selected_seat_agent_ids: list[str] = Field(default_factory=list, max_length=64)
|
||
assignments: list[MafManagerPlanAssignment] = Field(default_factory=list, max_length=64)
|
||
execution_mode: MafExecutionMode
|
||
convergence_path: list[str] = Field(min_length=1, max_length=16)
|
||
reason: str = Field(min_length=1, max_length=1000)
|
||
clarification_prompt: str | None = Field(default=None, min_length=1, max_length=1000)
|
||
|
||
@model_validator(mode="after")
|
||
def _validate_manager_shape(self) -> MafManagerPlanOutput:
|
||
selected = self.selected_seat_agent_ids
|
||
if len(set(selected)) != len(selected):
|
||
raise ValueError("manager plan contains duplicate seats")
|
||
assignment_ids = [item.agent_id for item in self.assignments]
|
||
if assignment_ids != selected:
|
||
raise ValueError("manager assignments must exactly match selected seats")
|
||
if [item.order for item in self.assignments] != list(range(1, len(selected) + 1)):
|
||
raise ValueError("manager assignment order must be contiguous")
|
||
|
||
required_mode: MafExecutionMode
|
||
if self.intent_type == "clarification":
|
||
if selected or self.assignments or self.execution_mode != "manager_only":
|
||
raise ValueError("clarification cannot select expert seats")
|
||
if not self.clarification_prompt:
|
||
raise ValueError("clarification requires clarification_prompt")
|
||
return self
|
||
if self.clarification_prompt is not None:
|
||
raise ValueError("only clarification may include clarification_prompt")
|
||
if self.intent_type == "direct_answer":
|
||
required_mode = "manager_only"
|
||
if selected:
|
||
raise ValueError("direct_answer cannot select expert seats")
|
||
elif self.intent_type in {"single_seat", "tool_write"}:
|
||
required_mode = "single"
|
||
if len(selected) != 1:
|
||
raise ValueError("single-seat intents require exactly one seat")
|
||
elif self.intent_type == "parallel_consultation":
|
||
required_mode = "parallel"
|
||
if not selected:
|
||
raise ValueError("parallel consultation requires a seat")
|
||
if len(selected) < 2:
|
||
raise ValueError("parallel consultation requires at least two seats")
|
||
elif self.intent_type == "chain_consultation":
|
||
required_mode = "chain"
|
||
if not selected:
|
||
raise ValueError("chain consultation requires a seat")
|
||
else:
|
||
raise ValueError("clarification cannot be dispatched")
|
||
allowed_modes = {required_mode}
|
||
if self.execution_mode == "group_chat" and required_mode != "manager_only":
|
||
allowed_modes.add("group_chat")
|
||
if self.execution_mode not in allowed_modes:
|
||
raise ValueError(
|
||
f"execution_mode {self.execution_mode!r} does not match {required_mode!r}"
|
||
)
|
||
return self
|
||
|
||
|
||
class MafDispatchPlan(BaseModel):
|
||
"""Validated plan containing only server-selected participants."""
|
||
|
||
model_config = ConfigDict(extra="forbid")
|
||
|
||
manager_agent_id: str = Field(min_length=1, max_length=64)
|
||
seat_agent_ids: list[str] = Field(default_factory=list, max_length=64)
|
||
assignments: list[MafDispatchAssignment] = Field(default_factory=list, max_length=64)
|
||
mode: MafMode = "structured"
|
||
intent_type: MafIntentType
|
||
business_code: str | None = Field(default=None, max_length=255)
|
||
max_rounds: int = Field(ge=1, le=64)
|
||
candidate_seat_agent_ids: list[str] = Field(default_factory=list, max_length=64)
|
||
execution_mode: MafExecutionMode = "parallel"
|
||
convergence_path: list[str] = Field(default_factory=list, max_length=16)
|
||
reason: str = Field(default="server-owned route", min_length=1, max_length=1000)
|
||
source: MafDispatchSource = "server"
|
||
|
||
@model_validator(mode="after")
|
||
def _validate_shape(self) -> MafDispatchPlan:
|
||
if self.manager_agent_id in self.seat_agent_ids:
|
||
raise ValueError("manager agent cannot also be a seat")
|
||
if len(set(self.seat_agent_ids)) != len(self.seat_agent_ids):
|
||
raise ValueError("duplicate seat agent")
|
||
if len(set(self.candidate_seat_agent_ids)) != len(self.candidate_seat_agent_ids):
|
||
raise ValueError("duplicate candidate seat agent")
|
||
if self.candidate_seat_agent_ids and any(
|
||
agent_id not in self.candidate_seat_agent_ids for agent_id in self.seat_agent_ids
|
||
):
|
||
raise ValueError("selected seat is outside candidate seats")
|
||
assignment_ids = [assignment.agent_id for assignment in self.assignments]
|
||
if assignment_ids != self.seat_agent_ids:
|
||
raise ValueError("assignments must exactly match seat_agent_ids")
|
||
if [a.sort_order for a in self.assignments] != list(range(1, len(self.assignments) + 1)):
|
||
raise ValueError("assignments must have contiguous sort_order")
|
||
if self.intent_type == "clarification":
|
||
raise ValueError("clarification cannot be dispatched")
|
||
if self.intent_type == "direct_answer":
|
||
if self.seat_agent_ids or self.execution_mode != "manager_only":
|
||
raise ValueError("direct_answer must use manager_only with no seats")
|
||
elif self.intent_type in {"single_seat", "tool_write"}:
|
||
if len(self.seat_agent_ids) != 1 or self.execution_mode not in {"single", "group_chat"}:
|
||
raise ValueError("single-seat intents require one seat and single mode")
|
||
elif self.intent_type == "parallel_consultation":
|
||
if not self.seat_agent_ids or self.execution_mode not in {"parallel", "group_chat"}:
|
||
raise ValueError("parallel consultation requires at least one configured seat")
|
||
if len(self.seat_agent_ids) < 2:
|
||
raise ValueError("parallel consultation requires at least two seats")
|
||
elif self.intent_type == "chain_consultation":
|
||
if not self.seat_agent_ids or self.execution_mode not in {"chain", "group_chat"}:
|
||
raise ValueError("chain consultation requires at least one configured seat")
|
||
if self.execution_mode == "group_chat" and self.mode != "group_chat":
|
||
raise ValueError("group_chat execution requires a group_chat route")
|
||
if self.intent_type != "direct_answer" and not self.seat_agent_ids:
|
||
raise ValueError("this intent requires at least one configured seat")
|
||
return self
|
||
|
||
|
||
def resolve_intent(content: str, task_context: dict[str, Any] | None = None) -> MafIntentDecision:
|
||
"""Resolve the common Stage 3 intents deterministically.
|
||
|
||
A production model can later supply a *description* for the intent, but
|
||
this policy remains the final validator. The fallback deliberately uses
|
||
conservative keywords and never extracts agent/tool ids from user text.
|
||
"""
|
||
|
||
text = " ".join(str(content or "").strip().split())
|
||
lowered = text.casefold()
|
||
if len(text) < 4 or lowered in {"?", "？", "不清楚", "不明确", "再说清楚"}:
|
||
return MafIntentDecision(
|
||
intent_type="clarification",
|
||
confidence=0.99,
|
||
reason_code="insufficient_task_detail",
|
||
clarification_prompt="请补充希望解决的具体问题、目标和必要的业务背景。",
|
||
)
|
||
|
||
write_markers = (
|
||
"写入", "发布", "创建工单", "更新任务", "提交", "删除", "write", "publish", "create ticket",
|
||
)
|
||
chain_markers = ("业务链", "链式", "逐阶段", "先…再", "先...再", "chain")
|
||
parallel_markers = ("多专家", "多角度", "并行", "同时分析", "比较", "parallel", "multi-agent")
|
||
single_markers = ("请某个专家", "单个席位", "单专家", "single seat", "指定专家")
|
||
|
||
if any(marker in lowered for marker in write_markers):
|
||
intent_type: MafIntentType = "tool_write"
|
||
reason_code = "write_request"
|
||
elif any(marker in lowered for marker in chain_markers):
|
||
intent_type = "chain_consultation"
|
||
reason_code = "chain_request"
|
||
elif any(marker in lowered for marker in parallel_markers):
|
||
intent_type = "parallel_consultation"
|
||
reason_code = "parallel_request"
|
||
elif any(marker in lowered for marker in single_markers):
|
||
intent_type = "single_seat"
|
||
reason_code = "single_seat_request"
|
||
else:
|
||
intent_type = "direct_answer"
|
||
reason_code = "default_direct_answer"
|
||
|
||
return MafIntentDecision(
|
||
intent_type=intent_type,
|
||
confidence=0.72 if task_context else 0.68,
|
||
reason_code=reason_code,
|
||
)
|
||
|
||
|
||
def _read(value: Any, name: str, default: Any = None) -> Any:
|
||
if isinstance(value, dict):
|
||
return value.get(name, default)
|
||
return getattr(value, name, default)
|
||
|
||
|
||
class MafDispatchCatalog:
|
||
"""Resolve routes from deployment config, never from a request body."""
|
||
|
||
def __init__(self, maf_config: Any) -> None:
|
||
self._config = maf_config
|
||
|
||
def readiness(self) -> tuple[bool, str]:
|
||
"""Return non-sensitive deployment readiness for the conversation UI.
|
||
|
||
A manager-only route is sufficient for a safe direct answer. Expert
|
||
routes are checked separately when a request selects a business chain
|
||
and when the trusted capability snapshot is built for the user.
|
||
"""
|
||
manager = str(_read(self._config, "default_manager_agent_id", "") or "").strip()
|
||
routes = _read(self._config, "business_routes", {}) or {}
|
||
if manager:
|
||
return True, "default_manager_configured"
|
||
for business_code, route in routes.items():
|
||
route_manager = str(_read(route, "manager_agent_id", "") or "").strip()
|
||
if route_manager:
|
||
return True, f"business_route_configured:{business_code}"
|
||
return False, "no_trusted_manager_route"
|
||
|
||
@property
|
||
def conversation_dispatch_ready(self) -> bool:
|
||
return self.readiness()[0]
|
||
|
||
def _route(self, business_code: str | None) -> tuple[str | None, list[str], MafMode | None]:
|
||
routes = _read(self._config, "business_routes", {}) or {}
|
||
if business_code:
|
||
route = routes.get(business_code)
|
||
if route is None:
|
||
# A supplied business code must match an explicitly configured
|
||
# route. Falling back here would silently cross business
|
||
# boundaries.
|
||
return None, [], None
|
||
return (
|
||
_read(route, "manager_agent_id"),
|
||
list(_read(route, "seat_agent_ids", []) or []),
|
||
_read(route, "mode"),
|
||
)
|
||
return (
|
||
_read(self._config, "default_manager_agent_id"),
|
||
list(_read(self._config, "default_seat_agent_ids", []) or []),
|
||
None,
|
||
)
|
||
|
||
def build_candidate_set(
|
||
self, *, task_context: dict[str, Any] | None
|
||
) -> MafDispatchCandidateSet:
|
||
"""Build only the trusted route; intent remains a manager decision."""
|
||
business_code = str((task_context or {}).get("business_code") or "").strip() or None
|
||
manager_id, route_seats, route_mode = self._route(business_code)
|
||
if not manager_id:
|
||
raise DispatchConfigurationError(
|
||
"no trusted MAF manager route is configured for this conversation"
|
||
)
|
||
|
||
normalized_manager = str(manager_id).strip()
|
||
candidates = [str(agent_id).strip() for agent_id in route_seats]
|
||
max_seats = int(_read(self._config, "max_seats_per_run", 16) or 16)
|
||
# A route may intentionally be manager-only, but it must always have
|
||
# a concrete manager. Candidate seats are optional until the manager
|
||
# chooses an expert intent.
|
||
if not normalized_manager:
|
||
raise DispatchConfigurationError("trusted manager route is blank")
|
||
if len(candidates) > max_seats:
|
||
raise DispatchConfigurationError("configured MAF route exceeds max_seats_per_run")
|
||
if normalized_manager in candidates or len(set(candidates)) != len(candidates):
|
||
raise DispatchConfigurationError("configured MAF route contains duplicate agents")
|
||
try:
|
||
return MafDispatchCandidateSet(
|
||
manager_agent_id=normalized_manager,
|
||
candidate_seat_agent_ids=candidates,
|
||
mode=route_mode or _read(self._config, "default_mode", "structured"),
|
||
business_code=business_code,
|
||
max_rounds=int(_read(self._config, "default_max_rounds", 8) or 8),
|
||
)
|
||
except (TypeError, ValueError) as exc:
|
||
raise DispatchConfigurationError("invalid trusted MAF dispatch route") from exc
|
||
|
||
def build_plan(
|
||
self,
|
||
*,
|
||
intent: MafIntentDecision,
|
||
task_context: dict[str, Any] | None,
|
||
) -> MafDispatchPlan:
|
||
business_code = str((task_context or {}).get("business_code") or "").strip() or None
|
||
manager_id, route_seats, route_mode = self._route(business_code)
|
||
if not manager_id:
|
||
raise DispatchConfigurationError(
|
||
"no trusted MAF manager route is configured for this conversation"
|
||
)
|
||
|
||
# The server owns the route. Intent only changes how many configured
|
||
# seats participate; it never adds an id.
|
||
candidate_seats = route_seats[:]
|
||
if intent.intent_type == "direct_answer":
|
||
seat_ids = []
|
||
execution_mode: MafExecutionMode = "manager_only"
|
||
elif intent.intent_type in {"single_seat", "tool_write"}:
|
||
seat_ids = route_seats[:1]
|
||
execution_mode = "single"
|
||
elif intent.intent_type == "chain_consultation":
|
||
seat_ids = route_seats[:]
|
||
execution_mode = "chain"
|
||
else:
|
||
seat_ids = route_seats[:]
|
||
execution_mode = "parallel"
|
||
|
||
max_seats = int(_read(self._config, "max_seats_per_run", 16) or 16)
|
||
if len(seat_ids) > max_seats:
|
||
raise DispatchConfigurationError("configured MAF route exceeds max_seats_per_run")
|
||
if intent.intent_type != "direct_answer" and not seat_ids:
|
||
raise DispatchConfigurationError("selected MAF intent requires a configured seat route")
|
||
if intent.intent_type == "parallel_consultation" and len(seat_ids) < 2:
|
||
raise DispatchConfigurationError(
|
||
"parallel consultation requires at least two configured seats"
|
||
)
|
||
|
||
try:
|
||
plan = MafDispatchPlan(
|
||
manager_agent_id=str(manager_id).strip(),
|
||
seat_agent_ids=[str(agent_id).strip() for agent_id in seat_ids],
|
||
assignments=[
|
||
MafDispatchAssignment(
|
||
agent_id=str(agent_id).strip(),
|
||
sort_order=index,
|
||
task_goal=f"Complete the {intent.intent_type} task",
|
||
delivery_requirements=["structured result", "evidence or limitation notes"],
|
||
)
|
||
for index, agent_id in enumerate(seat_ids, start=1)
|
||
],
|
||
mode=route_mode or _read(self._config, "default_mode", "structured"),
|
||
intent_type=intent.intent_type,
|
||
business_code=business_code,
|
||
max_rounds=int(_read(self._config, "default_max_rounds", 8) or 8),
|
||
candidate_seat_agent_ids=[str(agent_id).strip() for agent_id in candidate_seats],
|
||
execution_mode=execution_mode,
|
||
convergence_path=[
|
||
"manager answer" if intent.intent_type == "direct_answer" else "manager synthesis"
|
||
],
|
||
reason="server-owned candidate route",
|
||
source="server",
|
||
)
|
||
except (TypeError, ValueError) as exc:
|
||
raise DispatchConfigurationError("invalid trusted MAF dispatch route") from exc
|
||
|
||
trusted_ids = {str(manager_id), *(str(agent_id) for agent_id in route_seats)}
|
||
validate_dispatch_plan(plan, trusted_agent_ids=trusted_ids, max_seats=max_seats)
|
||
return plan
|
||
|
||
@staticmethod
|
||
def parse_manager_decision(
|
||
raw: str,
|
||
*,
|
||
candidate_set: MafDispatchCandidateSet,
|
||
max_seats: int,
|
||
capability_snapshot: dict[str, Any] | None = None,
|
||
) -> tuple[MafManagerPlanOutput, MafDispatchPlan | None]:
|
||
"""Parse the manager's complete intent + dispatch decision.
|
||
|
||
The manager may choose only from the server-provided candidate set.
|
||
A clarification is a valid terminal decision with no dispatch plan;
|
||
every malformed or unauthorized response raises for safe fallback.
|
||
"""
|
||
try:
|
||
payload = _extract_json_object(raw)
|
||
proposed = MafManagerPlanOutput.model_validate(payload)
|
||
except Exception as exc: # noqa: BLE001 - caller uses clarification fallback
|
||
raise InvalidDispatchPlanError("manager plan schema invalid") from exc
|
||
|
||
candidates = list(candidate_set.candidate_seat_agent_ids)
|
||
selected = list(proposed.selected_seat_agent_ids)
|
||
if any(agent_id not in candidates for agent_id in selected):
|
||
raise InvalidDispatchPlanError("manager selected a seat outside the candidate set")
|
||
if len(selected) > max_seats:
|
||
raise InvalidDispatchPlanError("manager plan exceeds the seat limit")
|
||
if candidate_set.manager_agent_id in selected:
|
||
raise InvalidDispatchPlanError("manager cannot select itself as a seat")
|
||
if proposed.execution_mode == "group_chat" and candidate_set.mode != "group_chat":
|
||
raise InvalidDispatchPlanError("group_chat is not enabled for this route")
|
||
if proposed.intent_type == "chain_consultation":
|
||
candidate_order = {agent_id: index for index, agent_id in enumerate(candidates)}
|
||
orders = [candidate_order[agent_id] for agent_id in selected]
|
||
if orders != sorted(orders):
|
||
raise InvalidDispatchPlanError(
|
||
"chain seats must preserve the configured business order"
|
||
)
|
||
if proposed.intent_type == "tool_write" and capability_snapshot is not None:
|
||
agents = capability_snapshot.get("agents")
|
||
by_id = {
|
||
str(agent.get("agent_id")): agent
|
||
for agent in agents or []
|
||
if isinstance(agent, dict)
|
||
}
|
||
if not any(
|
||
any(
|
||
isinstance(tool, dict) and tool.get("side_effect") == "write"
|
||
for tool in (by_id.get(agent_id, {}).get("tools") or [])
|
||
)
|
||
for agent_id in selected
|
||
):
|
||
raise InvalidDispatchPlanError("tool_write has no trusted write-capable seat")
|
||
|
||
if proposed.intent_type == "clarification":
|
||
return proposed, None
|
||
|
||
assignments = [
|
||
MafDispatchAssignment(
|
||
agent_id=item.agent_id,
|
||
sort_order=item.order,
|
||
task_goal=item.task_goal,
|
||
delivery_requirements=list(item.delivery_requirements),
|
||
)
|
||
for item in proposed.assignments
|
||
]
|
||
try:
|
||
plan = MafDispatchPlan(
|
||
manager_agent_id=candidate_set.manager_agent_id,
|
||
seat_agent_ids=selected,
|
||
assignments=assignments,
|
||
mode=candidate_set.mode,
|
||
intent_type=proposed.intent_type,
|
||
business_code=candidate_set.business_code,
|
||
max_rounds=candidate_set.max_rounds,
|
||
candidate_seat_agent_ids=candidates,
|
||
execution_mode=proposed.execution_mode,
|
||
convergence_path=list(proposed.convergence_path),
|
||
reason=proposed.reason,
|
||
source="manager",
|
||
)
|
||
except (TypeError, ValueError) as exc:
|
||
raise InvalidDispatchPlanError("manager plan failed final validation") from exc
|
||
return proposed, plan
|
||
|
||
@staticmethod
|
||
def parse_manager_plan(
|
||
raw: str,
|
||
*,
|
||
expected_intent: MafIntentType,
|
||
server_plan: MafDispatchPlan,
|
||
max_seats: int,
|
||
) -> MafDispatchPlan:
|
||
"""Compatibility wrapper for callers with a pre-resolved intent."""
|
||
candidate_set = MafDispatchCandidateSet(
|
||
manager_agent_id=server_plan.manager_agent_id,
|
||
candidate_seat_agent_ids=list(server_plan.candidate_seat_agent_ids),
|
||
mode=server_plan.mode,
|
||
business_code=server_plan.business_code,
|
||
max_rounds=server_plan.max_rounds,
|
||
)
|
||
proposed, plan = MafDispatchCatalog.parse_manager_decision(
|
||
raw,
|
||
candidate_set=candidate_set,
|
||
max_seats=max_seats,
|
||
)
|
||
if proposed.intent_type != expected_intent:
|
||
raise InvalidDispatchPlanError("manager changed the expected intent")
|
||
if plan is None:
|
||
raise InvalidDispatchPlanError("clarification is not a dispatch plan")
|
||
return plan
|
||
|
||
@staticmethod
|
||
def deterministic_fallback_plan(server_plan: MafDispatchPlan) -> MafDispatchPlan:
|
||
"""Use the server route without ever expanding model permissions."""
|
||
return server_plan.model_copy(
|
||
update={
|
||
"seat_agent_ids": list(server_plan.seat_agent_ids),
|
||
"assignments": [item.model_copy() for item in server_plan.assignments],
|
||
"source": "deterministic_fallback",
|
||
"reason": "manager_plan_invalid_safe_route",
|
||
}
|
||
)
|
||
|
||
@staticmethod
|
||
def participants(
|
||
plan: MafDispatchPlan, *, include_candidates: bool = True
|
||
) -> list[dict[str, Any]]:
|
||
"""Return persistence records; thread ids are intentionally absent."""
|
||
candidate_seats = (
|
||
list(plan.candidate_seat_agent_ids)
|
||
if include_candidates and plan.intent_type != "direct_answer"
|
||
else []
|
||
)
|
||
return [
|
||
{
|
||
"agent_id": plan.manager_agent_id,
|
||
"role": "manager",
|
||
"sort_order": 0,
|
||
"display_name": "",
|
||
},
|
||
*[
|
||
{
|
||
"agent_id": agent_id,
|
||
"role": "seat",
|
||
"sort_order": index,
|
||
"display_name": "",
|
||
}
|
||
for index, agent_id in enumerate(candidate_seats, start=1)
|
||
],
|
||
]
|
||
|
||
@staticmethod
|
||
def candidate_participants(
|
||
candidate_set: MafDispatchCandidateSet,
|
||
*,
|
||
include_seats: bool = True,
|
||
capability_snapshot: dict[str, Any] | None = None,
|
||
) -> list[dict[str, Any]]:
|
||
"""Build ephemeral snapshot inputs or persisted manager-only records."""
|
||
snapshots = {
|
||
str(agent.get("agent_id")): dict(agent)
|
||
for agent in (capability_snapshot or {}).get("agents", [])
|
||
if isinstance(agent, dict)
|
||
}
|
||
rows = [{
|
||
"agent_id": candidate_set.manager_agent_id,
|
||
"role": "manager",
|
||
"sort_order": 0,
|
||
"display_name": "",
|
||
"capability_snapshot": snapshots.get(candidate_set.manager_agent_id, {}),
|
||
}]
|
||
if include_seats:
|
||
rows.extend({
|
||
"agent_id": agent_id,
|
||
"role": "seat",
|
||
"sort_order": index,
|
||
"display_name": "",
|
||
"capability_snapshot": snapshots.get(agent_id, {}),
|
||
} for index, agent_id in enumerate(candidate_set.candidate_seat_agent_ids, start=1))
|
||
return rows
|
||
|
||
@staticmethod
|
||
def manager_capability_summary(
|
||
snapshot: dict[str, Any] | None,
|
||
candidate_set: MafDispatchCandidateSet,
|
||
) -> list[dict[str, Any]]:
|
||
"""Return only safe, non-secret capability facts for the manager prompt."""
|
||
by_id = {
|
||
str(agent.get("agent_id")): agent
|
||
for agent in (snapshot or {}).get("agents", [])
|
||
if isinstance(agent, dict)
|
||
}
|
||
result: list[dict[str, Any]] = []
|
||
for agent_id in candidate_set.candidate_seat_agent_ids:
|
||
agent = by_id.get(agent_id, {})
|
||
tools = [
|
||
{
|
||
"tool_name": str(tool.get("tool_name")),
|
||
"side_effect": tool.get("side_effect", "write"),
|
||
"risk_level": tool.get("risk_level", "R3"),
|
||
}
|
||
for tool in (agent.get("tools") or [])
|
||
if isinstance(tool, dict) and tool.get("tool_name")
|
||
]
|
||
result.append({
|
||
"agent_id": agent_id,
|
||
"model_name": agent.get("model_name"),
|
||
"tool_names": [tool["tool_name"] for tool in tools],
|
||
"tool_summary": tools,
|
||
})
|
||
return result
|
||
|
||
|
||
def validate_dispatch_plan(
|
||
plan: MafDispatchPlan,
|
||
*,
|
||
trusted_agent_ids: set[str],
|
||
max_seats: int,
|
||
) -> None:
|
||
"""Final allowlist check used before persistence and workflow launch."""
|
||
|
||
if plan.manager_agent_id not in trusted_agent_ids:
|
||
raise InvalidDispatchPlanError("manager agent is outside the trusted route")
|
||
if len(plan.seat_agent_ids) > max_seats:
|
||
raise InvalidDispatchPlanError("dispatch plan exceeds the seat limit")
|
||
if any(agent_id not in trusted_agent_ids for agent_id in plan.seat_agent_ids):
|
||
raise InvalidDispatchPlanError("seat agent is outside the trusted route")
|
||
if len(set(plan.seat_agent_ids)) != len(plan.seat_agent_ids):
|
||
raise InvalidDispatchPlanError("dispatch plan contains duplicate seats")
|
||
if plan.manager_agent_id in plan.seat_agent_ids:
|
||
raise InvalidDispatchPlanError("manager cannot be dispatched as a seat")
|
||
|
||
|
||
def _extract_json_object(raw: str) -> dict[str, Any]:
|
||
text = str(raw or "").strip()
|
||
if not text:
|
||
raise ValueError("empty manager plan")
|
||
try:
|
||
value = json.loads(text)
|
||
except json.JSONDecodeError:
|
||
start = text.find("{")
|
||
end = text.rfind("}")
|
||
if start < 0 or end <= start:
|
||
raise ValueError("manager plan is not JSON")
|
||
value = json.loads(text[start:end + 1])
|
||
if not isinstance(value, dict):
|
||
raise ValueError("manager plan must be a JSON object")
|
||
return value
|
||
|
||
|
||
__all__ = [
|
||
"DispatchConfigurationError",
|
||
"InvalidDispatchPlanError",
|
||
"MafDispatchAssignment",
|
||
"MafDispatchCandidateSet",
|
||
"MafExecutionMode",
|
||
"MafManagerPlanAssignment",
|
||
"MafManagerPlanOutput",
|
||
"MafDispatchPlan",
|
||
"MafDispatchCatalog",
|
||
"MafIntentDecision",
|
||
"MafIntentType",
|
||
"resolve_intent",
|
||
"validate_dispatch_plan",
|
||
]
|