IB-WP-0019-T06: workspace budget CLI

infospace-bench budget list <workspace> walks <workspace>/infospaces/*
and prints one row per infospace with slug, plans_count, runs_count,
total_tokens, total_cost_usd_known, total_cost_usd_estimated,
last_run_at, and latest_snapshot_id. infospace-bench budget show
<root> dumps the full plans/usage/summary structure for a single
infospace.

Missing budget directories are treated as zero rows rather than errors,
so the CLI is safe to run on partially-populated or fresh workspaces.

120 tests pass.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
tegwick 2026-05-17 20:44:40 +02:00
parent 110c78b9ad
commit 816a95b3ef
4 changed files with 190 additions and 1 deletions

View file

@ -375,6 +375,75 @@ def _post_json(url: str, payload: dict[str, Any], timeout: float) -> None:
response.read()
def budget_show(root: str | Path) -> dict[str, Any]:
"""Return the full per-infospace budget structure for inspection."""
root_path = Path(root)
plans_payload = _read_plans(root_path / PLANS_FILE)
usage_payload = _read_usage(root_path / USAGE_FILE)
variance = read_run_variance(root_path)
return {
"root": str(root_path),
"plans": {
"schema_version": plans_payload.get("schema_version"),
"pruned_count": plans_payload.get("pruned_count") or 0,
"snapshots": list(plans_payload.get("snapshots") or []),
},
"usage": {
"schema_version": usage_payload.get("schema_version"),
"runs": list(usage_payload.get("runs") or []),
},
"summary": variance,
}
def budget_list_workspace(workspace: str | Path) -> list[dict[str, Any]]:
"""Walk ``<workspace>/infospaces/*/output/budget/`` and roll up each infospace.
Missing budget directories are treated as zero, not as errors. Returns a
list of summaries in slug order.
"""
workspace_path = Path(workspace)
infospaces_dir = workspace_path / "infospaces"
if not infospaces_dir.is_dir():
return []
rollups: list[dict[str, Any]] = []
for entry in sorted(infospaces_dir.iterdir()):
if not entry.is_dir():
continue
rollups.append(_summarise_infospace(entry))
return rollups
def _summarise_infospace(root: Path) -> dict[str, Any]:
plans = read_plan_snapshots(root)
runs = read_usage_runs(root)
total_tokens = 0
total_cost_known = 0.0
total_cost_estimated = 0.0
last_run_at: str | None = None
for run in runs:
rollup = run.get("rollup") or {}
total_tokens += int(rollup.get("total_tokens") or 0)
total_cost_known += _coerce_float(rollup.get("total_cost_usd_known")) or 0.0
estimated = _coerce_float(rollup.get("total_cost_usd_estimated"))
if estimated is not None:
total_cost_estimated += estimated
completed_at = run.get("completed_at")
if isinstance(completed_at, str) and (last_run_at is None or completed_at > last_run_at):
last_run_at = completed_at
return {
"slug": root.name,
"root": str(root),
"plans_count": len(plans),
"runs_count": len(runs),
"total_tokens": total_tokens,
"total_cost_usd_known": round(total_cost_known, 6),
"total_cost_usd_estimated": round(total_cost_estimated, 6) if total_cost_estimated else None,
"last_run_at": last_run_at,
"latest_snapshot_id": plans[-1].get("snapshot_id") if plans else None,
}
def read_run_variance(root: str | Path) -> dict[str, Any] | None:
path = Path(root) / SUMMARY_FILE
if not path.is_file():