Merge pull request #37 from xmindlab-heicode/feat/usage-cost-phase

用量按 review-retry 成本归属：usage 事件标 cost_phase（Refs #16）

.github/workflows/ci.yml

@@ -66,6 +66,10 @@ jobs:
         env: { REDIS_FAKE: "1" }
         run: python scripts/test-contract-events.py
 
+      - name: Usage review-retry cost attribution (#16)
+        env: { REDIS_FAKE: "1" }
+        run: python scripts/test-usage-cost-phase.py
+
       - name: Audit / lineage trace (replayable) (#17)
         env: { REDIS_FAKE: "1" }
         run: python scripts/test-audit-trace.py

docs/integration/usage-billing-schema.md

@@ -62,6 +62,8 @@ Swarm 在运行中按时长/成本比例发 `budget.alert`（默认 80% 阈值
 {
   "model_id": "...", "model_tokens": 0, "prompt_tokens": 0, "completion_tokens": 0,
   "model_cost_usd": 0.0, "runtime_seconds": 0.0, "billing_source": "...",
+  "cost_phase": "initial | review_retry",   // #16：返工再执行的用量标 review_retry，供成本归属
+  "attempt": 0,                              // 该任务的重做次数（retry_count）
   "manager_deployment_id": "...", "swarm_id": "...", "task_id": "...",
   "agent_role": "...", "correlation_id": "...",
   "budget": { "max_tokens": null, "max_cost_usd": null, "consumed_usd": 0.0, "remaining_usd": null }
@@ -84,7 +86,8 @@ PayPal 说明 §5 给出的运行时用量回传目标：
 ## 5. 多 Agent / 评审重做 成本聚合
 
 - 一次 swarm 请求拆成多 task / 多 agent / 多评审轮；**每个 task 的 `usage` 可按 `swarm_id` 聚合**得到运行级合计（观测用途）。
-- **评审重做成本**：每轮重做都会重新执行任务并累计 `usage`，因此**已隐含计入**运行级合计；但当前**未单独打标**「review_retry 成本」。
+- **评审重做成本（#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. 字段覆盖与缺口
 
@@ -96,7 +99,7 @@ PayPal 说明 §5 给出的运行时用量回传目标：
 | `reasoning_tokens` / `cache_tokens` | 🔴 未采集（取决于 provider usage 返回） |
 | `tool_cost`（工具调用成本） | 🔴 未采集（无 SK 工具计量） |
 | `cpu_core_seconds` / `memory_mb_seconds`（基础设施） | 🔴 未采集（K8s 指标未接入账本） |
-| `review_retry` 成本单独打标 | 🟡 隐含累计，未单独标注 |
+| `review_retry` 成本单独打标 | ✅ 已打标（usage 事件 `cost_phase=review_retry`，#16） |
 | provider 成本拆分 | 🟡 由 NewAPI/账本侧负责，非本仓 |
 | tenant 归因 | ❌ 按标准不使用（见 §1） |
 

orchestrator/main.py

@@ -1739,6 +1739,15 @@ async def emit_usage_event(run, task, agent_id: str, result: Any):
 
     plan = run.request_body.get("orchestration_plan") or {}
     budget = plan.get("budget") or {}
+    # Review-retry cost attribution (#16): a task's usage is `review_retry` once cross-review has
+    # reopened it for rework (it appears in run.metadata["rework_attributions"]); otherwise
+    # `initial`. The first execution emits before any attribution exists → "initial"; each redo
+    # emits after run_cross_review recorded the attribution → "review_retry". So billing can split
+    # initial vs review-retry cost by summing usage events per cost_phase (usage-billing §5).
+    rework_targets = {
+        a.get("target_task_id") for a in (run.metadata.get("rework_attributions") or [])
+    }
+    cost_phase = "review_retry" if task.task_id in rework_targets else "initial"
     payload = {
         "model_id": usage.get("model_id") or task.context.get("model_id"),
         "model_tokens": usage.get("model_tokens", 0),
@@ -1747,6 +1756,8 @@ async def emit_usage_event(run, task, agent_id: str, result: Any):
         "model_cost_usd": usage.get("model_cost_usd", 0),
         "runtime_seconds": usage.get("runtime_seconds", 0),
         "billing_source": usage.get("billing_source", "unknown"),
+        "cost_phase": cost_phase,          # "initial" | "review_retry" (#16 attribution)
+        "attempt": task.retry_count,       # redo count for this task
         "manager_deployment_id": run.manager_deployment_id,
         "swarm_id": run.swarm_id,
         "task_id": task.task_id,
@@ -1913,6 +1924,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,
@@ -1930,6 +1953,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

@@ -0,0 +1,101 @@
+"""Usage review-retry cost attribution test (issue #16).
+
+Verifies emit_usage_event tags each usage event with `cost_phase`:
+  * "initial" before cross-review has reopened the task for rework;
+  * "review_retry" once the task appears in run.metadata["rework_attributions"]
+    (i.e. its re-execution cost is attributable as review-retry cost).
+
+Hermetic: REDIS_FAKE, no model key, no callback url.
+
+Run from agent_swarm_v6 (install deps first):
+    pip install -r orchestrator/requirements.txt
+    REDIS_FAKE=1 python scripts/test-usage-cost-phase.py
+"""
+import asyncio
+import json
+import os
+import sys
+from pathlib import Path
+
+os.environ["REDIS_FAKE"] = "1"
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+failures = []
+
+
+def check(name, cond):
+    print(("PASS" if cond else "FAIL"), "-", name)
+    if not cond:
+        failures.append(name)
+
+
+async def budget_events(swarm_runtime, swarm_id):
+    raw = await swarm_runtime.list_events(swarm_id, limit=500)
+    return [e for e in raw["events"] if e["event_type"] == "budget.alert"]
+
+
+async def main():
+    from orchestrator.redis_client import redis_client
+    from orchestrator.swarm_runtime import swarm_runtime
+    from orchestrator.task_queue import task_queue
+    from orchestrator import main as orch
+
+    await redis_client.connect()
+    body = {"mode": "swarm", "orchestration_plan": {"objective": "cost phase"},
+            "callback": {"url": "", "subscribed_events": []},
+            "metadata": {"manager_deployment_id": "m-cost"}}
+    run, _ = await swarm_runtime.get_or_create_run(body=body, idempotency_key=None, correlation_id="c-cost")
+    t1 = await task_queue.create_task(task_id=f"{run.swarm_id}-t1", description="impl",
+                                      agent_role="impl", required_capabilities=["python"], enqueue=True)
+    await swarm_runtime.attach_task(run, t1.task_id)
+
+    usage_result = {"usage": {"model_id": "gpt-x", "model_tokens": 100, "model_cost_usd": 0.01,
+                              "billing_source": "newapi"}}
+
+    # 1) Initial execution: no rework attribution yet → cost_phase=initial.
+    await orch.emit_usage_event(run, t1, "agent-1", usage_result)
+    evs = await budget_events(swarm_runtime, run.swarm_id)
+    check("initial usage tagged cost_phase=initial", evs and evs[-1]["payload"].get("cost_phase") == "initial")
+    check("usage payload carries attempt", "attempt" in evs[-1]["payload"])
+
+    # 2) Cross-review reopened t1 for rework → recorded in rework_attributions.
+    run.metadata["rework_attributions"] = [{"target_task_id": t1.task_id, "root_cause": "test"}]
+    await swarm_runtime.save_run(run)
+    t1.retry_count = 1
+    await orch.emit_usage_event(run, t1, "agent-1", usage_result)
+    evs = await budget_events(swarm_runtime, run.swarm_id)
+    check("redo usage tagged cost_phase=review_retry", evs[-1]["payload"].get("cost_phase") == "review_retry")
+    check("review_retry usage attributable by swarm_id + cost_phase",
+          any(e["payload"].get("cost_phase") == "review_retry" and e["payload"].get("swarm_id") == run.swarm_id
+              for e in evs))
+
+    # A non-reworked task stays initial even alongside the reworked one.
+    t2 = await task_queue.create_task(task_id=f"{run.swarm_id}-t2", description="doc",
+                                      agent_role="documentation", required_capabilities=["technical-writing"], enqueue=True)
+    await swarm_runtime.attach_task(run, t2.task_id)
+    await orch.emit_usage_event(run, t2, "agent-2", usage_result)
+    evs = await budget_events(swarm_runtime, run.swarm_id)
+    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}")
+        sys.exit(1)
+    print("all usage cost-phase checks passed")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())