)\d+(<)',
f'\\g<1>{compact}\\2',
content
)
# Patch: Compaction events util sub
content = re.sub(
r'(Session \d+ total • Triggered at ~78K)',
f'Session {session_num} total • Triggered at ~78K',
content
)
# Patch: Context utilisation value
content = re.sub(
r'(
)[\d]+%(<)',
f'\\g<1>{ctx_pct}%\\2',
content
)
# Patch: Knowledge Graph count
content = re.sub(
r'(
Knowledge Graph
\s*
)[\d+]+(<)',
f'\\g<1>{kg_total}+\\2',
content
)
# Patch: KG sub text with evolve cycle
content = re.sub(
r'(Entities • Axioms • Evolve cycles) [\d-]+',
f"Entities • Axioms • Evolve cycle {metrics['kg_evolve_cycle']}",
content
)
# Patch: Footer timestamp
content = re.sub(
r'(Built )\d{4}-\d{2}-\d{2}',
f'Built 2026-02-20 | Curated {aest_time}',
content
)
if content == original:
log.info("GOLD_MASTER_DASHBOARD — no changes needed (data already current)")
return True
if dry_run:
log.info("[DRY RUN] Would patch GOLD_MASTER_DASHBOARD.html with %d char changes",
sum(a != b for a, b in zip(content, original)))
return True
GOLD_DASHBOARD.write_text(content, encoding="utf-8")
log.info("GOLD_MASTER_DASHBOARD.html patched — session=%d cost=$%.2f compaction=%d",
session_num, session_cost, compact)
return True
# ══════════════════════════════════════════════════════════════════════════════
# JSON METRICS FILE (for browser fetch)
# ══════════════════════════════════════════════════════════════════════════════
def write_metrics_json(metrics: dict, dry_run: bool = False) -> bool:
"""
Write metrics to dashboard/genesis-metrics.json for dashboard JS to fetch.
This is the lowest-overhead way to update dashboards without a running server.
"""
output_path = DASHBOARD_DIR / "genesis-metrics.json"
if dry_run:
log.info("[DRY RUN] Would write: %s", output_path)
return True
try:
output_path.write_text(json.dumps(metrics, indent=2), encoding="utf-8")
log.info("Metrics JSON written: %s", output_path)
return True
except Exception as e:
log.error("Failed to write metrics JSON: %s", e)
return False
# ══════════════════════════════════════════════════════════════════════════════
# STATUS REPORT
# ══════════════════════════════════════════════════════════════════════════════
def format_status_report(metrics: dict) -> str:
"""Generate a human-readable status report."""
return f"""
=== GENESIS DASHBOARD CURATOR STATUS ===
Generated: {metrics['captured_at_aest']}
SESSION
Number: {metrics['session_number']}
Cost: ${metrics['session_cost_usd']:.2f} USD
Compactions: {metrics['compaction_events']}
Context used: {metrics['context_used_pct']}%
SWARM EXECUTION
Total stories: {metrics['swarm_stories_total']}
Successful: {metrics['swarm_stories_success']} ({metrics['swarm_success_rate_pct']}%)
Failed: {metrics['swarm_stories_failed']}
Tokens used: {metrics['swarm_tokens_total']:,}
Swarm cost: ${metrics['swarm_cost_usd']:.4f} USD
Cost/story: ${metrics['cost_per_story']:.4f} USD
MiniMax stories: {metrics['minimax_stories']}
Kimi stories: {metrics['kimi_stories']}
AGENTS
Active: {metrics['active_agents']}
Total spawns: {metrics['total_spawns']}
Utilisation: {metrics['utilisation_pct']}% of 4,650 capacity/day
KNOWLEDGE GRAPH
Entities: {metrics['kg_entities']}
Axioms: {metrics['kg_axioms']}
Evolve cycle: {metrics['kg_evolve_cycle']}
BURN RATE
Monthly: ${metrics['monthly_burn_usd']:,} USD / ~${metrics['monthly_burn_aud']:,} AUD
Break-even: {metrics['break_even_customers']} customers
=========================================
""".strip()
# ══════════════════════════════════════════════════════════════════════════════
# MAIN
# ══════════════════════════════════════════════════════════════════════════════
def main():
parser = argparse.ArgumentParser(description="Genesis Dashboard Curator")
parser.add_argument("--dry-run", action="store_true",
help="Print what would happen, write nothing")
parser.add_argument("--redis-only", action="store_true",
help="Only write to Redis, skip HTML patching")
parser.add_argument("--json-only", action="store_true",
help="Only write genesis-metrics.json, skip Redis and HTML")
parser.add_argument("--report", action="store_true",
help="Print status report to stdout and exit")
args = parser.parse_args()
log.info("Genesis Dashboard Curator v1.0.0 starting...")
# Build metrics
metrics = build_full_metrics()
if args.report:
print(format_status_report(metrics))
return 0
log.info("Metrics extracted: session=%d cost=$%.2f stories=%d",
metrics["session_number"],
metrics["session_cost_usd"],
metrics["swarm_stories_total"])
results = []
# Always write JSON (zero-dependency, enables browser fetch)
if not args.redis_only:
ok = write_metrics_json(metrics, dry_run=args.dry_run)
results.append(("JSON metrics", ok))
# Patch HTML dashboards
if not args.redis_only and not args.json_only:
ok = patch_gold_dashboard(metrics, dry_run=args.dry_run)
results.append(("GOLD_MASTER_DASHBOARD patch", ok))
# Write to Redis (best-effort)
if not args.json_only:
ok = write_to_redis(metrics, dry_run=args.dry_run)
results.append(("Redis write", ok))
# Summary
passed = sum(1 for _, ok in results if ok)
failed = len(results) - passed
log.info("Curator complete: %d/%d tasks succeeded", passed, len(results))
for name, ok in results:
status = "OK" if ok else "FAILED"
log.info(" [%s] %s", status, name)
if args.dry_run:
print("\n" + format_status_report(metrics))
return 0 if failed == 0 else 1
if __name__ == "__main__":
sys.exit(main())