董宏凯
1aeee86984
P0-1 migration: detect the same-named stage-3 execution_stage CHECK by sqltext (must contain rework + manager_review), SQLite rebuild / PG-MySQL drop+add, PG boolean default TRUE, validate_schema checks constraint content + seat_tasks status CHECK/unique/FKs; tests start from a real stage-3 schema. P0-2: a missing/invalid manager review (timeout/exception/empty/invalid JSON/missing per-seat decision) is never an acceptance - system rework under the cap, contract fallback (reassign/controlled_gap/ask_user) afterwards, source=system events. P0-3: unaccepted deliveries are isolated from the fact ledger, cross-review and final report (deferred merge, delivery_json envelope, accepted_task_results in checkpoint, accepted-only synthesis, no turn_results fallback). P0-4: seat-task init is atomic get-or-create over the full expected set; recovery fills missing tasks without overwriting frozen contracts; quality gate blocks missing required tasks. P1: zero config values honoured; duplicate detection persists SHA-256 conclusion digests in history (survives restarts); findings never satisfy evidence, evidence_refs must resolve to trusted ledger refs (unverified_evidence_ref); conversation SSE projection tests added and manager/finalizer messages are no longer mislabelled as seat.delivery.submitted.
2429 lines
107 KiB
Python
2429 lines
107 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.event_normalizer import (
|
||
make_manager_review_decision,
|
||
make_seat_delivery_submitted,
|
||
make_seat_reassigned,
|
||
make_seat_rework_assigned,
|
||
make_stage_quality_gate,
|
||
)
|
||
from app.maf_roundtable.fact_ledger import (
|
||
confirm_candidates,
|
||
empty_ledger,
|
||
known_evidence_refs,
|
||
merge_seat_output,
|
||
normalize_ledger,
|
||
record_action_transition,
|
||
record_tool_result,
|
||
)
|
||
from app.maf_roundtable.quality_contracts import (
|
||
OUTCOME_ACCEPTED,
|
||
OUTCOME_ASK_USER,
|
||
OUTCOME_CONTROLLED_GAP,
|
||
OUTCOME_REASSIGNED,
|
||
OUTCOME_REWORK,
|
||
TASK_ACCEPTED,
|
||
TASK_CONTROLLED_GAP,
|
||
TASK_EXHAUSTED,
|
||
TASK_REASSIGNED,
|
||
TASK_REWORK,
|
||
TASK_SUBMITTED,
|
||
TASK_TERMINAL_STATUSES,
|
||
MafTaskContract,
|
||
build_manager_review_prompt,
|
||
build_task_contract,
|
||
check_delivery_deterministic,
|
||
conclusion_digest,
|
||
delivery_from_seat_output,
|
||
finalize_review_decision,
|
||
make_controlled_gap,
|
||
parse_manager_review,
|
||
)
|
||
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,
|
||
quality_gate_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)
|
||
)
|
||
# Stage 4 (R3-Q): quality gate for V2 structured dispatch. Defaults to
|
||
# the dispatch-contract value when not explicitly resolved, so
|
||
# checkpoints written before Stage 4 keep the new behavior.
|
||
self.quality_gate_enabled = (
|
||
bool(self.dispatch_contract_enabled)
|
||
if quality_gate_enabled is None
|
||
else bool(quality_gate_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
|
||
# Stage 4 (R3-Q): per-task quality state machine summary.
|
||
# task_key → {status, agent_id, original_agent_id, attempt_count,
|
||
# rework_count, reassign_count, required}
|
||
self.seat_task_statuses: dict[str, Any] = {}
|
||
# Controlled-gap records to be carried into the final report's risk
|
||
# section. Each entry is a make_controlled_gap(...) dict.
|
||
self.controlled_gaps: list[dict[str, Any]] = []
|
||
# Stage 4 (R3-Q): the ONLY seat outputs that cross-review and the
|
||
# final synthesis may consume when the quality gate is enabled. An
|
||
# entry is appended exactly when a task's final decision is
|
||
# ``accepted`` — reworked / reassigned / gap deliveries never land
|
||
# here, so unaccepted content cannot leak into the final report.
|
||
self.accepted_task_results: list[dict[str, Any]] = []
|
||
|
||
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,
|
||
"quality_gate_enabled": self.quality_gate_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,
|
||
"seat_task_statuses": dict(self.seat_task_statuses),
|
||
"controlled_gaps": list(self.controlled_gaps),
|
||
"accepted_task_results": list(self.accepted_task_results),
|
||
}
|
||
|
||
@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"),
|
||
quality_gate_enabled=d.get("quality_gate_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.seat_task_statuses = dict(d.get("seat_task_statuses") or {})
|
||
state.controlled_gaps = list(d.get("controlled_gaps") or [])
|
||
state.accepted_task_results = list(d.get("accepted_task_results") or [])
|
||
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:
|
||
# V2 clarification checkpoints must always re-enter the clarification
|
||
# recovery branch, even when manager_dispatch is already marked
|
||
# completed — otherwise a recovered clarification run would stay
|
||
# "running" instead of landing in waiting_input.
|
||
if self._state.dispatch_contract_enabled and self._state.intent_type == "clarification":
|
||
self._state.stage = "planning"
|
||
await _resolve_manager_dispatch(self, ctx)
|
||
return
|
||
|
||
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,
|
||
execution_stage: str | None = None,
|
||
defer_fact_ledger_merge: bool = False,
|
||
) -> None:
|
||
super().__init__(id=f"seat_{agent_id}")
|
||
self._adapter = adapter
|
||
self._state = state
|
||
self._agent_id = agent_id
|
||
self._action_coordinator = action_coordinator
|
||
# Stage 4 (R3-Q): optional override for the execution_stage passed to
|
||
# run_seat_turn. When None, falls back to the mode-based default
|
||
# ("expert" for structured, "group_chat" for group_chat).
|
||
self._execution_stage = execution_stage
|
||
# Stage 4 (R3-Q): when True the structured output is validated, tool
|
||
# results are audited and events are recorded as usual, but the output
|
||
# is NOT merged into the fact ledger. The quality loop merges it only
|
||
# if the manager's final decision accepts the delivery — reworked,
|
||
# reassigned or gapped content must never become candidate facts.
|
||
self._defer_fact_ledger_merge = defer_fact_ledger_merge
|
||
|
||
@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=self._execution_stage or ("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") and not self._defer_fact_ledger_merge:
|
||
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": self._execution_stage or ("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"
|
||
|
||
# P0-3: with the Stage-4 quality gate the cross-review may only see
|
||
# ACCEPTED seat results — reworked/reassigned/gapped deliveries never
|
||
# enter the review prompt (their content is already absent from the
|
||
# fact ledger because the merge was deferred until acceptance).
|
||
if self._state.quality_gate_enabled:
|
||
accepted_results = list(self._state.accepted_task_results)
|
||
review_seats = sorted({
|
||
str(r.get("agent_id") or "")
|
||
for r in accepted_results
|
||
if r.get("agent_id")
|
||
})
|
||
else:
|
||
accepted_results = None
|
||
# 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:
|
||
if accepted_results is not None:
|
||
seat_results_for_review: list[dict[str, Any]] = accepted_results
|
||
else:
|
||
seat_results_for_review = [
|
||
r for r in self._state.turn_results
|
||
if r.get("agent_id") in review_seats
|
||
]
|
||
prompt = build_manager_synthesis_input(
|
||
task_text=self._state.task_text,
|
||
seat_results=seat_results_for_review,
|
||
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"
|
||
# Stage 4: a controlled gap is authoritative — the cross-review
|
||
# decision must reflect it even when everything else looks clean.
|
||
or (self._state.quality_gate_enabled and self._state.controlled_gaps)
|
||
)
|
||
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, "")
|
||
|
||
if state.quality_gate_enabled:
|
||
# P0-3: Stage 4 synthesizes ONLY accepted task results. There is
|
||
# deliberately NO fallback to ``turn_results`` — when no delivery was
|
||
# accepted, the report must be built from the controlled-gap risk
|
||
# section alone instead of smuggling rejected content back in.
|
||
seat_results_for_synthesis: list[dict[str, Any]] = list(state.accepted_task_results)
|
||
else:
|
||
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"]
|
||
seat_results_for_synthesis = usable or state.turn_results
|
||
|
||
synthesis_input = build_manager_synthesis_input(
|
||
task_text=state.task_text,
|
||
seat_results=seat_results_for_synthesis,
|
||
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]}"
|
||
|
||
# Stage 4 (R3-Q): controlled gaps must reach the final report traceably and
|
||
# must NEVER be presented as verified facts. They are injected into the
|
||
# synthesis prompt as an explicit risk/pending-info section.
|
||
if state.controlled_gaps:
|
||
gap_lines = [
|
||
f"- 任务 {gap.get('task_key')}（席位 {gap.get('agent_id')}，"
|
||
f"尝试 {gap.get('attempts', 0)} 次）：{gap.get('reason')}"
|
||
for gap in state.controlled_gaps
|
||
]
|
||
synthesis_input += (
|
||
"\n\n【受控缺口（尚未获得合格席位交付，严禁写成已验证事实）】\n"
|
||
+ "\n".join(gap_lines)
|
||
+ "\n要求：把上述每一条缺口完整列入最终结论的『风险/待补信息』章节，"
|
||
"逐条说明其对结论的影响；不得将缺口内容陈述为已验证结论。"
|
||
)
|
||
|
||
has_gaps = bool(state.controlled_gaps)
|
||
quality_decision = (
|
||
"controlled_gap"
|
||
if has_gaps
|
||
else (
|
||
(state.cross_review_result or {}).get("decision")
|
||
or ("controlled_gap" if state.partial_failure else "accept")
|
||
)
|
||
)
|
||
|
||
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 or has_gaps,
|
||
"quality_decision": quality_decision,
|
||
"controlled_gaps": list(state.controlled_gaps),
|
||
}
|
||
else:
|
||
state.final_result = {
|
||
"content": synthesis_input,
|
||
"agent_id": manager_id,
|
||
"note": "no_manager_thread",
|
||
"partial_failure": state.partial_failure,
|
||
"degraded": state.partial_failure or has_gaps,
|
||
"quality_decision": quality_decision,
|
||
"controlled_gaps": list(state.controlled_gaps),
|
||
}
|
||
|
||
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"})
|
||
|
||
|
||
class _SilentWorkflowCtx:
|
||
"""WorkflowContext stand-in for SeatExecutor invoked INSIDE the quality
|
||
loop. The embedded SeatExecutor reuses the outer executor's ctx when
|
||
called directly, which would otherwise publish its messages with the
|
||
outer node's source id and re-enter the graph. A silent ctx discards
|
||
those intermediate messages — the quality loop reads the delivery state
|
||
from ``turn_results`` / ``seat_outcomes`` instead."""
|
||
|
||
async def send_message(self, msg: Any) -> None:
|
||
return None
|
||
|
||
async def yield_output(self, msg: Any) -> None:
|
||
return None
|
||
|
||
|
||
class WaitingInputTerminalExecutor(Executor):
|
||
"""Terminal executor for the Stage-4 ask_user / waiting_input path.
|
||
|
||
Reached when a seat task's fallback policy requests user clarification
|
||
(``ask_user``) and the manager cannot resolve the gap. The run ends in
|
||
``waiting_input`` — never as a disguised success.
|
||
"""
|
||
|
||
def __init__(self) -> None:
|
||
super().__init__(id="waiting_input_terminal")
|
||
|
||
@handler(input=Any, workflow_output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
await ctx.yield_output({"status": "waiting_input"})
|
||
|
||
|
||
class QualitySeatTasksExecutor(Executor):
|
||
"""Stage 4 (R3-Q): run every seat task through the acceptance state machine.
|
||
|
||
Replaces the plain SeatExecutor fan-out for V2 structured dispatch. One
|
||
executor drives all tasks in bounded quality rounds::
|
||
|
||
delivering → submitted → manager_review →
|
||
accepted | rework(→delivering) | reassigned(→delivering) |
|
||
controlled_gap | ask_user(→ waiting_input)
|
||
|
||
Every delivery runs a deterministic quality check first; the manager model
|
||
then proposes one structured decision per task, and the server finalizes it
|
||
via :func:`quality_contracts.finalize_review_decision` — a delivery that
|
||
failed the deterministic checks can NEVER be accepted, reworks are capped
|
||
by the frozen contract, and an exhausted task ends as ``controlled_gap``
|
||
(or an explicit ``ask_user`` pause) instead of being disguised as success.
|
||
|
||
Task rows persist in ``maf_roundtable_seat_tasks``; reassignment only
|
||
changes the task's ``agent_id`` (a server-validated backup candidate) and
|
||
never creates a new task.
|
||
"""
|
||
|
||
def __init__(
|
||
self,
|
||
adapter: Any,
|
||
state: RoundtableState,
|
||
*,
|
||
action_coordinator: ActionCoordinator | None = None,
|
||
max_rework_attempts: int = 2,
|
||
max_reassignments: int = 1,
|
||
min_delivery_chars: int = 40,
|
||
min_delivery_evidence: int = 1,
|
||
fallback_policy: str = "reassign_then_gap",
|
||
) -> None:
|
||
super().__init__(id="quality_seat_tasks")
|
||
self._adapter = adapter
|
||
self._state = state
|
||
self._action_coordinator = action_coordinator
|
||
self._max_rework_attempts = max_rework_attempts
|
||
self._max_reassignments = max_reassignments
|
||
self._min_delivery_chars = min_delivery_chars
|
||
self._min_delivery_evidence = min_delivery_evidence
|
||
self._fallback_policy = fallback_policy
|
||
self._tasks: list[dict[str, Any]] = []
|
||
|
||
@property
|
||
def _store(self) -> Any:
|
||
return self._adapter._service.store
|
||
|
||
@handler(input=Any, output=dict)
|
||
async def handle(self, message: Any, ctx: WorkflowContext) -> None:
|
||
if self._state.is_completed("quality_seat_tasks"):
|
||
await ctx.send_message({
|
||
"tasks": dict(self._state.seat_task_statuses),
|
||
"replayed": True,
|
||
})
|
||
return
|
||
|
||
self._state.stage = "experts"
|
||
tasks = await self._load_or_create_tasks()
|
||
self._tasks = tasks
|
||
base_round = max(self._state.current_round, 1)
|
||
# P1: duplicate detection is seeded from the persisted task history
|
||
# (SHA-256 digests of prior conclusions), so a restart cannot re-submit
|
||
# an already-rejected conclusion and pass it off as new content.
|
||
prior_conclusions: dict[str, set[str]] = {}
|
||
for task in tasks:
|
||
digests = {
|
||
str(h.get("conclusion_sha256"))
|
||
for h in (task.get("history") or [])
|
||
if isinstance(h, dict) and h.get("conclusion_sha256")
|
||
}
|
||
prior_conclusions[str(task.get("task_key") or "")] = digests
|
||
|
||
# Bounded quality loop: initial delivery + max_rework_attempts reworks
|
||
# per task budget, plus one delivery round per possible reassignment.
|
||
max_rounds = (self._max_rework_attempts + 1) * (self._max_reassignments + 1) + 1
|
||
for round_offset in range(max_rounds):
|
||
self._state.current_round = base_round + round_offset
|
||
active = [t for t in tasks if t["status"] not in TASK_TERMINAL_STATUSES]
|
||
if not active or self._state.status == "waiting_input":
|
||
break
|
||
|
||
submitted: list[dict[str, Any]] = []
|
||
for task in active:
|
||
outcome = await self._deliver(task, prior_conclusions)
|
||
if outcome is not None:
|
||
submitted.append(outcome)
|
||
|
||
if not submitted:
|
||
continue
|
||
|
||
reviews = await self._manager_review(submitted)
|
||
for task in submitted:
|
||
final_outcome = await self._apply_decision(task, reviews.get(task["agent_id"]))
|
||
if final_outcome == OUTCOME_ASK_USER:
|
||
self._state.status = "waiting_input"
|
||
self._state.stage = "waiting_input"
|
||
|
||
self._state.seat_task_statuses = {
|
||
t["task_key"]: {
|
||
"task_key": t["task_key"],
|
||
"status": t["status"],
|
||
"agent_id": t["agent_id"],
|
||
"original_agent_id": t["original_agent_id"],
|
||
"attempt_count": t["attempt_count"],
|
||
"rework_count": t["rework_count"],
|
||
"reassign_count": t["reassign_count"],
|
||
"required": bool(t["required"]),
|
||
}
|
||
for t in tasks
|
||
}
|
||
self._state.mark_completed("quality_seat_tasks")
|
||
await ctx.send_message({"tasks": dict(self._state.seat_task_statuses)})
|
||
|
||
# ── task init (atomic, recovery-safe) ───────────────────────────────────
|
||
|
||
def _expected_task_specs(self) -> list[dict[str, Any]]:
|
||
"""The server-computed expected task set from the dispatch assignments.
|
||
|
||
``task_key`` is stable per original assignment, so recovery always
|
||
derives the identical key set — existing rows are kept, rows that a
|
||
crashed initialization never created are filled back in.
|
||
"""
|
||
plan = self._state.dispatch_plan or {}
|
||
assignments = plan.get("assignments") or []
|
||
specs: list[dict[str, Any]] = []
|
||
for index, assignment in enumerate(assignments, start=1):
|
||
agent_id = str(assignment.get("agent_id") or "").strip()
|
||
if not agent_id:
|
||
continue
|
||
contract = build_task_contract(
|
||
task_goal=str(assignment.get("task_goal") or ""),
|
||
delivery_requirements=list(assignment.get("delivery_requirements") or []),
|
||
required=True,
|
||
max_rework_attempts=self._max_rework_attempts,
|
||
fallback_policy=self._fallback_policy, # type: ignore[arg-type]
|
||
)
|
||
specs.append({
|
||
"task_key": f"{index:02d}:{agent_id}",
|
||
"agent_id": agent_id,
|
||
"contract": contract.model_dump(),
|
||
"required": True,
|
||
})
|
||
return specs
|
||
|
||
async def _load_or_create_tasks(self) -> list[dict[str, Any]]:
|
||
"""Atomic get-or-create over the FULL expected task set (P0-4).
|
||
|
||
Never short-circuits on "some tasks already exist": a crash mid-init
|
||
must not permanently drop the remaining required seats. Existing rows
|
||
keep their frozen contracts; concurrent recoveries converge without
|
||
unique-key errors.
|
||
"""
|
||
return await self._store.seat_tasks.get_or_create_tasks(
|
||
session_id=self._state.session_id,
|
||
run_id=self._state.run_id,
|
||
expected=self._expected_task_specs(),
|
||
)
|
||
|
||
# ── delivery step ───────────────────────────────────────────────────────
|
||
|
||
async def _deliver(
|
||
self,
|
||
task: dict[str, Any],
|
||
prior_conclusions: dict[str, set[str]],
|
||
) -> dict[str, Any] | None:
|
||
"""Run one delivery via SeatExecutor (approvals / ledger / outcomes
|
||
reuse the existing seat path), then submit it with its deterministic
|
||
check result. Returns the submitted task dict or ``None`` if the seat
|
||
run hard-failed without any output."""
|
||
agent_id = task["agent_id"]
|
||
|
||
# Re-delivery must not replay the completed node.
|
||
node_id = f"seat_{agent_id}"
|
||
if node_id in self._state.completed_nodes:
|
||
self._state.completed_nodes.remove(node_id)
|
||
|
||
# Reassigned tasks: add the validated backup seat to the selected set
|
||
# and ensure its participant row + thread before it delivers.
|
||
if task["status"] == TASK_REASSIGNED:
|
||
if self._state.selected_seat_agent_ids is not None and agent_id not in self._state.selected_seat_agent_ids:
|
||
self._state.selected_seat_agent_ids.append(agent_id)
|
||
await _ensure_selected_seat_threads(self._adapter, self._state)
|
||
|
||
is_rework = task["attempt_count"] > 0
|
||
seat_exec = SeatExecutor(
|
||
self._adapter,
|
||
self._state,
|
||
agent_id,
|
||
action_coordinator=self._action_coordinator,
|
||
execution_stage="rework" if is_rework else None,
|
||
# P0-3: the delivery must not touch the fact ledger until the
|
||
# manager's final decision accepts it.
|
||
defer_fact_ledger_merge=True,
|
||
)
|
||
await seat_exec.handle({}, _SilentWorkflowCtx())
|
||
|
||
turn_row = next(
|
||
(
|
||
r for r in reversed(self._state.turn_results)
|
||
if r.get("agent_id") == agent_id
|
||
and r.get("turn_index") == self._state.current_round
|
||
),
|
||
None,
|
||
)
|
||
raw_content = str((turn_row or {}).get("content") or "")
|
||
structured_output = (turn_row or {}).get("structured_output")
|
||
schema_status = str((turn_row or {}).get("schema_status") or "ok")
|
||
delivery = delivery_from_seat_output(
|
||
structured_output=structured_output, raw_content=raw_content
|
||
)
|
||
prior = prior_conclusions.get(task["task_key"])
|
||
if prior is None:
|
||
prior = set()
|
||
prior_conclusions[task["task_key"]] = prior
|
||
# P1: evidence must be traceable to the fact ledger (recorded evidence
|
||
# or successful tool results). Self-made findings / invented refs fail
|
||
# the deterministic gate.
|
||
verified_refs = known_evidence_refs(self._state.fact_ledger)
|
||
check = check_delivery_deterministic(
|
||
delivery=delivery,
|
||
raw_content=raw_content,
|
||
schema_status=schema_status,
|
||
min_length=self._min_delivery_chars,
|
||
min_evidence=self._min_delivery_evidence,
|
||
prior_conclusion_digests=prior,
|
||
verified_evidence_refs=verified_refs,
|
||
)
|
||
digest = conclusion_digest(delivery.conclusion) if delivery.conclusion else None
|
||
if digest:
|
||
prior.add(digest)
|
||
|
||
task["attempt_count"] = int(task.get("attempt_count") or 0) + 1
|
||
history = list(task.get("history") or [])
|
||
history.append({
|
||
"attempt": task["attempt_count"],
|
||
"agent_id": agent_id,
|
||
"round": self._state.current_round,
|
||
"schema_status": schema_status,
|
||
"failure_codes": list(check["failure_codes"]),
|
||
# P1: persist the digest only — never the conclusion body — so
|
||
# duplicate detection survives restarts.
|
||
"conclusion_sha256": digest,
|
||
})
|
||
# P0-3: persisted delivery envelope. ``contract_delivery`` is the
|
||
# acceptance-facing view; ``validated_seat_output`` is the schema-
|
||
# validated raw SeatTurnOutput kept back until acceptance merges it
|
||
# into the fact ledger. Runtime objects / tool-call arguments are
|
||
# never written here.
|
||
task["delivery"] = {
|
||
"contract_delivery": delivery.model_dump(),
|
||
"validated_seat_output": structured_output
|
||
if isinstance(structured_output, dict)
|
||
else None,
|
||
}
|
||
task["deterministic_check"] = check
|
||
task["history"] = history
|
||
# In-memory only (never persisted): the turn row of this attempt, so
|
||
# an acceptance can merge exactly what was submitted.
|
||
task["_turn_row"] = turn_row
|
||
await self._store.seat_tasks.update_task(task["id"], {
|
||
"status": TASK_SUBMITTED,
|
||
"attempt_count": task["attempt_count"],
|
||
"delivery": task["delivery"],
|
||
"deterministic_check": check,
|
||
"history": history,
|
||
})
|
||
task["status"] = TASK_SUBMITTED
|
||
|
||
await self._adapter._emit_critical(self._state.run_id, make_seat_delivery_submitted(
|
||
agent_id,
|
||
self._state.current_round,
|
||
task_key=task["task_key"],
|
||
attempt=task["attempt_count"],
|
||
deterministic_passed=bool(check["passed"]),
|
||
failure_codes=list(check["failure_codes"]),
|
||
))
|
||
return task
|
||
|
||
# ── manager review step ─────────────────────────────────────────────────
|
||
|
||
async def _manager_review(self, tasks: list[dict[str, Any]]) -> dict[str, Any]:
|
||
"""One batched manager review call for the round's submissions.
|
||
|
||
Returns ``{agent_id: MafManagerReviewItem}``. Any failure (timeout,
|
||
runtime error, empty output, invalid JSON, schema failure) degrades to
|
||
an empty dict — every task then finalizes as an INVALID review
|
||
(``source=system``): a system rework while the cap allows it, or the
|
||
contract's exhaustion outcome afterwards. A delivery is never accepted
|
||
without a valid manager review, and a seat missing from a partially
|
||
valid review is treated the same way.
|
||
"""
|
||
manager_id = self._state.manager_agent_id
|
||
thread_id = self._state.thread_map.get(manager_id or "", "")
|
||
if not (manager_id and thread_id):
|
||
return {}
|
||
|
||
review_items: list[dict[str, Any]] = []
|
||
for task in tasks:
|
||
contract = task.get("contract") or {}
|
||
envelope = task.get("delivery") or {}
|
||
# The manager reviews the contract-facing delivery view; the raw
|
||
# validated seat output stays server-side until acceptance.
|
||
contract_delivery = envelope.get("contract_delivery") or envelope
|
||
review_items.append({
|
||
"agent_id": task["agent_id"],
|
||
"task_key": task["task_key"],
|
||
"attempt": task["attempt_count"],
|
||
"contract": {
|
||
"objective": contract.get("objective"),
|
||
"expected_deliverable": contract.get("expected_deliverable"),
|
||
"acceptance_criteria": list(contract.get("acceptance_criteria") or []),
|
||
"required": bool(contract.get("required", True)),
|
||
},
|
||
"delivery": contract_delivery,
|
||
"check": task.get("deterministic_check") or {},
|
||
})
|
||
prompt = build_manager_review_prompt(
|
||
task_text=self._state.task_text, review_items=review_items
|
||
)
|
||
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,
|
||
execution_stage="manager_review",
|
||
message=prompt,
|
||
thread_id=thread_id,
|
||
model_name=self._state.model_name,
|
||
reasoning_effort=self._state.reasoning_effort,
|
||
task_text=prompt,
|
||
stage_instruction="返回严格的总控验收 JSON（每席位一条 decision）",
|
||
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 - degraded to deterministic finalization
|
||
return {}
|
||
if response.error:
|
||
return {}
|
||
model, _errors = parse_manager_review(response.content or "")
|
||
if model is None:
|
||
return {}
|
||
return {item.agent_id: item for item in model.decisions}
|
||
|
||
# ── decision application ────────────────────────────────────────────────
|
||
|
||
async def _apply_decision(
|
||
self,
|
||
task: dict[str, Any],
|
||
review_item: Any | None,
|
||
) -> str:
|
||
"""Finalize and persist one task's review decision; returns the outcome."""
|
||
contract = MafTaskContract.model_validate(task.get("contract") or {})
|
||
check = task.get("deterministic_check") or {"passed": False, "failure_codes": []}
|
||
final = finalize_review_decision(
|
||
contract=contract,
|
||
check_result=check,
|
||
review_item=review_item,
|
||
rework_count=int(task.get("rework_count") or 0),
|
||
reassign_count=int(task.get("reassign_count") or 0),
|
||
has_backup_seat=bool(self._backup_agents(task)),
|
||
max_reassignments=self._max_reassignments,
|
||
)
|
||
outcome = final["decision"]
|
||
|
||
await self._adapter._emit_critical(self._state.run_id, make_manager_review_decision(
|
||
task["agent_id"],
|
||
self._state.current_round,
|
||
task_key=task["task_key"],
|
||
decision=outcome,
|
||
reason=final["reason"],
|
||
source=final["source"],
|
||
attempt=task["attempt_count"],
|
||
))
|
||
|
||
if outcome == OUTCOME_ACCEPTED:
|
||
task["status"] = TASK_ACCEPTED
|
||
# P0-3: only an accepted delivery may enter the fact ledger and
|
||
# the accepted result set consumed by cross-review / the final
|
||
# synthesis. Reworked / reassigned / gapped attempts never do.
|
||
envelope = task.get("delivery") or {}
|
||
validated = envelope.get("validated_seat_output")
|
||
if isinstance(validated, dict):
|
||
merge_seat_output(
|
||
self._state.fact_ledger, validated, agent_id=task["agent_id"]
|
||
)
|
||
turn_row = task.get("_turn_row") or {}
|
||
self._state.accepted_task_results.append({
|
||
"task_key": task["task_key"],
|
||
"agent_id": task["agent_id"],
|
||
"attempt": task["attempt_count"],
|
||
"accepted": True,
|
||
"source": final["source"],
|
||
"turn_index": turn_row.get("turn_index", self._state.current_round),
|
||
"content": str(turn_row.get("content") or ""),
|
||
"execution_stage": turn_row.get("execution_stage"),
|
||
"schema_status": turn_row.get("schema_status"),
|
||
"structured_output": validated if isinstance(validated, dict) else None,
|
||
})
|
||
await self._store.seat_tasks.update_task(task["id"], {
|
||
"status": TASK_ACCEPTED,
|
||
"review_decision": final,
|
||
})
|
||
|
||
elif outcome == OUTCOME_REWORK:
|
||
task["rework_count"] = int(task.get("rework_count") or 0) + 1
|
||
task["status"] = TASK_REWORK
|
||
await self._store.seat_tasks.update_task(task["id"], {
|
||
"status": TASK_REWORK,
|
||
"rework_count": task["rework_count"],
|
||
"review_decision": final,
|
||
})
|
||
await self._adapter._emit_critical(self._state.run_id, make_seat_rework_assigned(
|
||
task["agent_id"],
|
||
self._state.current_round,
|
||
task_key=task["task_key"],
|
||
rework_count=task["rework_count"],
|
||
requirements=list(final.get("revision_requirements") or []),
|
||
))
|
||
|
||
elif outcome == OUTCOME_REASSIGNED:
|
||
from_agent_id = task["agent_id"]
|
||
backup = self._pick_backup(task)
|
||
task["reassign_count"] = int(task.get("reassign_count") or 0) + 1
|
||
task["agent_id"] = backup
|
||
task["status"] = TASK_REASSIGNED
|
||
await self._store.seat_tasks.update_task(task["id"], {
|
||
"agent_id": backup,
|
||
"status": TASK_REASSIGNED,
|
||
"reassign_count": task["reassign_count"],
|
||
"review_decision": final,
|
||
})
|
||
await self._adapter._emit_critical(self._state.run_id, make_seat_reassigned(
|
||
backup,
|
||
self._state.current_round,
|
||
task_key=task["task_key"],
|
||
from_agent_id=from_agent_id,
|
||
reassign_count=task["reassign_count"],
|
||
))
|
||
|
||
elif outcome == OUTCOME_CONTROLLED_GAP:
|
||
gap = make_controlled_gap(
|
||
task_key=task["task_key"],
|
||
agent_id=task["agent_id"],
|
||
objective=contract.objective,
|
||
reason=final["reason"],
|
||
failure_codes=list(check.get("failure_codes") or []),
|
||
attempts=task["attempt_count"],
|
||
)
|
||
task["gap"] = gap
|
||
task["status"] = TASK_CONTROLLED_GAP
|
||
self._state.controlled_gaps.append(gap)
|
||
await self._store.seat_tasks.update_task(task["id"], {
|
||
"status": TASK_CONTROLLED_GAP,
|
||
"gap": gap,
|
||
"review_decision": final,
|
||
})
|
||
|
||
else: # OUTCOME_ASK_USER
|
||
task["status"] = TASK_EXHAUSTED
|
||
await self._store.seat_tasks.update_task(task["id"], {
|
||
"status": TASK_EXHAUSTED,
|
||
"review_decision": final,
|
||
})
|
||
|
||
return outcome
|
||
|
||
# ── backup selection (server-validated) ─────────────────────────────────
|
||
|
||
def _backup_agents(self, task: dict[str, Any]) -> list[str]:
|
||
"""Ordered backup candidates for a task.
|
||
|
||
Candidates = server candidate seats minus every agent this task has
|
||
already used (original + rework owners) minus every agent that is the
|
||
current owner of ANY other task. A seat owns at most one task at a
|
||
time — a seat whose own task was accepted must not be double-booked as
|
||
a backup, because it would then hold two tasks and its deliveries
|
||
would collide on the (run, agent, turn, stage) idempotency key.
|
||
"""
|
||
candidates = list(self._state.candidate_seat_agent_ids or [])
|
||
used: set[str] = {
|
||
str(item.get("agent_id") or "")
|
||
for item in (task.get("history") or [])
|
||
if str(item.get("agent_id") or "")
|
||
}
|
||
used.add(str(task.get("agent_id") or ""))
|
||
for other in self._tasks:
|
||
if other.get("task_key") == task.get("task_key"):
|
||
continue
|
||
used.add(str(other.get("agent_id") or ""))
|
||
return [agent_id for agent_id in candidates if agent_id not in used]
|
||
|
||
def _pick_backup(self, task: dict[str, Any]) -> str:
|
||
backups = self._backup_agents(task)
|
||
if not backups:
|
||
raise RuntimeError(
|
||
f"MAF quality loop requested reassignment for task "
|
||
f"{task.get('task_key')} but no validated backup seat exists"
|
||
)
|
||
return backups[0]
|
||
|
||
|
||
class QualityGateExecutor(Executor):
|
||
"""Stage 4 (R3-Q): phase quality gate between seat tasks and cross-review.
|
||
|
||
All required seat tasks must have reached a terminal state (``accepted``
|
||
or ``controlled_gap``) before the phase may proceed. A required task left
|
||
non-terminal blocks the phase deterministically (hard run failure — the
|
||
quality loop guarantees terminal states, so a blocked gate is a safety
|
||
net, never a normal outcome). Controlled gaps do NOT block the phase:
|
||
they proceed as ``passed_with_gaps`` and are carried into the final
|
||
report's risk section.
|
||
"""
|
||
|
||
def __init__(self, adapter: Any, state: RoundtableState) -> None:
|
||
super().__init__(id="quality_gate")
|
||
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("quality_gate"):
|
||
await ctx.send_message({"decision": "passed", "replayed": True})
|
||
return
|
||
|
||
# ask_user path: route to the waiting_input terminal, never to
|
||
# cross-review / finalizer.
|
||
if self._state.status == "waiting_input":
|
||
self._state.mark_completed("quality_gate")
|
||
await ctx.send_message({"decision": "waiting_input"})
|
||
return
|
||
|
||
statuses = self._state.seat_task_statuses
|
||
required = {k: v for k, v in statuses.items() if v.get("required")}
|
||
non_terminal = [
|
||
k for k, v in required.items() if v.get("status") not in TASK_TERMINAL_STATUSES
|
||
]
|
||
# P0-4 safety net: every dispatch assignment must be represented by a
|
||
# task row. A missing required task blocks the gate exactly like a
|
||
# non-terminal one — the gate must never pass on a dropped seat.
|
||
expected_keys: list[str] = []
|
||
plan = self._state.dispatch_plan or {}
|
||
for index, assignment in enumerate(plan.get("assignments") or [], start=1):
|
||
agent_id = str(assignment.get("agent_id") or "").strip()
|
||
if agent_id:
|
||
expected_keys.append(f"{index:02d}:{agent_id}")
|
||
missing = [k for k in expected_keys if k not in statuses]
|
||
blocked = sorted({*non_terminal, *missing})
|
||
accepted = sum(1 for v in statuses.values() if v.get("status") == TASK_ACCEPTED)
|
||
gap_count = sum(1 for v in statuses.values() if v.get("status") == TASK_CONTROLLED_GAP)
|
||
decision = "blocked" if blocked else ("passed_with_gaps" if gap_count else "passed")
|
||
|
||
await self._adapter._emit_critical(self._state.run_id, make_stage_quality_gate(
|
||
self._state.manager_agent_id,
|
||
self._state.current_round,
|
||
decision=decision,
|
||
total=len(statuses),
|
||
accepted=accepted,
|
||
controlled_gap=gap_count,
|
||
blocked_required=blocked,
|
||
))
|
||
|
||
if blocked:
|
||
raise RuntimeError(
|
||
f"MAF quality gate blocked: required seat tasks not terminal or missing: "
|
||
f"{', '.join(blocked)}"
|
||
)
|
||
|
||
self._state.mark_completed("quality_gate")
|
||
self._state.stage = "review"
|
||
await ctx.send_message({"decision": decision, "tasks": dict(statuses)})
|
||
|
||
|
||
# ── 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
|