"""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 ''}) 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 "", ) 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