IB-WP-0019-T07: archive integration; close IB-WP-0019

The default archive include set already pulls output/ in wholesale, so
output/budget/ already lands inside the archive package with no code
change. Add a budget_summary block to ArchiveRecord.metadata so
catalog-level tools can see plans_count, runs_count, total_tokens,
total_cost_usd_known, total_cost_usd_estimated, and the
latest_snapshot_id without unpacking the archive. An infospace with no
budget data still archives cleanly with an empty metadata dict.

Closes IB-WP-0019 (Budget and Usage Registry): T01-T07 all done.
Three-layer design landed end-to-end — layer 1 (per-infospace
plans.yaml / usage.yaml / summary.yaml) and layer 3 (state-hub
record_token_event emission with failure isolation) live here; layer 2
(cross-application QualityLedger for adaptive routing) is parked in
llm-connect LLM-WP-0004 and infospace-bench IB-WP-0018 awaits it.

122 tests pass.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-17 21:53:28 +02:00
parent 816a95b3ef
commit bb70b2f4b9
3 changed files with 89 additions and 2 deletions

View File

@@ -337,12 +337,65 @@ async def _archive_infospace_async(
producer=PRODUCER,
subject=subject,
store_root=str(effective_store_root) if effective_store_root else None,
metadata=_archive_metadata(root),
skipped_top_level=skipped_top_level,
)
_append_index(root, record)
return record
def _archive_metadata(root: Path) -> dict[str, Any]:
"""Compute a small budget_summary for the archive manifest.
Lets catalog-level tools find an archived infospace's cost shape without
unpacking it. Returns an empty dict when no budget data exists.
"""
from .budget import (
read_plan_snapshots,
read_run_variance,
read_usage_runs,
)
try:
plans = read_plan_snapshots(root)
runs = read_usage_runs(root)
summary = read_run_variance(root)
except Exception:
return {}
if not plans and not runs and summary is None:
return {}
total_tokens = 0
total_cost_known = 0.0
total_cost_estimated = 0.0
for run in runs:
rollup = run.get("rollup") or {}
total_tokens += int(rollup.get("total_tokens") or 0)
try:
total_cost_known += float(rollup.get("total_cost_usd_known") or 0.0)
except (TypeError, ValueError):
pass
estimated = rollup.get("total_cost_usd_estimated")
if estimated is not None:
try:
total_cost_estimated += float(estimated)
except (TypeError, ValueError):
pass
budget_summary: dict[str, Any] = {
"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,
}
if plans:
budget_summary["latest_snapshot_id"] = plans[-1].get("snapshot_id")
if summary is not None and isinstance(summary.get("snapshot_id"), str):
budget_summary["last_run_snapshot_id"] = summary.get("snapshot_id")
return {"budget_summary": budget_summary}
def _collect_files(
root: Path,
*,