test: recognize workflow agent codes from snapshots

This commit is contained in:
2026-06-13 20:05:54 +08:00
parent 22fb065ed8
commit 374ab25977
@@ -145,11 +145,18 @@ async def verify(args: argparse.Namespace) -> int:
if len(execution_log) < args.min_steps: if len(execution_log) < args.min_steps:
raise RuntimeError(f"execution log too short: {len(execution_log)} < {args.min_steps}") raise RuntimeError(f"execution log too short: {len(execution_log)} < {args.min_steps}")
snapshot_nodes = (run_detail.get("definition_snapshot") or {}).get("nodes") or []
node_agent_codes = {
node.get("id"): (node.get("data") or {}).get("agent_code")
for node in snapshot_nodes
if isinstance(node, dict) and (node.get("data") or {}).get("agent_code")
}
agent_codes = sorted( agent_codes = sorted(
{ {
log.get("metadata", {}).get("agent_code") log.get("metadata", {}).get("agent_code") or node_agent_codes.get(log.get("node_id"))
for log in execution_log for log in execution_log
if isinstance(log, dict) and log.get("metadata", {}).get("agent_code") if isinstance(log, dict)
and (log.get("metadata", {}).get("agent_code") or node_agent_codes.get(log.get("node_id")))
} }
) )
if len(agent_codes) < args.min_agents: if len(agent_codes) < args.min_agents: