| name | team-adversarial-swarm |
| description | ACO swarm intelligence with modular Workflow composition and adversarial decision gates. Coordinator drives iteration loop; 4 composable Workflow scripts handle exploration, scoring, convergence, and synthesis — each with built-in adversarial patterns. |
| allowed-tools | ["AskUserQuestion","Bash","Edit","Glob","Grep","Read","Write","maestro","teammate"] |
| session-mode | run |
<required_reading>
@~/.maestro/workflows/run-mode-lite.md
</required_reading>
Team Adversarial Swarm
ACO 蚁群优化 + 模块化 Workflow 编排 + 对抗决策。
继承 team-swarm 的蚁群算法核心（Python ACO 脚本），用 4 个可组合的 Workflow 脚本
替代 team-worker 架构，在每个决策节点注入对抗性 agent 模式。
Architecture
SKILL.md (Coordinator — this file)
│
│ Phase 1: Config Generation (inline)
│ Phase 2: ACO Init (Bash: aco.py init)
│
│ Phase 3: Iteration Loop ×K
│ ┌──────────────────────────────────────────────────┐
│ │ 3a. Bash: aco.py select → assignments │
│ │ 3b. Workflow(wf-swarm-explore) ← 模块1 │
│ │ N ants parallel → ant_results │
│ │ 3c. Workflow(wf-swarm-score) ← 模块2 │
│ │ 3-vote adversarial scoring → verified_scores │
│ │ 3d. Write scores → Bash: aco.py update │
│ │ 3e. Workflow(wf-swarm-converge) ← 模块3 │
│ │ prosecutor/defender/judge → converged? │
│ │ 3f. if converged: break │
│ └──────────────────────────────────────────────────┘
│
│ Phase 4: Bash: aco.py report
│ Workflow(wf-swarm-synthesize) ← 模块4
│ 3-perspective analysis + arbitration → best-solution.md
Workflow Module Registry
| Module | Script | Args Interface | Adversarial Pattern | Returns |
|---|
| Explore | workflows/wf-swarm-explore.js | { iteration, assignments[], objective, session, config } | N ants parallel | { ant_results[] } |
| Score | workflows/wf-swarm-score.js | { iteration, ant_results[], objective, rubric? } | 3-vote per ant (prosecutor/defender/judge) | { scores{}, calibration } |
| Converge | workflows/wf-swarm-converge.js | { iteration, best, history[], config } | prosecutor(continue)/defender(stop)/judge | { converged, reason, confidence } |
| Synthesize | workflows/wf-swarm-synthesize.js | { best, top_k[], convergence_story, objective } | 3-perspective + arbitrator | { report, caveats } |
每个模块独立可用，也可由 Coordinator 组合编排。
Shared Dependencies
所有依赖均在本 skill 内部，无外部引用。
- Python ACO 脚本:
<this-skill>/scripts/aco.py
- 运行时解析:
Glob(".pi/skills/team-adversarial-swarm/scripts/aco.py")
- 依赖模块:
pheromone.py, scoring.py（同目录）
- 命令:
init / select / update / converged / report
- 协议: specs/swarm-protocol.md
- Workflow 脚本:
<this-skill>/workflows/wf-swarm-*.js
- 运行时解析:
Glob(".pi/skills/team-adversarial-swarm/workflows/wf-swarm-*.js")
Specs Reference
Session Directory
{run_dir}/work/team/
├── swarm-config.json # Phase 1 output
├── pheromone/ # ACO state (managed by aco.py)
│ ├── current.json
│ └── history/
├── trails/ # Per-iteration trails (managed by aco.py)
├── scores/ # Adversarial scoring results
│ └── iter-<k>-scores.json
├── artifacts/ # scratch/intermediate; formal deliverables go to {run_dir}/outputs/
│ ├── ant-<k>-<id>.json # Ant outputs
│ └── best-solution.md # Final synthesis -> {run_dir}/outputs/best-solution.md
├── workflows/ # Workflow run artifacts
│ ├── explore-<k>.json # Per-iteration explore results
│ ├── score-<k>.json # Per-iteration score results
│ └── converge-<k>.json # Per-iteration convergence decision
└── best.json # Canonical best (managed by aco.py)
Coordinator Execution Flow
Phase 0: Resume Check
Glob("{run_dir}/work/team/swarm-config.json") → 查找活跃 session- 若存在且有
workflows/converge-*.json 未标记 converged → 恢复到对应迭代
- 若无活跃 session → Phase 1
Phase 1: Config Generation
解析用户 intent，生成 swarm-config.json。
若 intent 不够明确，用 AskUserQuestion 澄清：
- 搜索空间是什么？（文件 glob / 节点列表 / 抽象决策集）
- 目标是什么？（最优方案 / 发现问题 / 优化路径）
- 如何评分？（测试通过率 / lint / 自定义规则 / LLM 对抗评分）
- 预算？（最大迭代 / 每轮蚁数 / token 预算）
生成 config 字段：
{
"task": { "objective": "...", "evidence_requirements": "..." },
"swarm": { "n_ants": 5, "max_iterations": 5 },
"aco": { "alpha": 1.0, "beta": 2.0, "rho": 0.1, "q": 1.0 },
"task_space": { "nodes": [...], "auto_discover_from": "..." },
"scoring": { "mode": "adversarial", "rubric": "..." },
"convergence": { "patience": 2, "min_improvement": 0.01, "max_iterations": 5 }
}
Write 到 {run_dir}/work/team/swarm-config.json。
Phase 2: ACO Init
- 创建 session 目录:
TAS-<slug>-<date>
- 解析 aco.py 路径（从 team-swarm skill 继承）
Bash: python <aco.py> --session {run_dir}/work/team init- 解析输出:
{ n_nodes, n_edges, pheromone_path }
Run Lifecycle Integration
After session folder creation and before role-spec generation:
- Resolve Run (birth-packet first): if the dispatch context already carries
run_id / run_dir (injected by an orchestrator), store them in team-session.json and skip create — a second create mints an empty duplicate Run. Otherwise: maestro run create team-adversarial-swarm --session <slug> --intent "<task summary>"
- Resume: Read
team-session.json.run.run_id → maestro run check <run_id> (idempotent). If status=sealed, create a new run and update the field.
Phase 3: Iteration Loop
for k in range(1, max_iterations + 1):
assignments = Bash("python aco.py --session {run_dir}/work/team select --iter k")
explore_result = Workflow({
scriptPath: "<skill>/workflows/wf-swarm-explore.js",
args: { iteration: k, assignments, objective, session, config }
})
score_result = Workflow({
scriptPath: "<skill>/workflows/wf-swarm-score.js",
args: { iteration: k, ant_results: explore_result.ant_results, objective, rubric }
})
Write("{run_dir}/work/team/scores/iter-k-scores.json", score_result)
Bash("python aco.py --session {run_dir}/work/team update --iter k")
converge_result = Workflow({
scriptPath: "<skill>/workflows/wf-swarm-converge.js",
args: { iteration: k, best: aco_best, history: iter_history, config }
})
Write("{run_dir}/work/team/workflows/converge-k.json", converge_result)
if converge_result.converged: break
注意：每次 Workflow 调用是独立的，数据通过 args 传入、返回值传出。
Coordinator 负责 Workflow 间的数据桥接和 Python 脚本调用。
Phase 4: Synthesis
Bash: python aco.py --session {run_dir}/work/team report → 获取 best + top_k + curve- 调用 Workflow Module 4:
Workflow({
scriptPath: "<skill>/workflows/wf-swarm-synthesize.js",
args: { best, top_k, convergence_story, objective }
})
- 将 synthesis 结果写入
{run_dir}/outputs/best-solution.md
- 展示完成摘要 + AskUserQuestion（归档 / 保留 / 导出 / 再跑一轮）
Module Composition Patterns
完整流水线（默认）
explore → score → update → converge → [loop] → synthesize
仅探索（跳过评分，用 self_score）
explore → update(self_score) → converge → synthesize
单次迭代调试
explore(k=1) → score(k=1) // 不循环，只看一轮
独立评分（已有 ant artifacts）
score(ant_results from files) → 输出 verified_scores
独立综合（已有 best + trails）
synthesize(best, top_k) → best-solution.md
Error Handling
| Scenario | Resolution |
|---|
| aco.py 未找到 | Glob team-swarm skill 路径；提示安装 |
| Python < 3.10 | 尝试 python3；报告依赖 |
| Workflow 执行失败 | 记录错误，提供 --resume 恢复点 |
| 所有蚁全部失败 | 暂停，AskUserQuestion（重试/终止/调整config） |
| 收敛从不触发 | max_iterations 安全网总会触发 |
| 幻觉集群 (>50% 蚁被降分) | 暂停，AskUserQuestion（继续/调整评分规则） |
Completion
Run lifecycle completion (before displaying results):
- Read run_id from team-session.json.run.run_id
- Write {run_dir}/report.md with frontmatter (verdict/summary/concerns)
- Run
maestro run complete <run_id>
- If complete fails: fix the blocking gate and retry once; still failing -> do NOT archive/clean - keep the team active (status=paused) and report the blocking gate
展示最终结果 + 交互选择:
- 归档: 保存 session，展示 best-solution.md
- 继续: 保持 session，可追加迭代
- 导出: 复制 best-solution.md 到目标位置
- 再跑: 重置收敛，继续 K 轮
