//! Workflow 反思（Review）控制器：将“反思阶段 -> 模型修订计划 -> 再执行”做成可测试、可复用的决策逻辑。
//!
//! **FSM**：[`WorkflowReflectionFsmPhase`] 描述会话状态（与 JSON 注入文案正交）；[`WorkflowReflectionController::reflection_fsm_phase`] 供观测。
use serde_json::{Value, json};
use std::collections::HashSet;
use std::fmt;
/// Do 阶段契约校验收集到的节点集合（id 集合 + 节点副本与 id）。
type WorkflowExecuteCollectedNodes = (HashSet<String>, Vec<(Value, String)>);
/// 反思首轮注入 JSON 的 `instruction_type`，与 `per_coord` 中「是否强制终答含 `agent_reply_plan`」对齐。
pub const INSTRUCTION_WORKFLOW_REFLECTION_PLAN_NEXT: &str = "workflow_reflection_plan_next";
pub const INSTRUCTION_WORKFLOW_REFLECTION_NEXT: &str = "workflow_reflection_next";
pub const INSTRUCTION_WORKFLOW_REFLECTION_LOCKED: &str = "workflow_reflection_locked";
pub const INSTRUCTION_WORKFLOW_REFLECTION_MAX_ROUNDS_REACHED: &str =
"workflow_reflection_max_rounds_reached";
/// 工作流反思控制器会话相位（**不含**单次 `decide` 内的 Plan vs Do 计划轮次；后者见 `round`）。
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum WorkflowReflectionFsmPhase {
/// 尚未进入反思会话：`reflection.enabled` 从未在本次会话内置 true。
Inactive,
/// 反思会话进行中：`workflow.done` 尚未置 true，且未因上限锁定。
Reflecting,
/// 会话已结束侧：`workflow.done=true` 已处理，或已达 `max_rounds`，控制器 **`locked`**。
SessionClosed,
}
impl fmt::Display for WorkflowReflectionFsmPhase {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str(match self {
Self::Inactive => "inactive",
Self::Reflecting => "reflecting",
Self::SessionClosed => "session_closed",
})
}
}
#[derive(Debug, Clone, Copy)]
struct ReflectionControl {
enabled: bool,
done: bool,
max_rounds: usize,
}
#[derive(Debug, Clone)]
pub struct WorkflowReflectionDecision {
/// 是否真的调用 `workflow_execute`（DAG 调度由 `workflow.rs` 内部完成；当 done=true 时会被跳过）。
pub execute: bool,
/// 当 `execute=false` 时，作为 tool_result 返回给模型/用户的输出。
pub stop_output: Option<Value>,
/// 在 tool_result 消息之后，运行时应额外注入给模型的“下一步反思指令”。
pub inject_instruction: Option<Value>,
/// 运行时在调用 `workflow_execute` 时，对 `workflow` 字段进行的额外覆盖（例如 validate_only=true）。
/// 如为 None 表示不修改模型传入的 args。
pub workflow_args_patch: Option<Value>,
}
/// 状态机：**Inactive** → **Reflecting**（多 stage）→ **SessionClosed**（`locked`）。
#[derive(Debug, Clone)]
pub struct WorkflowReflectionController {
mode_active: bool,
locked: bool,
max_rounds: usize,
round: usize,
default_max_rounds: usize,
}
impl WorkflowReflectionController {
pub fn new(default_max_rounds: usize) -> Self {
Self {
mode_active: false,
locked: false,
max_rounds: default_max_rounds.max(1),
round: 0,
default_max_rounds: default_max_rounds.max(1),
}
}
/// 反思会话内当前「stage」轮次（每轮 `decide` 可能递增），供 PER 日志与排错。
pub fn stage_round(&self) -> usize {
self.round
}
/// 当前 FSM 相位（纯函数于内存字段；**同一轮** `decide` 开头激活会话后仍为 [`Reflecting`]）。
pub(crate) fn reflection_fsm_phase(&self) -> WorkflowReflectionFsmPhase {
if !self.mode_active {
WorkflowReflectionFsmPhase::Inactive
} else if self.locked {
WorkflowReflectionFsmPhase::SessionClosed
} else {
WorkflowReflectionFsmPhase::Reflecting
}
}
/// 设置严格模式：是否要求终答 `agent_reply_plan` 覆盖全部 workflow 节点 id。
/// 该值来自 [`PerCoordinatorInit::final_plan_require_strict_workflow_node_coverage`]，
/// 在 workflow 反思控制器构造时由外部注入。
pub fn set_require_strict_workflow_node_coverage(&mut self, required: bool) {
// 目前 `WorkflowReflectionController` 不直接持有此标志，
// 这里预留 setter 接口供 `PerCoordinator` 初始化后注入配置。
// 实际校验逻辑在 `per_coord/final_plan_gate.rs`（经 `after_final_assistant`）中；
// 若后续需要在此处内联，可改为 `self.require_strict_workflow_node_coverage = required`。
let _ = required;
}
fn parse_control(&self, args_json: &str) -> ReflectionControl {
let default = self.default_max_rounds;
let enabled = false;
let done = false;
let max_rounds = default;
let Ok(v) = serde_json::from_str::<Value>(args_json) else {
return ReflectionControl {
enabled,
done,
max_rounds,
};
};
let wf_v = v.get("workflow").unwrap_or(&v);
let enabled = wf_v
.get("reflection")
.and_then(|r| r.get("enabled"))
.and_then(|x| x.as_bool())
.unwrap_or(enabled);
let done = wf_v.get("done").and_then(|x| x.as_bool()).unwrap_or(done);
let max_rounds = wf_v
.get("reflection")
.and_then(|r| r.get("max_rounds"))
.and_then(|x| x.as_u64())
.map(|n| (n as usize).max(1))
.unwrap_or(max_rounds);
ReflectionControl {
enabled,
done,
max_rounds,
}
}
pub fn decide(&mut self, args_json: &str) -> WorkflowReflectionDecision {
let control = self.parse_control(args_json);
self.maybe_activate_reflection_session(&control);
match self.reflection_fsm_phase() {
WorkflowReflectionFsmPhase::Inactive => decision_passthrough_execute(),
WorkflowReflectionFsmPhase::Reflecting => self.decide_while_reflecting(&control),
WorkflowReflectionFsmPhase::SessionClosed => decision_locked_stop(self.max_rounds),
}
}
fn maybe_activate_reflection_session(&mut self, control: &ReflectionControl) {
if control.enabled && !self.mode_active {
self.mode_active = true;
self.locked = false;
self.max_rounds = control.max_rounds;
self.round = 0;
}
}
fn decide_while_reflecting(
&mut self,
control: &ReflectionControl,
) -> WorkflowReflectionDecision {
if control.done {
self.locked = true;
return decision_done_close_session();
}
if self.round >= self.max_rounds {
self.locked = true;
return decision_max_rounds_stop(self.max_rounds);
}
self.round += 1;
let workflow_args_patch = Some(json!({
"validate_only": self.round == 1
}));
WorkflowReflectionDecision {
execute: true,
stop_output: None,
inject_instruction: Some(json!({
"instruction_type": if self.round == 1 {
INSTRUCTION_WORKFLOW_REFLECTION_PLAN_NEXT
} else {
INSTRUCTION_WORKFLOW_REFLECTION_NEXT
},
"round": self.round,
"max_rounds": self.max_rounds,
"required_model_action": if self.round == 1 {
"plan_from_validate_only_result: you MUST embed a ```json fenced block whose JSON is {\"type\":\"agent_reply_plan\",\"version\":1,\"steps\":[{\"id\":\"...\",\"description\":\"...\",\"workflow_node_id\":\"<node id from validate result>\"},...]}. If the validate_only tool_result includes a non-empty `nodes` array, you MUST: (1) set `steps.len()` equal to `nodes.length`; (2) set `workflow_node_id` on EVERY step; (3) make the multiset of all `workflow_node_id` values exactly match `nodes[].id` (including duplicates; step order may differ from DAG order). Each description should reflect the validate_only tool_result (execution_layers, deps, required_approval, timeout_secs, compensate_with, mapping layer->nodes). Natural language outside the fence is OK. After producing the plan, call workflow_execute again with workflow.validate_only=false (Do stage), unless you already set workflow.done=true."
} else {
"revise_workflow_then_call_workflow_execute_or_set_workflow_done_true_when_goal_reached"
},
"next_call_hint": {
"call_tool": "workflow_execute",
"workflow_done": false
}
})),
workflow_args_patch,
}
}
}
fn decision_passthrough_execute() -> WorkflowReflectionDecision {
WorkflowReflectionDecision {
execute: true,
stop_output: None,
inject_instruction: None,
workflow_args_patch: None,
}
}
fn decision_done_close_session() -> WorkflowReflectionDecision {
WorkflowReflectionDecision {
execute: true,
stop_output: None,
inject_instruction: None,
workflow_args_patch: None,
}
}
fn decision_max_rounds_stop(max_rounds: usize) -> WorkflowReflectionDecision {
WorkflowReflectionDecision {
execute: false,
stop_output: Some(json!({
"type": "workflow_reflection_stop",
"instruction_type": INSTRUCTION_WORKFLOW_REFLECTION_MAX_ROUNDS_REACHED,
"max_rounds": max_rounds,
"human_summary": format!(
"workflow_execute 已停止：达到反思重试上限（max_rounds={}）。",
max_rounds
),
})),
inject_instruction: Some(json!({
"instruction_type": INSTRUCTION_WORKFLOW_REFLECTION_MAX_ROUNDS_REACHED,
"max_rounds": max_rounds
})),
workflow_args_patch: None,
}
}
fn decision_locked_stop(max_rounds: usize) -> WorkflowReflectionDecision {
WorkflowReflectionDecision {
execute: false,
stop_output: Some(json!({
"type": "workflow_reflection_stop",
"instruction_type": INSTRUCTION_WORKFLOW_REFLECTION_LOCKED,
"max_rounds": max_rounds,
"human_summary": format!(
"workflow_execute 已停止：反思已锁定（max_rounds={}）。",
max_rounds
),
})),
inject_instruction: Some(json!({
"instruction_type": INSTRUCTION_WORKFLOW_REFLECTION_LOCKED,
"max_rounds": max_rounds
})),
workflow_args_patch: None,
}
}
/// 将 `workflow_args_patch` 应用到原始 `args_json` 中，并返回新的 args_json 字符串。
/// - 若 args_json 顶层包含 `workflow` 对象，则 patch 合并进 `workflow`
/// - 否则 patch 合并进顶层对象
pub fn apply_workflow_patch(args_json: &str, workflow_patch: &Value) -> String {
let Ok(mut v) = serde_json::from_str::<Value>(args_json) else {
return args_json.to_string();
};
let Some(patch_obj) = workflow_patch.as_object() else {
return v.to_string();
};
// 优先写入 workflow 字段
let target = v.get_mut("workflow").and_then(|w| w.as_object_mut());
if let Some(t) = target {
for (k, val) in patch_obj {
t.insert(k.clone(), val.clone());
}
return v.to_string();
}
// 否则写入顶层
if let Some(root) = v.as_object_mut() {
for (k, val) in patch_obj {
root.insert(k.clone(), val.clone());
}
}
v.to_string()
}
fn workflow_contract_error(summary: impl Into<String>) -> Value {
json!({
"type": "workflow_execute_do_contract_error",
"human_summary": summary.into(),
})
}
fn collect_workflow_execute_nodes(nodes_v: &Value) -> Result<WorkflowExecuteCollectedNodes, Value> {
let mut node_ids: HashSet<String> = HashSet::new();
let mut entries: Vec<(Value, String)> = Vec::new();
if let Some(arr) = nodes_v.as_array() {
if arr.is_empty() {
return Err(workflow_contract_error("Do 阶段 workflow.nodes 不能为空"));
}
for node in arr.iter() {
let id = node
.get("id")
.and_then(|x| x.as_str())
.map(|s| s.to_string());
let Some(id) = id else {
return Err(workflow_contract_error(
"Do 阶段 nodes 数组中的每个 node 都必须有字符串 id",
));
};
node_ids.insert(id.clone());
entries.push((node.clone(), id));
}
} else if let Some(obj) = nodes_v.as_object() {
if obj.is_empty() {
return Err(workflow_contract_error(
"Do 阶段 workflow.nodes 对象不能为空",
));
}
for (id, node) in obj.iter() {
node_ids.insert(id.clone());
entries.push((node.clone(), id.clone()));
}
} else {
return Err(workflow_contract_error(
"Do 阶段 workflow.nodes 必须是数组或对象",
));
}
Ok((node_ids, entries))
}
fn validate_one_workflow_node_contract(
node: &Value,
id: &str,
node_ids: &HashSet<String>,
) -> Result<(), Value> {
let node_obj = node
.as_object()
.ok_or_else(|| workflow_contract_error(format!("node {} 必须是对象", id)))?;
let deps_values: Vec<Value> = match node_obj.get("deps") {
None => Vec::new(),
Some(dv) => dv
.as_array()
.ok_or_else(|| workflow_contract_error(format!("node {} 的 deps 必须是数组", id)))?
.clone(),
};
for dep in deps_values.iter() {
let dep_id = dep.as_str().ok_or_else(|| {
workflow_contract_error(format!("node {} 的 deps 元素必须是字符串", id))
})?;
if !node_ids.contains(dep_id) {
return Err(workflow_contract_error(format!(
"node {} 的 deps 引用了未知节点 {}",
id, dep_id
)));
}
}
let tool_name = node_obj
.get("tool_name")
.or_else(|| node_obj.get("tool"))
.and_then(|x| x.as_str())
.unwrap_or("");
if tool_name.trim().is_empty() {
return Err(workflow_contract_error(format!(
"node {} 缺少 tool_name（或 tool）",
id
)));
}
Ok(())
}
/// Do 阶段契约校验：当 `workflow.validate_only != true` 时，
/// 在真正执行 DAG 之前，确保 workflow 的 nodes/依赖结构是一个可执行的基本形态。
///
/// 返回 Err(Value) 用于直接作为 tool_result 返回给模型（包含 human_summary）。
pub fn validate_workflow_execute_do_contract(args_json: &str) -> Result<(), Value> {
let v: Value = match serde_json::from_str(args_json) {
Ok(v) => v,
Err(_) => {
return Err(workflow_contract_error(
"workflow_execute_do_contract：参数不是合法 JSON",
));
}
};
let wf_v = v.get("workflow").unwrap_or(&v);
let validate_only = wf_v
.get("validate_only")
.and_then(|x| x.as_bool())
.unwrap_or(false);
// Plan 阶段只做 validate，不检查契约
if validate_only {
return Ok(());
}
let nodes_v = match wf_v.get("nodes") {
Some(n) => n,
None => {
return Err(workflow_contract_error(
"Do 阶段必须提供 workflow.nodes（不能为空）",
));
}
};
let (node_ids, entries) = collect_workflow_execute_nodes(nodes_v)?;
for (node, id) in entries.iter() {
validate_one_workflow_node_contract(node, id.as_str(), &node_ids)?;
}
Ok(())
}
#[cfg(test)]
mod tests {
use super::*;
fn base_args(enabled: bool, done: bool, max_rounds: usize) -> String {
format!(
r#"{{
"workflow": {{
"reflection": {{"enabled": {}, "max_rounds": {}}},
"done": {}
}}
}}"#,
enabled, max_rounds, done
)
}
#[test]
fn reflection_fsm_phase_tracks_activation_done_and_lock() {
let mut c = WorkflowReflectionController::new(5);
assert_eq!(
c.reflection_fsm_phase(),
WorkflowReflectionFsmPhase::Inactive
);
let _ = c.decide(&base_args(true, false, 3));
assert_eq!(
c.reflection_fsm_phase(),
WorkflowReflectionFsmPhase::Reflecting
);
let _ = c.decide(&base_args(true, true, 3));
assert_eq!(
c.reflection_fsm_phase(),
WorkflowReflectionFsmPhase::SessionClosed
);
}
#[test]
fn test_activate_and_stage_injection_then_stop_on_max_rounds() {
let mut c = WorkflowReflectionController::new(5);
let max_rounds = 2;
// round=0 -> first call => round becomes 1, execute=true, injection
let d1 = c.decide(&base_args(true, false, max_rounds));
assert!(d1.execute);
assert!(d1.inject_instruction.is_some());
assert!(d1.stop_output.is_none());
assert!(
d1.workflow_args_patch
.as_ref()
.and_then(|v| v.get("validate_only"))
.and_then(|x| x.as_bool())
.unwrap_or(false)
);
// second call => round becomes 2, execute=true
let d2 = c.decide(&base_args(true, false, max_rounds));
assert!(d2.execute);
assert!(d2.inject_instruction.is_some());
assert!(
!d2.workflow_args_patch
.as_ref()
.and_then(|v| v.get("validate_only"))
.and_then(|x| x.as_bool())
.unwrap_or(true)
);
// third call => round>=max_rounds, stop and lock
let d3 = c.decide(&base_args(true, false, max_rounds));
assert!(!d3.execute);
assert!(d3.stop_output.is_some());
assert!(d3.inject_instruction.is_some());
assert_eq!(
d3.inject_instruction
.as_ref()
.and_then(|v| v.get("instruction_type"))
.and_then(|v| v.as_str())
.unwrap_or(""),
INSTRUCTION_WORKFLOW_REFLECTION_MAX_ROUNDS_REACHED
);
// subsequent call => locked stop (different stop text)
let d4 = c.decide(&base_args(false, false, max_rounds));
assert!(!d4.execute);
assert_eq!(
d4.stop_output
.as_ref()
.and_then(|v| v.get("instruction_type"))
.and_then(|v| v.as_str())
.unwrap_or(""),
INSTRUCTION_WORKFLOW_REFLECTION_LOCKED
);
assert_eq!(
d4.inject_instruction
.as_ref()
.and_then(|v| v.get("instruction_type"))
.and_then(|v| v.as_str())
.unwrap_or(""),
INSTRUCTION_WORKFLOW_REFLECTION_LOCKED
);
}
#[test]
fn test_done_true_ends_without_injection() {
let mut c = WorkflowReflectionController::new(5);
let d1 = c.decide(&base_args(true, true, 3));
assert!(d1.execute);
assert!(d1.inject_instruction.is_none());
assert!(d1.stop_output.is_none());
// locked now, done=false -> stop
let d2 = c.decide(&base_args(false, false, 3));
assert!(!d2.execute);
}
#[test]
fn test_validate_do_contract_requires_nodes_and_deps_shape() {
// validate_only=false + nodes 缺失 => 错
let bad_args =
r#"{"workflow":{"reflection":{"enabled":true,"max_rounds":5},"done":false}}"#;
let err = validate_workflow_execute_do_contract(bad_args).unwrap_err();
assert_eq!(
err.get("type").and_then(|v| v.as_str()).unwrap_or(""),
"workflow_execute_do_contract_error"
);
// nodes 存在但 deps 非数组 => 错
let bad_deps = r#"{
"workflow":{
"validate_only":false,
"nodes":[{"id":"a","tool_name":"calc","deps":"not_array"}]
}
}"#;
let err = validate_workflow_execute_do_contract(bad_deps).unwrap_err();
assert_eq!(
err.get("type").and_then(|v| v.as_str()).unwrap_or(""),
"workflow_execute_do_contract_error"
);
// deps 引用了未知节点 => 错
let unknown_dep = r#"{
"workflow":{
"validate_only":false,
"nodes":[
{"id":"a","tool_name":"calc","deps":["b"]}
]
}
}"#;
let err = validate_workflow_execute_do_contract(unknown_dep).unwrap_err();
assert_eq!(
err.get("type").and_then(|v| v.as_str()).unwrap_or(""),
"workflow_execute_do_contract_error"
);
// 结构正确 => ok
let ok_args = r#"{
"workflow":{
"validate_only":false,
"nodes":[
{"id":"a","tool_name":"calc","deps":[]}
]
}
}"#;
assert!(validate_workflow_execute_do_contract(ok_args).is_ok());
// deps 缺失 => 允许（按 parse_workflow_spec 视为 []）
let ok_args_no_deps = r#"{
"workflow":{
"validate_only":false,
"nodes":[
{"id":"a","tool_name":"calc"}
]
}
}"#;
assert!(validate_workflow_execute_do_contract(ok_args_no_deps).is_ok());
}
}