diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 9e76ea6..057dc99 100644 --- a/.github/workflows/ci.yml +++ b/.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 diff --git a/docs/integration/usage-billing-schema.md b/docs/integration/usage-billing-schema.md index 597716c..4e32902 100644 --- a/docs/integration/usage-billing-schema.md +++ b/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) | diff --git a/orchestrator/main.py b/orchestrator/main.py index 2b411f7..57b0ce9 100644 --- a/orchestrator/main.py +++ b/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, diff --git a/scripts/test-usage-cost-phase.py b/scripts/test-usage-cost-phase.py new file mode 100644 index 0000000..12716d6 --- /dev/null +++ b/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())