董宏凯andCursor
1a080311bd
Make ensure_participants concurrent-safe, restore clarification checkpoints without re-planning, and only create threads for validated seats. Co-authored-by: Cursor <cursoragent@cursor.com>
1683 lines
74 KiB
Python
1683 lines
74 KiB
Python
"""MAF Workflow node executors for the roundtable experiment.
|
||
|
||
Each executor is a real MAF ``Executor`` subclass instance. Every seat gets
|
||
its own executor bound to a fixed ``agent_id`` — no shared function that
|
||
guesses the current speaker from input metadata.
|
||
|
||
Phase 3 structured topology:
|
||
- PlannerExecutor → Expert SeatExecutors (fan-out) → CrossReviewExecutor →
|
||
ApprovalGate → Finalizer | WaitingTerminal
|
||
|
||
Group-chat still uses CoordinatorExecutor for speaker selection.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from typing import Any
|
||
|
||
from agent_framework import Executor, WorkflowContext, handler
|
||
|
||
from app.maf_roundtable.action_coordinator import (
|
||
ActionCoordinator,
|
||
classify_tool,
|
||
tool_needs_approval,
|
||
)
|
||
from app.maf_roundtable.context_builder import build_manager_synthesis_input
|
||
from app.maf_roundtable.dispatch_contracts import (
|
||
InvalidDispatchPlanError,
|
||
MafDispatchCandidateSet,
|
||
MafDispatchCatalog,
|
||
MafDispatchPlan,
|
||
)
|
||
from app.maf_roundtable.fact_ledger import (
|
||
confirm_candidates,
|
||
empty_ledger,
|
||
merge_seat_output,
|
||
normalize_ledger,
|
||
record_action_transition,
|
||
record_tool_result,
|
||
)
|
||
from app.maf_roundtable.seat_output import (
|
||
build_repair_prompt,
|
||
parse_cross_review_output,
|
||
seat_output_instruction,
|
||
validate_or_repair_seat_output,
|
||
)
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class WriteGuardConfigurationError(RuntimeError):
|
||
"""Raised when WriteGuard should have intercepted a write tool but didn't."""
|
||
|
||
|
||
# ── Serializable workflow state ─────────────────────────────────────────────
|
||
|
||
|
||
class RoundtableState:
|
||
"""Serializable shared state for the MAF roundtable workflow.
|
||
|
||
Persisted to the run's ``maf_checkpoint_json`` so the workflow survives
|
||
restarts and approval-pause/resume cycles.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
*,
|
||
session_id: str,
|
||
run_id: str,
|
||
task_text: str,
|
||
mode: str,
|
||
model_name: str | None,
|
||
reasoning_effort: str | None,
|
||
max_rounds: int,
|
||
write_policy: str,
|
||
manager_agent_id: str | None,
|
||
seat_agent_ids: list[str],
|
||
thread_map: dict[str, str],
|
||
capability_snapshot: dict[str, Any] | None = None,
|
||
seat_timeout_seconds: float = 600.0,
|
||
schema_repair_enabled: bool = True,
|
||
intent_type: str | None = None,
|
||
dispatch_plan: dict[str, Any] | None = None,
|
||
candidate_seat_agent_ids: list[str] | None = None,
|
||
dispatch_contract_enabled: bool | None = None,
|
||
) -> None:
|
||
self.session_id = session_id
|
||
self.run_id = run_id
|
||
self.task_text = task_text
|
||
self.mode = mode
|
||
self.model_name = model_name
|
||
self.reasoning_effort = reasoning_effort
|
||
self.max_rounds = max_rounds
|
||
self.write_policy = write_policy
|
||
self.manager_agent_id = manager_agent_id
|
||
self.seat_agent_ids = seat_agent_ids
|
||
self.thread_map = thread_map
|
||
self.capability_snapshot = capability_snapshot
|
||
self.seat_timeout_seconds = seat_timeout_seconds
|
||
self.schema_repair_enabled = schema_repair_enabled
|
||
# Stage 3 server-validated routing metadata. The workflow may use
|
||
# this for diagnostics/checkpoints, but it never accepts a new agent
|
||
# id from manager model output.
|
||
self.intent_type = intent_type
|
||
self.dispatch_plan = dispatch_plan
|
||
self.dispatch_contract_enabled = (
|
||
bool(dispatch_plan is not None or intent_type is not None)
|
||
if dispatch_contract_enabled is None
|
||
else bool(dispatch_contract_enabled)
|
||
)
|
||
plan_seats = list((dispatch_plan or {}).get("seat_agent_ids") or [])
|
||
self.candidate_seat_agent_ids = list(
|
||
(dispatch_plan or {}).get("candidate_seat_agent_ids")
|
||
or candidate_seat_agent_ids
|
||
or seat_agent_ids
|
||
)
|
||
self.selected_seat_agent_ids: list[str] | None = plan_seats or None
|
||
|
||
self.current_round = 0
|
||
self.turn_results: list[dict[str, Any]] = []
|
||
self.fact_ledger: dict[str, Any] = empty_ledger()
|
||
self.final_result: dict[str, Any] | None = None
|
||
self.status = "running"
|
||
self.pending_approvals: list[dict[str, Any]] = []
|
||
|
||
# Phase 3 orchestration fields
|
||
self.plan: dict[str, Any] | None = None
|
||
self.seat_outcomes: dict[str, Any] = {}
|
||
self.completed_nodes: list[str] = []
|
||
self.stage: str = "planning"
|
||
self.partial_failure: bool = False
|
||
self.cross_review_result: dict[str, Any] | None = None
|
||
|
||
def mark_completed(self, node_id: str) -> None:
|
||
if node_id not in self.completed_nodes:
|
||
self.completed_nodes.append(node_id)
|
||
|
||
def is_completed(self, node_id: str) -> bool:
|
||
return node_id in self.completed_nodes
|
||
|
||
def to_dict(self) -> dict[str, Any]:
|
||
return {
|
||
"session_id": self.session_id,
|
||
"run_id": self.run_id,
|
||
"task_text": self.task_text,
|
||
"mode": self.mode,
|
||
"model_name": self.model_name,
|
||
"reasoning_effort": self.reasoning_effort,
|
||
"max_rounds": self.max_rounds,
|
||
"write_policy": self.write_policy,
|
||
"manager_agent_id": self.manager_agent_id,
|
||
"seat_agent_ids": self.seat_agent_ids,
|
||
"thread_map": self.thread_map,
|
||
"capability_snapshot": self.capability_snapshot,
|
||
"seat_timeout_seconds": self.seat_timeout_seconds,
|
||
"schema_repair_enabled": self.schema_repair_enabled,
|
||
"intent_type": self.intent_type,
|
||
"dispatch_plan": self.dispatch_plan,
|
||
"dispatch_contract_enabled": self.dispatch_contract_enabled,
|
||
"candidate_seat_agent_ids": list(self.candidate_seat_agent_ids),
|
||
"selected_seat_agent_ids": self.selected_seat_agent_ids,
|
||
"current_round": self.current_round,
|
||
"turn_results": self.turn_results,
|
||
"fact_ledger": self.fact_ledger,
|
||
"final_result": self.final_result,
|
||
"status": self.status,
|
||
"pending_approvals": self.pending_approvals,
|
||
"plan": self.plan,
|
||
"seat_outcomes": self.seat_outcomes,
|
||
"completed_nodes": list(self.completed_nodes),
|
||
"stage": self.stage,
|
||
"partial_failure": self.partial_failure,
|
||
"cross_review_result": self.cross_review_result,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, d: dict[str, Any]) -> RoundtableState:
|
||
state = cls(
|
||
session_id=d["session_id"],
|
||
run_id=d["run_id"],
|
||
task_text=d["task_text"],
|
||
mode=d["mode"],
|
||
model_name=d.get("model_name"),
|
||
reasoning_effort=d.get("reasoning_effort"),
|
||
max_rounds=d.get("max_rounds", 8),
|
||
write_policy=d.get("write_policy", "confirm_high_risk"),
|
||
manager_agent_id=d.get("manager_agent_id"),
|
||
seat_agent_ids=d.get("seat_agent_ids", []),
|
||
thread_map=d.get("thread_map", {}),
|
||
capability_snapshot=d.get("capability_snapshot"),
|
||
seat_timeout_seconds=float(d.get("seat_timeout_seconds") or 600.0),
|
||
schema_repair_enabled=bool(d.get("schema_repair_enabled", True)),
|
||
intent_type=d.get("intent_type"),
|
||
dispatch_plan=d.get("dispatch_plan"),
|
||
candidate_seat_agent_ids=d.get("candidate_seat_agent_ids"),
|
||
dispatch_contract_enabled=d.get("dispatch_contract_enabled"),
|
||
)
|
||
state.current_round = d.get("current_round", 0)
|
||
state.turn_results = d.get("turn_results", [])
|
||
state.fact_ledger = normalize_ledger(d.get("fact_ledger"))
|
||
state.final_result = d.get("final_result")
|
||
state.status = d.get("status", "running")
|
||
pending = d.get("pending_approvals")
|
||
if pending is None and d.get("pending_approval"):
|
||
pending = [d["pending_approval"]]
|
||
state.pending_approvals = pending or []
|
||
state.plan = d.get("plan")
|
||
state.seat_outcomes = d.get("seat_outcomes") or {}
|
||
state.completed_nodes = list(d.get("completed_nodes") or [])
|
||
state.stage = d.get("stage") or "planning"
|
||
state.partial_failure = bool(d.get("partial_failure", False))
|
||
state.cross_review_result = d.get("cross_review_result")
|
||
state.candidate_seat_agent_ids = list(
|
||
d.get("candidate_seat_agent_ids") or state.candidate_seat_agent_ids
|
||
)
|
||
selected = d.get("selected_seat_agent_ids")
|
||
state.selected_seat_agent_ids = list(selected) if selected is not None else None
|
||
state.dispatch_contract_enabled = bool(
|
||
d.get("dispatch_contract_enabled", state.dispatch_contract_enabled)
|
||
)
|
||
return state
|
||
|
||
|
||
# ── Planner (structured) ────────────────────────────────────────────────────
|
||
|
||
|
||
class ManagerAnswerExecutor(Executor):
|
||
"""Manager-only topology for ordinary direct answers."""
|
||
|
||
def __init__(self, adapter: Any, state: RoundtableState) -> None:
|
||
super().__init__(id="manager_answer")
|
||
self._adapter = adapter
|
||
self._state = state
|
||
|
||
@handler(input=Any, workflow_output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
if self._state.is_completed("manager_answer") and self._state.final_result is not None:
|
||
self._state.status = "completed"
|
||
await ctx.yield_output({"final_result": self._state.final_result, "replayed": True})
|
||
return
|
||
|
||
manager_id = self._state.manager_agent_id or ""
|
||
thread_id = self._state.thread_map.get(manager_id, "")
|
||
if not thread_id:
|
||
self._state.status = "failed"
|
||
self._state.final_result = {
|
||
"content": "MAF manager thread is unavailable",
|
||
"agent_id": manager_id,
|
||
"error_code": "manager_thread_unavailable",
|
||
}
|
||
else:
|
||
try:
|
||
response = await asyncio.wait_for(
|
||
self._adapter.run_seat_turn(
|
||
session_id=self._state.session_id,
|
||
run_id=self._state.run_id,
|
||
agent_id=manager_id,
|
||
turn_index=0,
|
||
execution_stage="direct_answer",
|
||
message=self._state.task_text,
|
||
thread_id=thread_id,
|
||
model_name=self._state.model_name,
|
||
reasoning_effort=self._state.reasoning_effort,
|
||
task_text=self._state.task_text,
|
||
stage_instruction="Answer the user directly; do not delegate to expert seats",
|
||
fact_ledger=self._state.fact_ledger,
|
||
capability_snapshot=self._state.capability_snapshot,
|
||
write_policy=self._state.write_policy,
|
||
wait_timeout_seconds=self._state.seat_timeout_seconds,
|
||
),
|
||
timeout=self._state.seat_timeout_seconds,
|
||
)
|
||
if response.error:
|
||
self._state.status = "failed"
|
||
else:
|
||
self._state.status = "completed"
|
||
self._state.final_result = {
|
||
"content": response.content,
|
||
"agent_id": manager_id,
|
||
"error": response.error,
|
||
"quality_decision": "accept" if not response.error else "controlled_gap",
|
||
}
|
||
except Exception as exc: # noqa: BLE001 - runner persists a safe failure
|
||
self._state.status = "failed"
|
||
self._state.final_result = {
|
||
"content": "MAF manager answer failed",
|
||
"agent_id": manager_id,
|
||
"error_code": type(exc).__name__,
|
||
}
|
||
|
||
self._state.stage = "done"
|
||
self._state.mark_completed("manager_answer")
|
||
dispatch_plan = MafDispatchPlan.model_validate(self._state.dispatch_plan or {
|
||
"manager_agent_id": manager_id,
|
||
"seat_agent_ids": [],
|
||
"assignments": [],
|
||
"intent_type": "direct_answer",
|
||
"max_rounds": self._state.max_rounds,
|
||
"candidate_seat_agent_ids": list(self._state.candidate_seat_agent_ids),
|
||
"execution_mode": "manager_only",
|
||
"convergence_path": ["manager answer"],
|
||
"reason": "direct answer uses manager only",
|
||
"source": "server",
|
||
})
|
||
self._state.dispatch_plan = dispatch_plan.model_dump(mode="json")
|
||
store = getattr(getattr(self._adapter, "_service", None), "store", None)
|
||
if store is not None:
|
||
await store.runs.update_run(
|
||
self._state.run_id,
|
||
{"maf_checkpoint": self._state.to_dict()},
|
||
)
|
||
if not self._state.dispatch_contract_enabled:
|
||
await self._adapter._emit_critical(self._state.run_id, {
|
||
"event_type": "manager.dispatch",
|
||
"agent_id": manager_id,
|
||
"turn_index": 0,
|
||
"payload": {
|
||
"intent_type": dispatch_plan.intent_type,
|
||
"execution_mode": dispatch_plan.execution_mode,
|
||
"convergence_path": list(dispatch_plan.convergence_path),
|
||
"selected_seat_agent_ids": [],
|
||
"assignments": [],
|
||
"reason": dispatch_plan.reason,
|
||
"source": dispatch_plan.source,
|
||
},
|
||
})
|
||
from app.maf_roundtable.event_normalizer import make_run_completed
|
||
|
||
await self._adapter._emit_critical(
|
||
self._state.run_id,
|
||
make_run_completed(manager_id, 1, {"manager_only": True}),
|
||
)
|
||
await ctx.yield_output({"final_result": self._state.final_result})
|
||
|
||
|
||
class ManagerDispatchExecutor(Executor):
|
||
"""Understand intent and emit one server-validated dispatch decision."""
|
||
|
||
def __init__(
|
||
self, adapter: Any, state: RoundtableState, *, executor_id: str = "manager_dispatch"
|
||
) -> None:
|
||
super().__init__(id=executor_id)
|
||
self._adapter = adapter
|
||
self._state = state
|
||
self._node_id = executor_id
|
||
|
||
def _dispatch_decision(self) -> str:
|
||
plan = self._state.dispatch_plan or {}
|
||
if self._state.intent_type == "clarification" or not plan:
|
||
return "clarification"
|
||
if plan.get("execution_mode") == "manager_only":
|
||
return "manager_only"
|
||
if plan.get("mode") == "group_chat" or plan.get("execution_mode") == "group_chat":
|
||
return "group_chat"
|
||
return "selected_seats"
|
||
|
||
@handler(input=Any, output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
if self._state.is_completed(self._node_id):
|
||
await ctx.send_message({
|
||
"decision": self._dispatch_decision(),
|
||
"agents": list(self._state.selected_seat_agent_ids or []),
|
||
"execution_mode": (self._state.dispatch_plan or {}).get("execution_mode"),
|
||
"replayed": True,
|
||
})
|
||
return
|
||
|
||
self._state.stage = "planning"
|
||
|
||
# Legacy MAF roundtable runs do not carry the V2 conversation dispatch
|
||
# contract. Keep their planner semantics isolated from the new MAF
|
||
# conversation workflow; in particular, old seat-only runs may have no
|
||
# manager at all.
|
||
if not self._state.dispatch_contract_enabled:
|
||
self._state.current_round = max(self._state.current_round, 1)
|
||
await _run_legacy_planner(self, ctx)
|
||
return
|
||
|
||
# The manager plan is a strict JSON contract. Legacy free-text
|
||
# planning is isolated in _run_legacy_planner; new Stage 3 runs
|
||
# always return through this verifier.
|
||
await _resolve_manager_dispatch(self, ctx)
|
||
return
|
||
|
||
|
||
async def _run_legacy_planner(executor: ManagerDispatchExecutor, ctx: WorkflowContext) -> None:
|
||
"""Keep the pre-V2 free-text planner isolated from new conversations."""
|
||
state = executor._state
|
||
if state.is_completed(executor._node_id):
|
||
await ctx.send_message({"decision": "parallel_seats", "agents": list(state.seat_agent_ids), "replayed": True})
|
||
return
|
||
assignments = {
|
||
agent_id: {
|
||
"role": f"Expert seat {agent_id}",
|
||
"instruction": f"Independently analyze the task. {seat_output_instruction()}",
|
||
}
|
||
for agent_id in state.seat_agent_ids
|
||
}
|
||
manager_id = state.manager_agent_id
|
||
thread_id = state.thread_map.get(manager_id or "", "")
|
||
if manager_id and thread_id:
|
||
try:
|
||
response = await asyncio.wait_for(
|
||
executor._adapter.run_seat_turn(
|
||
session_id=state.session_id,
|
||
run_id=state.run_id,
|
||
agent_id=manager_id,
|
||
turn_index=0,
|
||
execution_stage="planner",
|
||
message=state.task_text,
|
||
thread_id=thread_id,
|
||
model_name=state.model_name,
|
||
reasoning_effort=state.reasoning_effort,
|
||
task_text=state.task_text,
|
||
stage_instruction="Produce a concise assignment for each configured expert seat.",
|
||
fact_ledger=state.fact_ledger,
|
||
capability_snapshot=state.capability_snapshot,
|
||
write_policy=state.write_policy,
|
||
wait_timeout_seconds=state.seat_timeout_seconds,
|
||
),
|
||
timeout=state.seat_timeout_seconds,
|
||
)
|
||
plan_text = (response.content or "").strip()
|
||
if plan_text:
|
||
for agent_id in state.seat_agent_ids:
|
||
assignments[agent_id]["instruction"] = f"{plan_text[:1500]}\n\nSeat id: {agent_id}.\n{seat_output_instruction()}"
|
||
state.plan = {"source": "manager", "text": plan_text, "assignments": assignments}
|
||
else:
|
||
state.plan = {"source": "deterministic", "assignments": assignments}
|
||
except Exception as exc: # noqa: BLE001 - legacy path deliberately degrades
|
||
logger.warning("legacy planner failed (%s); using deterministic plan", type(exc).__name__)
|
||
state.plan = {"source": "deterministic", "assignments": assignments}
|
||
else:
|
||
state.plan = {"source": "deterministic", "assignments": assignments}
|
||
state.mark_completed(executor._node_id)
|
||
state.stage = "experts"
|
||
await ctx.send_message({"decision": "parallel_seats", "agents": list(state.seat_agent_ids)})
|
||
|
||
|
||
class PlannerExecutor(ManagerDispatchExecutor):
|
||
"""Compatibility node retained solely for pre-V2 MAF roundtable runs."""
|
||
|
||
def __init__(self, adapter: Any, state: RoundtableState) -> None:
|
||
super().__init__(adapter, state, executor_id="planner")
|
||
|
||
|
||
class DispatchRouteExecutor(Executor):
|
||
"""Route only from the verified, persisted manager dispatch decision."""
|
||
|
||
def __init__(self, state: RoundtableState) -> None:
|
||
super().__init__(id="dispatch_route")
|
||
self._state = state
|
||
|
||
@handler(input=Any, output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
plan = self._state.dispatch_plan or {}
|
||
if self._state.intent_type == "clarification" or not plan:
|
||
decision = "clarification"
|
||
elif plan.get("execution_mode") == "manager_only":
|
||
decision = "manager_only"
|
||
elif plan.get("mode") == "group_chat" or plan.get("execution_mode") == "group_chat":
|
||
decision = "group_chat"
|
||
else:
|
||
decision = "selected_seats"
|
||
await ctx.send_message({"decision": decision, "agents": list(self._state.selected_seat_agent_ids or [])})
|
||
|
||
|
||
class DispatchSeatEntryExecutor(Executor):
|
||
"""Start seat work only after the manager's plan and threads are durable."""
|
||
|
||
def __init__(self, state: RoundtableState) -> None:
|
||
super().__init__(id="dispatch_seat_entry")
|
||
self._state = state
|
||
|
||
@handler(input=Any, output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
self._state.stage = "experts"
|
||
await ctx.send_message({"agents": list(self._state.selected_seat_agent_ids or [])})
|
||
|
||
|
||
class ClarificationTerminalExecutor(Executor):
|
||
"""Safe terminal branch for invalid, unauthorized, or timed-out planning."""
|
||
|
||
def __init__(self, state: RoundtableState) -> None:
|
||
super().__init__(id="clarification_terminal")
|
||
self._state = state
|
||
|
||
@handler(input=Any, workflow_output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
self._state.status = "waiting_input"
|
||
self._state.stage = "waiting_input"
|
||
self._state.mark_completed("clarification_terminal")
|
||
await ctx.yield_output({"status": "waiting_input", "clarification_prompt": (self._state.dispatch_plan or {}).get("clarification_prompt")})
|
||
|
||
|
||
class DispatchPreparationTerminalExecutor(Executor):
|
||
"""End the manager-only preparation graph before topology is selected."""
|
||
|
||
def __init__(self, state: RoundtableState) -> None:
|
||
super().__init__(id="dispatch_preparation_terminal")
|
||
self._state = state
|
||
|
||
@handler(input=Any, workflow_output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
self._state.mark_completed("dispatch_preparation_terminal")
|
||
await ctx.yield_output({
|
||
"decision": "clarification" if self._state.intent_type == "clarification" else "dispatched",
|
||
"status": self._state.status,
|
||
})
|
||
|
||
|
||
class CoordinatorExecutor(Executor):
|
||
"""Choose the next selected group-chat speaker without widening the plan."""
|
||
|
||
def __init__(self, adapter: Any, state: RoundtableState) -> None:
|
||
super().__init__(id="coordinator")
|
||
self._adapter = adapter
|
||
self._state = state
|
||
|
||
@handler(input=Any, output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
if self._state.stage == "dispatch_validated" and self._state.current_round == 0:
|
||
self._state.current_round = 1
|
||
else:
|
||
self._state.current_round += 1
|
||
if self._state.current_round > self._state.max_rounds:
|
||
await ctx.send_message({"decision": "done", "reason": "max_rounds_reached"})
|
||
return
|
||
eligible = list(self._state.selected_seat_agent_ids) if self._state.selected_seat_agent_ids is not None else list(self._state.seat_agent_ids)
|
||
remaining = [
|
||
agent_id for agent_id in eligible
|
||
if not any(row.get("agent_id") == agent_id and row.get("turn_index", 0) >= self._state.current_round - 1 for row in self._state.turn_results)
|
||
]
|
||
if not remaining:
|
||
await ctx.send_message({"decision": "done", "reason": "all_selected_seats_spoke"})
|
||
return
|
||
next_agent = remaining[(self._state.current_round - 1) % len(remaining)]
|
||
await self._adapter._emit_critical(self._state.run_id, {
|
||
"event_type": "manager.decision",
|
||
"agent_id": self._state.manager_agent_id or "coordinator",
|
||
"turn_index": self._state.current_round,
|
||
"payload": {"next_speaker": next_agent, "reason": "selected_plan_round_robin", "round": self._state.current_round},
|
||
})
|
||
await ctx.send_message({"decision": "next_speaker", "agent_id": next_agent})
|
||
|
||
|
||
async def _persist_dispatch_checkpoint(adapter: Any, state: RoundtableState) -> None:
|
||
store = getattr(getattr(adapter, "_service", None), "store", None)
|
||
if store is not None:
|
||
await store.runs.update_run(state.run_id, {"maf_checkpoint": state.to_dict()})
|
||
|
||
|
||
async def _emit_dispatch_event(
|
||
adapter: Any,
|
||
state: RoundtableState,
|
||
*,
|
||
clarification_prompt: str | None = None,
|
||
) -> None:
|
||
plan = state.dispatch_plan or {}
|
||
payload: dict[str, Any] = {
|
||
"intent_type": state.intent_type,
|
||
"execution_mode": plan.get("execution_mode", "manager_only"),
|
||
"selected_seat_agent_ids": list(state.selected_seat_agent_ids or []),
|
||
"assignments": plan.get("assignments", []),
|
||
"convergence_path": plan.get("convergence_path", []),
|
||
"reason": plan.get("reason", "manager_plan_invalid_safe_clarification"),
|
||
"source": plan.get("source", "deterministic_fallback"),
|
||
}
|
||
if clarification_prompt:
|
||
payload["clarification_prompt"] = clarification_prompt
|
||
await adapter._emit_critical(state.run_id, {
|
||
"event_type": "manager.dispatch",
|
||
"agent_id": state.manager_agent_id or "",
|
||
"turn_index": 0,
|
||
"payload": payload,
|
||
})
|
||
|
||
|
||
async def _ensure_selected_seat_threads(adapter: Any, state: RoundtableState) -> None:
|
||
"""Persist and bind threads only for server-validated selected seats."""
|
||
selected = list(state.selected_seat_agent_ids or [])
|
||
if not selected:
|
||
return
|
||
store = getattr(getattr(adapter, "_service", None), "store", None)
|
||
participants_repo = getattr(store, "participants", None)
|
||
if participants_repo is None:
|
||
return
|
||
route_mode = state.mode if state.mode in {"structured", "group_chat"} else "structured"
|
||
candidate_set = MafDispatchCandidateSet(
|
||
manager_agent_id=str(state.manager_agent_id or ""),
|
||
candidate_seat_agent_ids=list(state.candidate_seat_agent_ids),
|
||
mode=route_mode,
|
||
max_rounds=state.max_rounds,
|
||
)
|
||
candidate_rows = MafDispatchCatalog.candidate_participants(
|
||
candidate_set,
|
||
capability_snapshot=state.capability_snapshot,
|
||
)
|
||
selected_rows = [row for row in candidate_rows if row.get("agent_id") in selected]
|
||
await participants_repo.ensure_participants(state.session_id, selected_rows)
|
||
persisted = await participants_repo.list_participants(state.session_id)
|
||
selected_persisted = [row for row in persisted if row.get("agent_id") in selected]
|
||
if selected_persisted:
|
||
state.thread_map.update(
|
||
await adapter.ensure_threads(state.session_id, state.run_id, selected_persisted)
|
||
)
|
||
|
||
|
||
async def _resolve_manager_dispatch(
|
||
executor: ManagerDispatchExecutor, ctx: WorkflowContext
|
||
) -> None:
|
||
"""Run, validate, persist, and route the sole V2 manager dispatch turn."""
|
||
adapter = executor._adapter
|
||
state = executor._state
|
||
manager_id = str(state.manager_agent_id or "").strip()
|
||
route_mode = state.mode if state.mode in {"structured", "group_chat"} else "structured"
|
||
candidate_set = MafDispatchCandidateSet(
|
||
manager_agent_id=manager_id,
|
||
candidate_seat_agent_ids=list(state.candidate_seat_agent_ids),
|
||
mode=route_mode,
|
||
max_rounds=state.max_rounds,
|
||
)
|
||
|
||
async def clarification(reason: str) -> None:
|
||
prompt = "Please clarify the goal, expected outcome, and relevant business context."
|
||
state.intent_type = "clarification"
|
||
state.dispatch_plan = {
|
||
"execution_mode": "manager_only",
|
||
"convergence_path": ["clarification"],
|
||
"reason": reason,
|
||
"source": "deterministic_fallback",
|
||
"clarification_prompt": prompt,
|
||
}
|
||
state.selected_seat_agent_ids = []
|
||
state.plan = None
|
||
state.status = "waiting_input"
|
||
state.stage = "waiting_input"
|
||
state.mark_completed(executor._node_id)
|
||
await _persist_dispatch_checkpoint(adapter, state)
|
||
await _emit_dispatch_event(adapter, state, clarification_prompt=prompt)
|
||
await adapter._emit_critical(state.run_id, {
|
||
"event_type": "clarification.required",
|
||
"agent_id": manager_id,
|
||
"turn_index": 0,
|
||
"payload": {"prompt": prompt, "reason": reason},
|
||
})
|
||
await ctx.send_message({"decision": "clarification", "agents": []})
|
||
|
||
# ── P1-2: Clarification checkpoint recovery ──────────────────────────
|
||
# A clarification checkpoint stores a lightweight dict (not a full
|
||
# MafDispatchPlan). Recovery must restore the original prompt without
|
||
# calling the manager model again, creating seat threads, or emitting
|
||
# duplicate audit events.
|
||
if state.intent_type == "clarification":
|
||
prompt = str((state.dispatch_plan or {}).get("clarification_prompt") or "").strip()
|
||
if not prompt:
|
||
await clarification("checkpoint_clarification_prompt_missing")
|
||
return
|
||
|
||
state.selected_seat_agent_ids = []
|
||
state.plan = None
|
||
state.status = "waiting_input"
|
||
state.stage = "waiting_input"
|
||
state.mark_completed(executor._node_id)
|
||
await _persist_dispatch_checkpoint(adapter, state)
|
||
await ctx.send_message({"decision": "clarification", "agents": [], "replayed": True})
|
||
return
|
||
|
||
# A checkpoint written after validation survives process recovery and is
|
||
# never re-planned. It still creates any selected threads that were not
|
||
# bound before interruption.
|
||
if state.dispatch_plan and state.intent_type:
|
||
try:
|
||
existing = MafDispatchPlan.model_validate(state.dispatch_plan)
|
||
except (TypeError, ValueError):
|
||
await clarification("checkpoint_dispatch_plan_invalid")
|
||
return
|
||
state.selected_seat_agent_ids = list(existing.seat_agent_ids)
|
||
state.plan = {
|
||
"source": existing.source,
|
||
"assignments": {
|
||
assignment.agent_id: {
|
||
"role": assignment.role,
|
||
"instruction": "\n".join(
|
||
[assignment.task_goal, *assignment.delivery_requirements]
|
||
),
|
||
}
|
||
for assignment in existing.assignments
|
||
},
|
||
}
|
||
await _ensure_selected_seat_threads(adapter, state)
|
||
state.mark_completed(executor._node_id)
|
||
await _persist_dispatch_checkpoint(adapter, state)
|
||
await ctx.send_message({
|
||
"decision": executor._dispatch_decision(),
|
||
"agents": list(state.selected_seat_agent_ids or []),
|
||
"replayed": True,
|
||
})
|
||
return
|
||
|
||
thread_id = state.thread_map.get(manager_id, "")
|
||
if not manager_id or not thread_id:
|
||
await clarification("manager_thread_unavailable")
|
||
return
|
||
|
||
capability_summary = MafDispatchCatalog.manager_capability_summary(
|
||
state.capability_snapshot,
|
||
candidate_set,
|
||
)
|
||
manager_tool_names = [
|
||
str(tool.get("tool_name"))
|
||
for agent in (state.capability_snapshot or {}).get("agents", [])
|
||
if isinstance(agent, dict) and agent.get("agent_id") == manager_id
|
||
for tool in (agent.get("tools") or [])
|
||
if isinstance(tool, dict) and tool.get("tool_name")
|
||
]
|
||
contract = {
|
||
"intent_type": "direct_answer|clarification|single_seat|parallel_consultation|chain_consultation|tool_write",
|
||
"selected_seat_agent_ids": "candidate IDs only",
|
||
"assignments": "one ordered task_goal and delivery_requirements entry per selected seat",
|
||
"execution_mode": "manager_only|single|parallel|chain|group_chat",
|
||
"convergence_path": ["how work converges"],
|
||
"reason": "brief explanation",
|
||
"clarification_prompt": "required only for clarification",
|
||
}
|
||
prompt = (
|
||
"You are the MAF manager. Determine the user intent and return exactly one JSON object, "
|
||
"without Markdown. You may select only from the trusted candidate seats and their redacted "
|
||
"capabilities. Invalid or uncertain requests must use clarification; never invent a seat, "
|
||
"tool, capability, or route.\n\n"
|
||
f"User task:\n{state.task_text}\n\n"
|
||
f"Trusted route mode: {candidate_set.mode}\n"
|
||
f"Trusted candidate seats: {json.dumps(candidate_set.candidate_seat_agent_ids)}\n"
|
||
f"Trusted capability summary: {json.dumps(capability_summary, ensure_ascii=False)}\n"
|
||
f"Required JSON shape: {json.dumps(contract)}"
|
||
)
|
||
try:
|
||
response = await asyncio.wait_for(
|
||
adapter.run_seat_turn(
|
||
session_id=state.session_id,
|
||
run_id=state.run_id,
|
||
agent_id=manager_id,
|
||
turn_index=0,
|
||
execution_stage="manager_dispatch",
|
||
message=prompt,
|
||
thread_id=thread_id,
|
||
model_name=state.model_name,
|
||
reasoning_effort=state.reasoning_effort,
|
||
task_text=state.task_text,
|
||
stage_instruction="Return only the strict MAF dispatch JSON object.",
|
||
excluded_tools=manager_tool_names,
|
||
fact_ledger=state.fact_ledger,
|
||
capability_snapshot=state.capability_snapshot,
|
||
write_policy=state.write_policy,
|
||
wait_timeout_seconds=state.seat_timeout_seconds,
|
||
),
|
||
timeout=state.seat_timeout_seconds,
|
||
)
|
||
except TimeoutError:
|
||
await clarification("manager_dispatch_timeout")
|
||
return
|
||
except Exception: # noqa: BLE001 - model/runtime details must not reach the user
|
||
await clarification("manager_dispatch_failed")
|
||
return
|
||
if response.error:
|
||
await clarification("manager_dispatch_failed")
|
||
return
|
||
|
||
try:
|
||
proposed, plan = MafDispatchCatalog.parse_manager_decision(
|
||
response.content or "",
|
||
candidate_set=candidate_set,
|
||
max_seats=len(candidate_set.candidate_seat_agent_ids),
|
||
capability_snapshot=state.capability_snapshot,
|
||
)
|
||
except InvalidDispatchPlanError:
|
||
await clarification("manager_dispatch_invalid")
|
||
return
|
||
|
||
state.intent_type = proposed.intent_type
|
||
if plan is None:
|
||
state.dispatch_plan = {
|
||
"execution_mode": "manager_only",
|
||
"convergence_path": list(proposed.convergence_path),
|
||
"reason": proposed.reason,
|
||
"source": "manager",
|
||
"clarification_prompt": proposed.clarification_prompt,
|
||
}
|
||
state.selected_seat_agent_ids = []
|
||
state.plan = None
|
||
state.status = "waiting_input"
|
||
state.stage = "waiting_input"
|
||
state.mark_completed(executor._node_id)
|
||
await _persist_dispatch_checkpoint(adapter, state)
|
||
await _emit_dispatch_event(
|
||
adapter,
|
||
state,
|
||
clarification_prompt=proposed.clarification_prompt,
|
||
)
|
||
await adapter._emit_critical(state.run_id, {
|
||
"event_type": "clarification.required",
|
||
"agent_id": manager_id,
|
||
"turn_index": 0,
|
||
"payload": {"prompt": proposed.clarification_prompt, "reason": proposed.reason},
|
||
})
|
||
await ctx.send_message({"decision": "clarification", "agents": []})
|
||
return
|
||
|
||
state.dispatch_plan = plan.model_dump(mode="json")
|
||
state.selected_seat_agent_ids = list(plan.seat_agent_ids)
|
||
state.plan = {
|
||
"source": plan.source,
|
||
"assignments": {
|
||
assignment.agent_id: {
|
||
"role": assignment.role,
|
||
"instruction": "\n".join(
|
||
[assignment.task_goal, *assignment.delivery_requirements]
|
||
),
|
||
}
|
||
for assignment in plan.assignments
|
||
},
|
||
}
|
||
state.stage = "dispatch_validated"
|
||
# Persist before adding a participant or thread: recovery cannot obtain a
|
||
# different model decision after a process interruption.
|
||
await _persist_dispatch_checkpoint(adapter, state)
|
||
await _ensure_selected_seat_threads(adapter, state)
|
||
state.mark_completed(executor._node_id)
|
||
await _persist_dispatch_checkpoint(adapter, state)
|
||
await _emit_dispatch_event(adapter, state)
|
||
await ctx.send_message({
|
||
"decision": executor._dispatch_decision(),
|
||
"agents": list(state.selected_seat_agent_ids),
|
||
"execution_mode": plan.execution_mode,
|
||
})
|
||
|
||
|
||
# ── Seat executor ───────────────────────────────────────────────────────────
|
||
|
||
|
||
class SeatExecutor(Executor):
|
||
"""Runs one seat turn via DeerFlowAgentAdapter.
|
||
|
||
Phase 3: failures/timeouts do NOT raise through the graph — they are
|
||
recorded in ``seat_outcomes`` so fan-in / cross-review can proceed.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
adapter: Any,
|
||
state: RoundtableState,
|
||
agent_id: str,
|
||
*,
|
||
action_coordinator: ActionCoordinator | None = None,
|
||
) -> None:
|
||
super().__init__(id=f"seat_{agent_id}")
|
||
self._adapter = adapter
|
||
self._state = state
|
||
self._agent_id = agent_id
|
||
self._action_coordinator = action_coordinator
|
||
|
||
@handler(input=Any, output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
node_id = self.id
|
||
if self._state.is_completed(node_id):
|
||
prior = self._state.seat_outcomes.get(self._agent_id) or {"status": "completed", "replayed": True}
|
||
await ctx.send_message({"agent_id": self._agent_id, **prior, "replayed": True})
|
||
return
|
||
|
||
selected = self._state.selected_seat_agent_ids
|
||
if selected is not None and self._agent_id not in selected:
|
||
self._state.seat_outcomes[self._agent_id] = {"status": "skipped", "reason": "not_selected"}
|
||
self._state.mark_completed(node_id)
|
||
await ctx.send_message({
|
||
"agent_id": self._agent_id,
|
||
"status": "skipped",
|
||
"reason": "not_selected",
|
||
})
|
||
return
|
||
|
||
thread_id = self._state.thread_map.get(self._agent_id, "")
|
||
if not thread_id:
|
||
self._record_failure("no_thread", f"no thread for agent {self._agent_id}")
|
||
await ctx.send_message({"agent_id": self._agent_id, "error": "no_thread", "status": "failed"})
|
||
return
|
||
|
||
prior_turns = [
|
||
{"agent_id": r["agent_id"], "turn_index": r["turn_index"], "summary": r.get("content", "")[:500]}
|
||
for r in self._state.turn_results
|
||
]
|
||
caps = _agent_capabilities_from_snapshot(self._state.capability_snapshot, self._agent_id)
|
||
excluded = list(caps.get("excluded_tools", []))
|
||
|
||
assignment = {}
|
||
if isinstance(self._state.plan, dict):
|
||
assignment = (self._state.plan.get("assignments") or {}).get(self._agent_id) or {}
|
||
stage_instruction = str(
|
||
assignment.get("instruction")
|
||
or f"第{self._state.current_round}轮分析。{seat_output_instruction()}"
|
||
)
|
||
role_description = str(assignment.get("role") or "")
|
||
|
||
try:
|
||
response = await asyncio.wait_for(
|
||
self._adapter.run_seat_turn(
|
||
session_id=self._state.session_id,
|
||
run_id=self._state.run_id,
|
||
agent_id=self._agent_id,
|
||
turn_index=self._state.current_round,
|
||
execution_stage="group_chat" if self._state.mode == "group_chat" else "expert",
|
||
message=self._state.task_text,
|
||
thread_id=thread_id,
|
||
model_name=caps.get("model_name") or self._state.model_name,
|
||
reasoning_effort=caps.get("reasoning_effort") or self._state.reasoning_effort,
|
||
thinking_enabled=caps.get("thinking_enabled", True),
|
||
subagent_enabled=caps.get("subagent_enabled", False),
|
||
task_text=self._state.task_text,
|
||
stage_instruction=stage_instruction,
|
||
role_description=role_description,
|
||
prior_turns=prior_turns,
|
||
fact_ledger=self._state.fact_ledger,
|
||
excluded_tools=excluded,
|
||
memory_recall_disabled=caps.get("memory_recall_disabled"),
|
||
memory_injection_enabled=caps.get("memory_injection_enabled"),
|
||
capability_snapshot=self._state.capability_snapshot,
|
||
write_policy=self._state.write_policy,
|
||
wait_timeout_seconds=self._state.seat_timeout_seconds,
|
||
),
|
||
timeout=self._state.seat_timeout_seconds,
|
||
)
|
||
except TimeoutError:
|
||
self._record_failure("timeout", f"seat {self._agent_id} timed out")
|
||
await ctx.send_message({"agent_id": self._agent_id, "status": "timeout", "error": "timeout"})
|
||
return
|
||
except Exception as exc: # noqa: BLE001
|
||
self._record_failure("error", f"{type(exc).__name__}")
|
||
await ctx.send_message({"agent_id": self._agent_id, "status": "failed", "error": type(exc).__name__})
|
||
return
|
||
|
||
if response.error:
|
||
self._record_failure("error", str(response.error)[:200])
|
||
# Still capture any partial content / tool calls below.
|
||
|
||
raw_content = response.content or ""
|
||
schema_result = await self._validate_output(raw_content)
|
||
content = schema_result.get("raw_content") or raw_content
|
||
|
||
if schema_result.get("output"):
|
||
merge_seat_output(self._state.fact_ledger, schema_result["output"], agent_id=self._agent_id)
|
||
|
||
await self._capture_pending_approvals(response.tool_calls or [])
|
||
|
||
if response.tool_calls:
|
||
for tc in response.tool_calls:
|
||
tc_name = tc.get("name", "")
|
||
is_intercepted = any(
|
||
a.get("tool_name") == tc_name for a in (self._state.pending_approvals or [])
|
||
)
|
||
record_tool_result(
|
||
self._state.fact_ledger,
|
||
tool_name=tc_name,
|
||
status="waiting_approval" if is_intercepted else "completed",
|
||
agent_id=self._agent_id,
|
||
turn_index=self._state.current_round,
|
||
)
|
||
|
||
turn_row = {
|
||
"agent_id": self._agent_id,
|
||
"turn_index": self._state.current_round,
|
||
# Phase-3 P0-1: record which execution stage produced this turn so
|
||
# recovery can distinguish an expert's original output from its
|
||
# format_repair / cross_review / finalizer re-runs.
|
||
"execution_stage": "group_chat" if self._state.mode == "group_chat" else "expert",
|
||
"content": content,
|
||
"tool_calls": response.tool_calls,
|
||
"error": response.error,
|
||
"usage": response.usage,
|
||
"schema_status": schema_result.get("schema_status"),
|
||
"schema_errors": schema_result.get("errors") or [],
|
||
"structured_output": schema_result.get("output"),
|
||
}
|
||
self._state.turn_results.append(turn_row)
|
||
|
||
status = "failed" if response.error else "completed"
|
||
if schema_result.get("schema_status") == "schema_invalid":
|
||
# Invalid schema does not by itself fail the seat — content is kept.
|
||
pass
|
||
self._state.seat_outcomes[self._agent_id] = {
|
||
"status": status,
|
||
"schema_status": schema_result.get("schema_status"),
|
||
"error": response.error,
|
||
}
|
||
if status == "failed":
|
||
self._state.partial_failure = True
|
||
|
||
self._state.mark_completed(node_id)
|
||
await ctx.send_message({
|
||
"agent_id": self._agent_id,
|
||
"content": content,
|
||
"error": response.error,
|
||
"status": status,
|
||
"schema_status": schema_result.get("schema_status"),
|
||
})
|
||
|
||
async def _validate_output(self, raw_content: str) -> dict[str, Any]:
|
||
if self._state.mode != "structured":
|
||
return {
|
||
"schema_status": "skipped",
|
||
"output": None,
|
||
"errors": [],
|
||
"raw_content": raw_content,
|
||
"repair_attempted": False,
|
||
}
|
||
|
||
repair_runner = None
|
||
if self._state.schema_repair_enabled:
|
||
async def _repair(prompt: str) -> str:
|
||
thread_id = self._state.thread_map.get(self._agent_id, "")
|
||
resp = await self._adapter.run_seat_turn(
|
||
session_id=self._state.session_id,
|
||
run_id=self._state.run_id,
|
||
agent_id=self._agent_id,
|
||
turn_index=self._state.current_round,
|
||
execution_stage="format_repair",
|
||
message=prompt,
|
||
thread_id=thread_id,
|
||
model_name=self._state.model_name,
|
||
reasoning_effort=self._state.reasoning_effort,
|
||
task_text=prompt,
|
||
stage_instruction="格式修复（仅一次）",
|
||
fact_ledger=self._state.fact_ledger,
|
||
capability_snapshot=self._state.capability_snapshot,
|
||
write_policy=self._state.write_policy,
|
||
wait_timeout_seconds=self._state.seat_timeout_seconds,
|
||
)
|
||
return resp.content or ""
|
||
|
||
repair_runner = _repair
|
||
|
||
return await validate_or_repair_seat_output(raw_content=raw_content, repair_runner=repair_runner)
|
||
|
||
async def _capture_pending_approvals(self, tool_calls: list[dict[str, Any]]) -> None:
|
||
if not tool_calls or self._state.write_policy == "auto":
|
||
return
|
||
pending_calls = [
|
||
tc for tc in tool_calls
|
||
if tc.get("name") and tool_needs_approval(
|
||
tc.get("name"), self._state.write_policy,
|
||
self._state.capability_snapshot, self._agent_id,
|
||
)
|
||
]
|
||
if not pending_calls:
|
||
return
|
||
|
||
all_actions = await self._store.actions.list_actions(
|
||
run_id=self._state.run_id, status="waiting_approval"
|
||
)
|
||
turn_actions = [
|
||
a for a in all_actions
|
||
if a.get("agent_id") == self._agent_id
|
||
and a.get("turn_index") == self._state.current_round
|
||
]
|
||
matched: list[dict[str, Any]] = []
|
||
for tc in pending_calls:
|
||
tc_id = tc.get("id") or ""
|
||
tc_name = tc.get("name", "")
|
||
if tc_id:
|
||
hits = [a for a in turn_actions if (a.get("logical_action_id") or "") == tc_id]
|
||
else:
|
||
hits = [
|
||
a for a in turn_actions
|
||
if a.get("tool_name") == tc_name and not a.get("logical_action_id")
|
||
]
|
||
if len(hits) > 1:
|
||
raise RuntimeError(
|
||
f"seat {self._agent_id} turn {self._state.current_round}: "
|
||
f"pending tool call '{tc_name}' (id={tc_id or '<none>'}) matched "
|
||
f"{len(hits)} waiting actions (expected exactly 1) — refusing to guess"
|
||
)
|
||
if len(hits) == 0:
|
||
logger.warning(
|
||
"seat %s turn %s: pending tool call '%s' (id=%s) classified as "
|
||
"needing approval but matched 0 waiting actions — tool may be "
|
||
"unclassified or executed as read by WriteGuard; skipping",
|
||
self._agent_id, self._state.current_round, tc_name, tc_id or "<none>",
|
||
)
|
||
continue
|
||
matched.append(hits[0])
|
||
|
||
# Merge into pending_approvals (multi-seat).
|
||
existing_ids = {p.get("action_id") for p in self._state.pending_approvals}
|
||
for a in matched:
|
||
if a["id"] in existing_ids:
|
||
continue
|
||
self._state.pending_approvals.append({
|
||
"action_id": a["id"],
|
||
"approval_id": a.get("approval_id", ""),
|
||
"agent_id": self._agent_id,
|
||
"tool_name": a.get("tool_name", ""),
|
||
"turn_index": self._state.current_round,
|
||
})
|
||
# Phase-3 P1-3: mirror the pending external write into the ledger so
|
||
# the approval lifecycle is fully traceable there. This row moves
|
||
# to executed/failed/rejected only via record_action_transition in
|
||
# WorkflowRunner.resume — it must not be left dangling.
|
||
record_action_transition(
|
||
self._state.fact_ledger,
|
||
action_id=a["id"],
|
||
tool_name=a.get("tool_name", ""),
|
||
status="pending",
|
||
agent_id=self._agent_id,
|
||
)
|
||
|
||
def _record_failure(self, status: str, detail: str) -> None:
|
||
self._state.partial_failure = True
|
||
self._state.seat_outcomes[self._agent_id] = {"status": status, "error": detail}
|
||
self._state.turn_results.append({
|
||
"agent_id": self._agent_id,
|
||
"turn_index": self._state.current_round,
|
||
"content": "",
|
||
"tool_calls": [],
|
||
"error": detail,
|
||
"usage": None,
|
||
"schema_status": None,
|
||
})
|
||
self._state.mark_completed(self.id)
|
||
|
||
@property
|
||
def _store(self) -> Any:
|
||
return self._adapter._service.store
|
||
|
||
|
||
# ── Cross-review (structured) ───────────────────────────────────────────────
|
||
|
||
|
||
class CrossReviewExecutor(Executor):
|
||
"""Barrier after parallel experts: review evidence, conflicts, tool/write status."""
|
||
|
||
def __init__(self, adapter: Any, state: RoundtableState) -> None:
|
||
super().__init__(id="cross_review")
|
||
self._adapter = adapter
|
||
self._state = state
|
||
|
||
@handler(input=Any, output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
if self._state.is_completed("cross_review"):
|
||
await ctx.send_message({"decision": "reviewed", "replayed": True})
|
||
return
|
||
|
||
self._state.stage = "review"
|
||
|
||
# Barrier: ensure every seat has an outcome (fan-in should guarantee this).
|
||
review_seats = self._state.selected_seat_agent_ids or self._state.seat_agent_ids
|
||
missing = [a for a in review_seats if a not in self._state.seat_outcomes]
|
||
for aid in missing:
|
||
self._state.seat_outcomes[aid] = {"status": "missing", "error": "no_outcome"}
|
||
self._state.partial_failure = True
|
||
|
||
failed_seats = [
|
||
a for a, o in self._state.seat_outcomes.items()
|
||
if isinstance(o, dict) and o.get("status") in ("failed", "timeout", "missing", "error")
|
||
]
|
||
tool_failures = [
|
||
tr for tr in self._state.fact_ledger.get("tool_results") or []
|
||
if isinstance(tr, dict) and tr.get("status") in ("failed", "unknown")
|
||
]
|
||
pending_writes = list(self._state.fact_ledger.get("pending_actions") or [])
|
||
# Also count pending approvals still in workflow state.
|
||
if self._state.pending_approvals:
|
||
pending_writes = pending_writes or list(self._state.pending_approvals)
|
||
|
||
review: dict[str, Any] = {
|
||
"summary": "交叉审查完成",
|
||
"failed_seats": failed_seats,
|
||
"tool_failures": tool_failures,
|
||
"pending_writes": len(self._state.pending_approvals),
|
||
"conflicts": list(self._state.fact_ledger.get("conflicts") or []),
|
||
"partial_failure": self._state.partial_failure,
|
||
}
|
||
|
||
manager_id = self._state.manager_agent_id
|
||
thread_id = self._state.thread_map.get(manager_id or "", "")
|
||
if manager_id and thread_id:
|
||
try:
|
||
prompt = build_manager_synthesis_input(
|
||
task_text=self._state.task_text,
|
||
seat_results=[r for r in self._state.turn_results if r.get("agent_id") in review_seats],
|
||
fact_ledger=self._state.fact_ledger,
|
||
)
|
||
# Surface the candidate ids + the evidence refs the ledger has
|
||
# backing for, so the manager can confirm by id rather than
|
||
# re-asserting facts in prose. Phase-3 P1-1: confirmations must
|
||
# bind evidence; only listed evidence refs may be cited.
|
||
candidates = self._state.fact_ledger.get("candidate_facts") or []
|
||
candidate_ids = [c.get("candidate_id") for c in candidates if isinstance(c, dict) and c.get("candidate_id")]
|
||
from app.maf_roundtable.fact_ledger import known_evidence_refs
|
||
valid_refs = sorted(known_evidence_refs(self._state.fact_ledger))
|
||
prompt = (
|
||
"【交叉审查】请检查证据是否充分、冲突、遗漏、工具失败与外部写入状态。"
|
||
"只输出一个 JSON 对象，字段：summary(必填)、evidence_ok(bool,默认false)、"
|
||
"gaps(string[])、conflicts(string[])、tool_failures(string[])、"
|
||
"write_status_notes(string[])、recommendations(string[])、"
|
||
"confirmed_candidates(对象数组，每项 {candidate_id, evidence_refs:string[]}）。"
|
||
"evidence_ok 只在确有充分证据时设 true。"
|
||
"confirmed_candidates.candidate_id 只能从下列候选 id 中选择："
|
||
f"{candidate_ids[:200]}；evidence_refs 只能从下列已记录的证据引用中选择："
|
||
f"{valid_refs[:200]}；无证据支撑的候选不得确认。\n\n" + prompt
|
||
)
|
||
response = await asyncio.wait_for(
|
||
self._adapter.run_seat_turn(
|
||
session_id=self._state.session_id,
|
||
run_id=self._state.run_id,
|
||
agent_id=manager_id,
|
||
turn_index=self._state.current_round + 1,
|
||
execution_stage="cross_review",
|
||
message=prompt,
|
||
thread_id=thread_id,
|
||
model_name=self._state.model_name,
|
||
reasoning_effort=self._state.reasoning_effort,
|
||
task_text=prompt,
|
||
stage_instruction="交叉审查",
|
||
fact_ledger=self._state.fact_ledger,
|
||
capability_snapshot=self._state.capability_snapshot,
|
||
write_policy=self._state.write_policy,
|
||
wait_timeout_seconds=self._state.seat_timeout_seconds,
|
||
),
|
||
timeout=self._state.seat_timeout_seconds,
|
||
)
|
||
raw = response.content or ""
|
||
await self._apply_cross_review_output(review, raw, manager_id)
|
||
except Exception as exc: # noqa: BLE001
|
||
self._apply_deterministic_degradation(review, type(exc).__name__)
|
||
else:
|
||
self._apply_deterministic_degradation(review, None)
|
||
if failed_seats:
|
||
review["gaps"] = [f"席位失败: {', '.join(failed_seats)}"]
|
||
|
||
# The quality gate is server-owned. A manager may recommend acceptance,
|
||
# but gaps/failures/schema degradation always become a controlled gap
|
||
# and are carried into the final result instead of being labelled as a
|
||
# verified fact.
|
||
has_quality_gap = bool(
|
||
review.get("partial_failure")
|
||
or review.get("failed_seats")
|
||
or review.get("gaps")
|
||
or review.get("conflicts")
|
||
or review.get("tool_failures")
|
||
or review.get("schema_status") == "schema_invalid"
|
||
)
|
||
review["decision"] = "controlled_gap" if has_quality_gap else "accept"
|
||
review["reason"] = (
|
||
"required evidence or seat output is incomplete"
|
||
if has_quality_gap
|
||
else "required seat deliveries passed deterministic checks"
|
||
)
|
||
self._state.cross_review_result = review
|
||
# Surface unresolved conflicts from review notes.
|
||
for c in review.get("conflicts") or []:
|
||
if c and c not in self._state.fact_ledger["conflicts"]:
|
||
self._state.fact_ledger["conflicts"].append(c)
|
||
|
||
await self._adapter._emit_critical(self._state.run_id, {
|
||
"event_type": "manager.review.decision",
|
||
"agent_id": self._state.manager_agent_id or "",
|
||
"turn_index": self._state.current_round,
|
||
"payload": {
|
||
"decision": review["decision"],
|
||
"reason": review["reason"],
|
||
"quality": {
|
||
"failed_seat_count": len(review.get("failed_seats") or []),
|
||
"gap_count": len(review.get("gaps") or []),
|
||
"conflict_count": len(review.get("conflicts") or []),
|
||
},
|
||
},
|
||
})
|
||
|
||
self._state.mark_completed("cross_review")
|
||
self._state.stage = "approval"
|
||
await ctx.send_message({"decision": "reviewed", "review": review})
|
||
|
||
async def _apply_cross_review_output(self, review: dict[str, Any], raw: str, manager_id: str) -> None:
|
||
"""Parse + validate the manager's cross-review output.
|
||
|
||
On success, write the validated fields and confirm candidate facts. On
|
||
schema failure, run one dedicated ``cross_review_repair`` seat turn; if
|
||
that still fails, fall back to a deterministic review (never paste the
|
||
raw JSON into ``summary``).
|
||
|
||
Phase-3 P1-1: confirmation is evidence-bound. The model supplies
|
||
``confirmed_candidates`` (``[{candidate_id, evidence_refs}]``); the
|
||
ledger only promotes a candidate whose evidence resolves to entries it
|
||
actually recorded (see :func:`fact_ledger.confirm_candidates`).
|
||
``evidence_ok`` defaults to ``False`` so a model that omits it never
|
||
silently passes the evidence gate.
|
||
"""
|
||
model, errors = parse_cross_review_output(raw)
|
||
if model is not None:
|
||
self._write_validated_review(review, model, source="manager", schema_status="ok")
|
||
self._confirm_reviewed_candidates(model, manager_id, review)
|
||
return
|
||
|
||
# Schema invalid — one repair attempt.
|
||
repaired_model = await self._run_cross_review_repair(raw, errors, manager_id)
|
||
if repaired_model is not None:
|
||
self._write_validated_review(review, repaired_model, source="manager", schema_status="repaired")
|
||
self._confirm_reviewed_candidates(repaired_model, manager_id, review)
|
||
return
|
||
|
||
self._apply_deterministic_degradation(review, None)
|
||
|
||
def _confirm_reviewed_candidates(self, model: Any, manager_id: str, review: dict[str, Any]) -> None:
|
||
"""Promote only evidence-bound confirmed candidates; record rejections.
|
||
|
||
Phase-3 P1-2: if the model says evidence is NOT sufficient
|
||
(``evidence_ok=False``), no candidate may be promoted regardless of
|
||
what ``confirmed_candidates`` it supplies. The attempted confirmations
|
||
are recorded as rejections for audit.
|
||
"""
|
||
if not getattr(model, "evidence_ok", False):
|
||
attempted = [
|
||
c.candidate_id
|
||
for c in (getattr(model, "confirmed_candidates", None) or [])
|
||
]
|
||
if attempted:
|
||
review["confirmation_rejections"] = [
|
||
{
|
||
"candidate_id": candidate_id,
|
||
"reason": "review_did_not_confirm_evidence",
|
||
}
|
||
for candidate_id in attempted
|
||
]
|
||
return
|
||
|
||
rejections: list[dict[str, Any]] = []
|
||
accepted: list[dict[str, Any]] = []
|
||
if getattr(model, "confirmed_candidates", None):
|
||
confirm_candidates(
|
||
self._state.fact_ledger,
|
||
confirmed_candidates=[
|
||
{"candidate_id": c.candidate_id, "evidence_refs": list(c.evidence_refs or [])}
|
||
for c in model.confirmed_candidates
|
||
],
|
||
confirmed_by=manager_id,
|
||
rejections=rejections,
|
||
accepted=accepted,
|
||
)
|
||
elif getattr(model, "confirmed_candidate_ids", None):
|
||
from app.maf_roundtable.fact_ledger import confirm_candidate_facts
|
||
|
||
confirm_candidate_facts(
|
||
self._state.fact_ledger,
|
||
confirmed_candidate_ids=list(model.confirmed_candidate_ids or []),
|
||
confirmed_by=manager_id,
|
||
evidence_refs=[],
|
||
accepted=accepted,
|
||
)
|
||
# ``confirmed_candidates`` is an API-facing success field. Populate it
|
||
# only from the server-validated promotions, never from the model's
|
||
# request. The raw model request remains in ``requested_confirmations``
|
||
# for audit/debugging.
|
||
review["confirmed_candidates"] = accepted
|
||
review["confirmed_candidate_ids"] = [item["candidate_id"] for item in accepted]
|
||
if rejections:
|
||
review["confirmation_rejections"] = rejections
|
||
|
||
async def _run_cross_review_repair(self, raw: str, errors: list[str], manager_id: str) -> Any:
|
||
"""One-shot repair turn for an invalid CrossReviewOutput.
|
||
|
||
Returns the validated model or ``None``. Runs the repair through the
|
||
DeerFlow adapter so it executes under the full WriteGuard /
|
||
approval regime (a repair turn can itself produce tool calls).
|
||
"""
|
||
thread_id = self._state.thread_map.get(manager_id or "", "")
|
||
if not (manager_id and thread_id and self._state.schema_repair_enabled):
|
||
return None
|
||
repair_prompt = build_repair_prompt(raw, errors, schema_name="CrossReviewOutput")
|
||
try:
|
||
response = await asyncio.wait_for(
|
||
self._adapter.run_seat_turn(
|
||
session_id=self._state.session_id,
|
||
run_id=self._state.run_id,
|
||
agent_id=manager_id,
|
||
turn_index=self._state.current_round + 1,
|
||
execution_stage="cross_review_repair",
|
||
message=repair_prompt,
|
||
thread_id=thread_id,
|
||
model_name=self._state.model_name,
|
||
reasoning_effort=self._state.reasoning_effort,
|
||
task_text=repair_prompt,
|
||
stage_instruction="交叉审查格式修复（仅一次）",
|
||
fact_ledger=self._state.fact_ledger,
|
||
capability_snapshot=self._state.capability_snapshot,
|
||
write_policy=self._state.write_policy,
|
||
wait_timeout_seconds=self._state.seat_timeout_seconds,
|
||
),
|
||
timeout=self._state.seat_timeout_seconds,
|
||
)
|
||
except Exception: # noqa: BLE001
|
||
return None
|
||
model, _errors = parse_cross_review_output(response.content or "")
|
||
return model
|
||
|
||
def _write_validated_review(
|
||
self, review: dict[str, Any], model: Any, *, source: str, schema_status: str
|
||
) -> None:
|
||
review["source"] = source
|
||
review["schema_status"] = schema_status
|
||
review["summary"] = model.summary
|
||
review["gaps"] = list(model.gaps or [])
|
||
review["conflicts"] = list(model.conflicts or [])
|
||
review["tool_failures"] = list(model.tool_failures or [])
|
||
review["write_status_notes"] = list(model.write_status_notes or [])
|
||
review["recommendations"] = list(model.recommendations or [])
|
||
review["evidence_ok"] = bool(model.evidence_ok)
|
||
# Preserve the model's request for audit, but do not label it as an
|
||
# accepted confirmation. _confirm_reviewed_candidates fills the latter
|
||
# only after the evidence and candidate-ID checks pass server-side.
|
||
review["requested_confirmations"] = [
|
||
{
|
||
"candidate_id": item.candidate_id,
|
||
"evidence_refs": list(item.evidence_refs),
|
||
}
|
||
for item in (getattr(model, "confirmed_candidates", None) or [])
|
||
]
|
||
review["requested_confirmations"].extend(
|
||
{
|
||
"candidate_id": candidate_id,
|
||
"evidence_refs": [],
|
||
}
|
||
for candidate_id in (model.confirmed_candidate_ids or [])
|
||
)
|
||
review["confirmed_candidates"] = []
|
||
review["confirmed_candidate_ids"] = []
|
||
|
||
def _apply_deterministic_degradation(self, review: dict[str, Any], error: str | None) -> None:
|
||
"""Deterministic fallback when the manager output is unparseable.
|
||
|
||
Never pastes the raw model JSON into ``summary``; instead marks the
|
||
review as deterministic so downstream knows it is not model-derived.
|
||
"""
|
||
review["source"] = "deterministic"
|
||
review["schema_status"] = "schema_invalid"
|
||
review["raw_response_ignored"] = True
|
||
if error:
|
||
review["manager_error"] = error
|
||
|
||
|
||
# ── Approval gate ───────────────────────────────────────────────────────────
|
||
|
||
|
||
class ApprovalGateExecutor(Executor):
|
||
"""Pauses for external-write approval before proceeding."""
|
||
|
||
def __init__(self, adapter: Any, state: RoundtableState, service: Any) -> None:
|
||
super().__init__(id="approval_gate")
|
||
self._adapter = adapter
|
||
self._state = state
|
||
self._service = service
|
||
|
||
@handler(input=Any, output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
if self._state.pending_approvals:
|
||
all_decided = True
|
||
for pa in self._state.pending_approvals:
|
||
approval_id = pa.get("approval_id", "")
|
||
approval = await self._service.store.approvals.get_approval(approval_id)
|
||
if approval is None or approval["status"] == "pending":
|
||
all_decided = False
|
||
continue
|
||
if approval["status"] == "approved":
|
||
from app.maf_roundtable.event_normalizer import make_action_approved
|
||
await self._adapter._emit_critical(self._state.run_id, make_action_approved(
|
||
pa.get("agent_id", ""),
|
||
self._state.current_round,
|
||
action_id=pa.get("action_id", ""),
|
||
approval_id=approval_id,
|
||
decided_by=approval.get("decided_by", ""),
|
||
))
|
||
elif approval["status"] == "rejected":
|
||
from app.maf_roundtable.event_normalizer import make_action_rejected
|
||
await self._adapter._emit_critical(self._state.run_id, make_action_rejected(
|
||
pa.get("agent_id", ""),
|
||
self._state.current_round,
|
||
action_id=pa.get("action_id", ""),
|
||
approval_id=approval_id,
|
||
decided_by=approval.get("decided_by", ""),
|
||
))
|
||
|
||
if all_decided:
|
||
self._state.pending_approvals = []
|
||
self._state.mark_completed("approval_gate")
|
||
await ctx.send_message({"decision": "approved"})
|
||
return
|
||
|
||
await ctx.send_message({"decision": "waiting_approval"})
|
||
return
|
||
|
||
needs_approval = self._check_needs_approval()
|
||
if not needs_approval:
|
||
self._state.mark_completed("approval_gate")
|
||
await ctx.send_message({"decision": "approved", "policy": self._state.write_policy})
|
||
return
|
||
|
||
if not self._state.pending_approvals:
|
||
raise WriteGuardConfigurationError(
|
||
f"Write policy '{self._state.write_policy}' requires approval "
|
||
f"for detected write tools, but WriteGuard did not create any "
|
||
f"pending actions. Check that write_guard.enabled is true and "
|
||
f"the provider is configured in config.yaml."
|
||
)
|
||
|
||
await self._service.store.runs.update_run(self._state.run_id, {"status": "waiting_approval"})
|
||
await self._service.store.sessions.update_session(
|
||
self._state.session_id, user_id=self._adapter._user_id, fields={"status": "waiting_approval"}
|
||
)
|
||
self._state.status = "waiting_approval"
|
||
await ctx.send_message({"decision": "waiting_approval"})
|
||
|
||
def _check_needs_approval(self) -> bool:
|
||
if self._state.write_policy == "auto":
|
||
return False
|
||
# pending_approvals is the authoritative source: it is populated by
|
||
# SeatExecutor._capture_pending_approvals which only adds tool calls
|
||
# that WriteGuard actually intercepted (i.e. have a database action).
|
||
# Tools that were classified as needing approval but had 0 matching
|
||
# actions (unclassified/hallucinated/read tools) are intentionally
|
||
# skipped and must NOT re-trigger the gate here.
|
||
return bool(self._state.pending_approvals)
|
||
|
||
|
||
# ── Finalizer ───────────────────────────────────────────────────────────────
|
||
|
||
|
||
async def generate_final_report(adapter: Any, state: RoundtableState) -> dict[str, Any]:
|
||
"""Run the manager synthesis and set ``state.final_result``."""
|
||
from app.maf_roundtable.event_normalizer import make_run_completed
|
||
|
||
manager_id = state.manager_agent_id or "coordinator"
|
||
thread_id = state.thread_map.get(manager_id, "")
|
||
|
||
usable = [
|
||
r for r in state.turn_results
|
||
if not r.get("error") or r.get("content")
|
||
]
|
||
if state.partial_failure:
|
||
usable = [r for r in usable if (state.seat_outcomes.get(r.get("agent_id")) or {}).get("status") == "completed"]
|
||
|
||
synthesis_input = build_manager_synthesis_input(
|
||
task_text=state.task_text,
|
||
seat_results=usable or state.turn_results,
|
||
fact_ledger=state.fact_ledger,
|
||
)
|
||
if state.partial_failure:
|
||
failed = [
|
||
a for a, o in state.seat_outcomes.items()
|
||
if isinstance(o, dict) and o.get("status") not in ("completed", None)
|
||
]
|
||
synthesis_input = (
|
||
f"【降级综合】以下席位失败/超时，请仅基于可用结果与事实账本总结："
|
||
f"{', '.join(failed)}\n\n" + synthesis_input
|
||
)
|
||
if state.cross_review_result:
|
||
synthesis_input += f"\n\n【交叉审查摘要】\n{state.cross_review_result.get('summary', '')[:2000]}"
|
||
|
||
if thread_id:
|
||
response = await adapter.run_seat_turn(
|
||
session_id=state.session_id,
|
||
run_id=state.run_id,
|
||
agent_id=manager_id,
|
||
turn_index=state.current_round + 1,
|
||
execution_stage="finalizer",
|
||
message=synthesis_input,
|
||
thread_id=thread_id,
|
||
model_name=state.model_name,
|
||
reasoning_effort=state.reasoning_effort,
|
||
task_text=synthesis_input,
|
||
stage_instruction="综合所有席位分析，生成最终结论",
|
||
fact_ledger=state.fact_ledger,
|
||
capability_snapshot=state.capability_snapshot,
|
||
write_policy=state.write_policy,
|
||
wait_timeout_seconds=state.seat_timeout_seconds,
|
||
)
|
||
state.final_result = {
|
||
"content": response.content,
|
||
"agent_id": manager_id,
|
||
"partial_failure": state.partial_failure,
|
||
"degraded": state.partial_failure,
|
||
"quality_decision": (
|
||
(state.cross_review_result or {}).get("decision")
|
||
or ("controlled_gap" if state.partial_failure else "accept")
|
||
),
|
||
}
|
||
else:
|
||
state.final_result = {
|
||
"content": synthesis_input,
|
||
"agent_id": manager_id,
|
||
"note": "no_manager_thread",
|
||
"partial_failure": state.partial_failure,
|
||
"degraded": state.partial_failure,
|
||
"quality_decision": (
|
||
(state.cross_review_result or {}).get("decision")
|
||
or ("controlled_gap" if state.partial_failure else "accept")
|
||
),
|
||
}
|
||
|
||
await adapter._emit_critical(state.run_id, make_run_completed(manager_id, state.current_round + 1, {
|
||
"total_rounds": state.current_round,
|
||
"seats_completed": len([
|
||
a for a, o in state.seat_outcomes.items()
|
||
if isinstance(o, dict) and o.get("status") == "completed"
|
||
]),
|
||
"partial_failure": state.partial_failure,
|
||
}))
|
||
return state.final_result
|
||
|
||
|
||
class FinalizerExecutor(Executor):
|
||
"""Generates the final synthesis and ends the run."""
|
||
|
||
def __init__(self, adapter: Any, state: RoundtableState) -> None:
|
||
super().__init__(id="finalizer")
|
||
self._adapter = adapter
|
||
self._state = state
|
||
|
||
@handler(input=Any, workflow_output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
if self._state.is_completed("finalizer") and self._state.final_result is not None:
|
||
self._state.status = "completed"
|
||
await ctx.yield_output({"final_result": self._state.final_result, "replayed": True})
|
||
return
|
||
|
||
self._state.stage = "finalizing"
|
||
await generate_final_report(self._adapter, self._state)
|
||
self._state.status = "completed"
|
||
self._state.stage = "done"
|
||
self._state.mark_completed("finalizer")
|
||
await ctx.yield_output({"final_result": self._state.final_result})
|
||
|
||
|
||
class WaitingTerminalExecutor(Executor):
|
||
"""Terminal executor for the approval-waiting path."""
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__(id="waiting_terminal")
|
||
|
||
@handler(input=Any, workflow_output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
await ctx.yield_output({"status": "waiting_approval"})
|
||
|
||
|
||
# ── helpers ─────────────────────────────────────────────────────────────────
|
||
|
||
|
||
def _agent_capabilities_from_snapshot(snapshot: dict[str, Any] | None, agent_id: str) -> dict[str, Any]:
|
||
if not snapshot:
|
||
return {}
|
||
agents = snapshot.get("agents")
|
||
if not isinstance(agents, list):
|
||
return {}
|
||
for a in agents:
|
||
if isinstance(a, dict) and a.get("agent_id") == agent_id:
|
||
return {
|
||
"model_name": a.get("model_name"),
|
||
"reasoning_effort": a.get("reasoning_effort"),
|
||
"thinking_enabled": a.get("thinking_enabled", True),
|
||
"subagent_enabled": a.get("subagent_enabled", False),
|
||
"excluded_tools": a.get("excluded_tools") or [],
|
||
"memory_recall_disabled": a.get("memory_recall_disabled"),
|
||
"memory_injection_enabled": a.get("memory_injection_enabled"),
|
||
"write_policy": a.get("write_policy"),
|
||
}
|
||
return {}
|
||
|
||
|
||
def _is_high_risk_tool(tool_name: str, snapshot: dict[str, Any] | None = None) -> bool:
|
||
side_effect, risk_level = classify_tool(tool_name, snapshot)
|
||
if side_effect in ("read", "skill"):
|
||
return False
|
||
risk_order = {"R0": 0, "R1": 1, "R2": 2, "R3": 3, "R4": 4}
|
||
return risk_order.get(risk_level, 3) >= 3
|