---
name: catlog22-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."
disable-model-invocation: true
metadata:
  decimalai-skill-id: "abc91341-6668-4ff5-b570-a525f31e2751"
  decimalai-version-id: "f0ade514-d507-4608-a110-888f7d877a88"
  decimalai-version: "2"
  decimalai-body-sha256: "2a74708d32b97963196c0fd92eb034cd71628e50b40432a38384a005d0799e41"
  decimalai-name: "catlog22/team-adversarial-swarm"
  decimalai-format: "1"
  decimalai-invocation: "user"
  decimalai-category: "agent-tooling"
  decimalai-source: "https://app.decimal.ai/s/catlog22-team-adversarial-swarm@2/SKILL.md"
---
<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(".claude/skills/team-adversarial-swarm/scripts/aco.py")`
  - 依赖模块: `pheromone.py`, `scoring.py`（同目录）
  - 命令: `init` / `select` / `update` / `converged` / `report`
  - 协议: [specs/swarm-protocol.md](specs/swarm-protocol.md)
- **Workflow 脚本**: `<this-skill>/workflows/wf-swarm-*.js`
  - 运行时解析: `Glob(".claude/skills/team-adversarial-swarm/workflows/wf-swarm-*.js")`

## Specs Reference

| Spec | Purpose |
|------|---------|
| [specs/swarm-protocol.md](specs/swarm-protocol.md) | Coordinator ↔ Script ↔ Workflow 三方协议 |
| [specs/pheromone-schema.md](specs/pheromone-schema.md) | 信息素矩阵结构、更新公式、蒸发规则 |
| [specs/ant-output-schema.md](specs/ant-output-schema.md) | 蚁输出 JSON 合约（三层评分） |
| [specs/convergence-criteria.md](specs/convergence-criteria.md) | 双层收敛：Python 信号 + 对抗辩论 |
| [specs/swarm-config-template.json](specs/swarm-config-template.json) | 用户配置模板 |

## 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
├── {run_dir}/outputs/      # Formal deliverables
│   ├── ant-<k>-<id>.json   # Ant outputs
│   └── best-solution.md    # Final synthesis
├── 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

1. `Glob("{run_dir}/work/team/swarm-config.json")` → 查找活跃 session
2. 若存在且有 `workflows/converge-*.json` 未标记 converged → 恢复到对应迭代
3. 若无活跃 session → Phase 1

### Phase 1: Config Generation

解析用户 intent，生成 `swarm-config.json`。

若 intent 不够明确，用 request_user_input 澄清：
- 搜索空间是什么？（文件 glob / 节点列表 / 抽象决策集）
- 目标是什么？（最优方案 / 发现问题 / 优化路径）
- 如何评分？（测试通过率 / lint / 自定义规则 / LLM 对抗评分）
- 预算？（最大迭代 / 每轮蚁数 / token 预算）

生成 config 字段：
```json
{
  "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

1. 创建 session 目录: `TAS-<slug>-<date>`
2. 解析 aco.py 路径（从 team-swarm skill 继承）
3. `Bash: python <aco.py> --session {run_dir}/work/team init`
4. 解析输出: `{ n_nodes, n_edges, pheromone_path }`

### Run Lifecycle Integration

After session folder creation and before role-spec generation:

1. **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>"`
   - Slug format: `YYYYMMDD-team-adversarial-swarm-<topic>` (ASCII, ≤64 chars)
   - Store returned `run_id` and `run_dir` in `team-session.json`:
     ```json
     "run": { "run_id": "<id>", "run_dir": "<path>" }
     ```
2. **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. If `run.run_id` is missing, resolve in order: birth-packet injection, then `<session>/artifacts/`; if all are absent, fail closed — report session corruption and do NOT create a new Run.

### Phase 3: Iteration Loop

```python
for k in range(1, max_iterations + 1):
    # 3a. ACO selection
    assignments = Bash("python aco.py --session {run_dir}/work/team select --iter k")
    
    # 3b. Parallel exploration (Workflow Module 1)
    explore_result = Workflow({
        scriptPath: "<skill>/workflows/wf-swarm-explore.js",
        args: { iteration: k, assignments, objective, session, config }
    })
    
    # 3c. Adversarial scoring (Workflow Module 2)
    score_result = Workflow({
        scriptPath: "<skill>/workflows/wf-swarm-score.js",
        args: { iteration: k, ant_results: explore_result.ant_results, objective, rubric }
    })
    
    # 3d. Write scores + pheromone update
    Write("{run_dir}/work/team/scores/iter-k-scores.json", score_result)
    Bash("python aco.py --session {run_dir}/work/team --run-dir <run_dir> update --iter k")
    
    # 3e. Adversarial convergence check (Workflow Module 3)
    converge_result = Workflow({
        scriptPath: "<skill>/workflows/wf-swarm-converge.js",
        args: { iteration: k, best: aco_best, history: iter_history, config }
    })
    
    # 3f. Save + check
    Write("{run_dir}/work/team/workflows/converge-k.json", converge_result)
    if converge_result.converged: break
```

**注意**：每次 Workflow 调用是独立的，数据通过 args 传入、返回值传出。
Coordinator 负责 Workflow 间的数据桥接和 Python 脚本调用。

### Phase 4: Synthesis

1. `Bash: python aco.py --session {run_dir}/work/team report` → 获取 best + top_k + curve
2. 调用 Workflow Module 4:
   ```
   Workflow({
     scriptPath: "<skill>/workflows/wf-swarm-synthesize.js",
     args: { best, top_k, convergence_story, objective }
   })
   ```
3. 将 synthesis 结果写入 `{run_dir}/outputs/best-solution.md`
4. 展示完成摘要 + request_user_input（归档 / 保留 / 导出 / 再跑一轮）

---

## 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 恢复点 |
| 所有蚁全部失败 | 暂停，request_user_input（重试/终止/调整config） |
| 收敛从不触发 | max_iterations 安全网总会触发 |
| 幻觉集群 (>50% 蚁被降分) | 暂停，request_user_input（继续/调整评分规则） |

## 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 session done <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 轮