scripts.context_usage_report

Reference › scripts › context_usage_report

Aggregates the context-token spend recorded by the context layer, per layer, per Volk and per subject. tests/scripts/test_context_usage_report.py imports it, and it uses read_usage_records and aggregate_usage from usage_log.

Invocation: main takes --log-path (default the ORCHESTRATA_CONTEXT_USAGE_LOG variable, else logs/context_usage.jsonl) and --format (table, the default, or json). It reads the JSONL log, aggregates it and prints the report with _format_table or as JSON. It writes nothing and returns 0.

At a glance

Class diagram

No classes; see Functions.

Dependency diagram

API

Functions

Source

Permalink: https://gitlab.com/MSamy/orchestrata-ultimate-architecture/-/blob/a4cbc2bdf/scripts/context_usage_report.py

#!/usr/bin/env python3
"""Aggregate per-layer, per-volk, and per-subject context token spend.

Reads ``logs/context_usage.jsonl`` (or the file pointed to by
``ORCHESTRATA_CONTEXT_USAGE_LOG``) and emits a summary report.

Usage:
    python scripts/context_usage_report.py
    python scripts/context_usage_report.py --log-path logs/context_usage.jsonl \
        --format table
    python scripts/context_usage_report.py --format json
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from pathlib import Path
from typing import Any

from orchestrata_ultimate.context.usage_log import (
    aggregate_usage,
    read_usage_records,
)


def _format_table(report: dict[str, Any]) -> str:
    lines: list[str] = []
    lines.append("Context Usage Report")
    lines.append("=" * 60)
    lines.append(f"Total calls: {report['total_calls']}")
    lines.append(f"Total tokens spent: {report['total_spent']}")
    lines.append("")
    lines.append("Per layer:")
    lines.extend(
        f"  {layer:12s} {spent:>10d}"
        for layer, spent in sorted(report["per_layer"].items())
    )
    lines.append("")
    lines.append("Per volk:")
    lines.extend(
        f"  {volk:12s} {spent:>10d}"
        for volk, spent in sorted(report["per_volk"].items())
    )
    lines.append("")
    lines.append("Per subject:")
    lines.extend(
        f"  {subject:12s} {spent:>10d}"
        for subject, spent in sorted(report["per_subject"].items())
    )
    return "\n".join(lines)


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Aggregate context-usage token spend.",
    )
    parser.add_argument(
        "--log-path",
        default=os.environ.get(
            "ORCHESTRATA_CONTEXT_USAGE_LOG", "logs/context_usage.jsonl"
        ),
        help="Path to the JSONL usage log (default: logs/context_usage.jsonl).",
    )
    parser.add_argument(
        "--format",
        choices=["json", "table"],
        default="table",
        help="Output format (default: table).",
    )
    args = parser.parse_args(argv)

    path = Path(args.log_path)
    records = read_usage_records(path)
    report = aggregate_usage(records)

    if args.format == "json":
        print(json.dumps(report, indent=2))
    else:
        print(_format_table(report))

    return 0


if __name__ == "__main__":
    sys.exit(main())