/metrics 暴露 run 级 cost_by_phase{initial,review_retry}（Refs #16 #37）

回应 @Mem0ried 在 PR #37 的对接问题：除逐条 budget.alert 事件的 cost_phase 外，详情
聚合也需能拆 initial/review_retry。build_runtime_metrics 新增 cost_by_phase 滚动汇总
（按 swarm_id，从各任务 usage.model_cost_usd/model_tokens 按 rework_attributions 归类），
GET …/{id}/metrics 返回 {initial,review_retry}:{cost_usd,model_tokens}。HM 详情聚合 /
客户端 08 用量抽屉可直接展示拆分，无需从事件推导。

测试 test-usage-cost-phase.py 增 /metrics 断言（reworked→review_retry、非 reworked→initial）；
usage-billing §5 记录该字段。

Refs #16
Refs #37

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

docs/integration/usage-billing-schema.md

@@ -87,6 +87,7 @@ PayPal 说明 §5 给出的运行时用量回传目标：
 
 - 一次 swarm 请求拆成多 task / 多 agent / 多评审轮；**每个 task 的 `usage` 可按 `swarm_id` 聚合**得到运行级合计（观测用途）。
 - **评审重做成本（#16，已打标）**：每轮重做重新执行任务并累计 `usage`，且其用量事件带 `cost_phase="review_retry"`（cross-review 已把该任务记入 `rework_attributions` 之后的再执行）；初次执行为 `cost_phase="initial"`。计费/采集侧按 `swarm_id` 聚合时，可用 `cost_phase` 拆分 **initial vs review_retry 成本**——「review retry 成本可归属」由此满足，无需从 `retry_count` 反推。
+- **详情聚合（#37）**：除逐条事件的 `cost_phase` 外，`GET …/{id}/metrics` 另返回 run 级滚动汇总 `cost_by_phase: { initial: {cost_usd, model_tokens}, review_retry: {cost_usd, model_tokens} }`，供 HM 详情用量聚合 / 客户端用量抽屉直接展示「初次 vs 返工」拆分，无需从事件流推导。
 
 ## 6. 字段覆盖与缺口
 

orchestrator/main.py

@@ -1886,6 +1886,18 @@ async def build_runtime_metrics(run, window: str, step: str) -> Dict[str, Any]:
     budget = ((run.request_body.get("orchestration_plan") or {}).get("budget") or {})
     duration_budget = budget.get("duration_seconds") or budget.get("max_duration_seconds")
     budget_ratio = duration_seconds / float(duration_budget) if duration_budget else None
+    # Run-level cost split by phase (#16/#37): lets HM's detail usage aggregation show
+    # initial vs review_retry without deriving from the event stream. Same attribution as
+    # emit_usage_event — a task counts as review_retry once cross-review recorded it in
+    # run.metadata["rework_attributions"].
+    rework_targets = {a.get("target_task_id") for a in (run.metadata.get("rework_attributions") or [])}
+    cost_by_phase = {"initial": {"cost_usd": 0.0, "model_tokens": 0},
+                     "review_retry": {"cost_usd": 0.0, "model_tokens": 0}}
+    for task in tasks:
+        usage = (parse_task_result(task) or {}).get("usage") or {}
+        phase = "review_retry" if task.task_id in rework_targets else "initial"
+        cost_by_phase[phase]["cost_usd"] += float(usage.get("model_cost_usd") or 0.0)
+        cost_by_phase[phase]["model_tokens"] += int(usage.get("model_tokens") or 0)
     return {
         "deployment_id": run.deployment_id,
         "swarm_id": run.swarm_id,
@@ -1903,6 +1915,7 @@ async def build_runtime_metrics(run, window: str, step: str) -> Dict[str, Any]:
             sum(completed_durations) / len(completed_durations)
             if completed_durations else 0
         ),
+        "cost_by_phase": cost_by_phase,   # {initial,review_retry}:{cost_usd,model_tokens} (#16/#37)
         "budget": {
             "duration_seconds": duration_budget,
             "duration_ratio": budget_ratio,

scripts/test-usage-cost-phase.py

@@ -78,6 +78,18 @@ async def main():
     t2_ev = [e for e in evs if e["payload"].get("task_id") == t2.task_id][-1]
     check("non-reworked task stays cost_phase=initial", t2_ev["payload"].get("cost_phase") == "initial")
 
+    # --- /metrics run-level cost_by_phase rollup (#37) ---
+    # Persist task results with usage so build_runtime_metrics can split them; t1 is a rework target.
+    await task_queue.complete_task(t1.task_id, result=json.dumps({"usage": {"model_cost_usd": 0.05, "model_tokens": 500}}))
+    await task_queue.complete_task(t2.task_id, result=json.dumps({"usage": {"model_cost_usd": 0.02, "model_tokens": 200}}))
+    refreshed = await swarm_runtime.get_run(run.swarm_id)
+    metrics = await orch.build_runtime_metrics(refreshed, "15m", "60s")
+    cbp = metrics.get("cost_by_phase") or {}
+    check("/metrics exposes cost_by_phase", set(cbp.keys()) == {"initial", "review_retry"})
+    check("/metrics review_retry cost = reworked task (t1=0.05)", abs(cbp["review_retry"]["cost_usd"] - 0.05) < 1e-9)
+    check("/metrics initial cost = non-reworked task (t2=0.02)", abs(cbp["initial"]["cost_usd"] - 0.02) < 1e-9)
+    check("/metrics review_retry tokens = 500", cbp["review_retry"]["model_tokens"] == 500)
+
     print()
     if failures:
         print(f"{len(failures)} usage cost-phase check(s) FAILED: {failures}")