#!/usr/bin/env python3 """ CODITECT Cost Report Command (H.8.4.7)

CLI implementation for /cost-report — token cost attribution by task, track, and session using the Ralph Wiggum TokenEconomicsService (ADR-111).

Usage: python3 cost-report-command.py python3 cost-report-command.py --budgets python3 cost-report-command.py --from 2026-02-10 --to 2026-02-16 python3 cost-report-command.py --track H python3 cost-report-command.py --models python3 cost-report-command.py --efficiency python3 cost-report-command.py --json

Author: CODITECT Framework Version: 1.0.0 Created: 2026-02-16 Task Reference: H.8.4.7 ADR Reference: ADR-111 """

import argparse import json import os import sys from datetime import datetime, date, timedelta, timezone from pathlib import Path from collections import defaultdict

Add parent directories to path

SCRIPT_DIR = Path(file).resolve().parent CORE_DIR = SCRIPT_DIR.parent.parent sys.path.insert(0, str(CORE_DIR)) sys.path.insert(0, str(SCRIPT_DIR))

Data paths

DATA_DIR = Path.home() / "PROJECTS" / ".coditect-data" / "token-economics" RECORDS_DIR = DATA_DIR / "records" TOTALS_FILE = DATA_DIR / "running_totals.json" BUDGETS_FILE = DATA_DIR / "budgets.json"

Default pricing (from ADR-111 / token_economics.py)

DEFAULT_PRICING = { "claude-opus-4-6": {"input": 15.00, "output": 75.00}, "claude-opus-4-5": {"input": 15.00, "output": 75.00}, "claude-sonnet-4-5": {"input": 3.00, "output": 15.00}, "claude-haiku-4-5": {"input": 0.80, "output": 4.00}, }

def parse_args(): parser = argparse.ArgumentParser( description="CODITECT Cost Report (ADR-111)", formatter_class=argparse.RawDescriptionHelpFormatter, ) parser.add_argument("--from", dest="from_date", help="Start date (YYYY-MM-DD)") parser.add_argument("--to", dest="to_date", help="End date (YYYY-MM-DD)") parser.add_argument("--track", help="Filter by track letter (e.g., H)") parser.add_argument("--task", help="Filter by task ID (e.g., H.8.4.7)") parser.add_argument("--budgets", action="store_true", help="Show budget utilization") parser.add_argument("--models", action="store_true", help="Show per-model breakdown") parser.add_argument("--efficiency", action="store_true", help="Show efficiency metrics") parser.add_argument("--json", action="store_true", help="JSON output") return parser.parse_args()

def load_records(from_date=None, to_date=None): """Load consumption records from JSONL files.""" records = [] if not RECORDS_DIR.exists(): return records

today = date.today()
start = datetime.strptime(from_date, "%Y-%m-%d").date() if from_date else today
end = datetime.strptime(to_date, "%Y-%m-%d").date() if to_date else today

current = start
while current <= end:
    record_file = RECORDS_DIR / f"{current.isoformat()}.jsonl"
    if record_file.exists():
        for line in record_file.read_text().strip().split("\n"):
            if line.strip():
                try:
                    records.append(json.loads(line))
                except json.JSONDecodeError:
                    continue
    current += timedelta(days=1)

return records

def load_running_totals(): """Load running totals.""" if TOTALS_FILE.exists(): try: return json.loads(TOTALS_FILE.read_text()) except json.JSONDecodeError: pass return {}

def load_budgets(): """Load budget definitions.""" if BUDGETS_FILE.exists(): try: return json.loads(BUDGETS_FILE.read_text()) except json.JSONDecodeError: pass return {}

def calculate_cost(input_tokens, output_tokens, model="claude-opus-4-6"): """Calculate cost for given token counts.""" pricing = DEFAULT_PRICING.get(model, DEFAULT_PRICING["claude-opus-4-6"]) input_cost = (input_tokens / 1_000_000) * pricing["input"] output_cost = (output_tokens / 1_000_000) * pricing["output"] return input_cost + output_cost

def filter_records(records, track=None, task=None): """Filter records by track or task.""" filtered = records if track: filtered = [r for r in filtered if r.get("task_id", "").startswith(f"{track}.")] if task: filtered = [r for r in filtered if r.get("task_id", "").startswith(task)] return filtered

def format_tokens(count): """Format token count for display.""" if count >= 1_000_000: return f"{count / 1_000_000:.1f}M" if count >= 1_000: return f"{count / 1_000:.0f}K" return str(count)

def display_summary(records, from_date, to_date, args): """Display cost summary.""" if not records: print("No token consumption records found for the specified period.") print(f"Data directory: {DATA_DIR}") print("\nToken tracking is recorded by the TokenEconomicsService (ADR-111).") print("Records are stored in: ~/PROJECTS/.coditect-data/token-economics/records/") return

# Aggregate by model
by_model = defaultdict(lambda: {"input": 0, "output": 0, "cost": 0})
by_track = defaultdict(lambda: {"input": 0, "output": 0, "cost": 0})
by_task = defaultdict(lambda: {"input": 0, "output": 0, "cost": 0})
total_input = 0
total_output = 0
total_cost = 0

for r in records:
    model = r.get("model", "claude-opus-4-6")
    inp = r.get("input_tokens", 0)
    out = r.get("output_tokens", 0)
    cost = r.get("cost", calculate_cost(inp, out, model))
    task_id = r.get("task_id", "unknown")
    track_letter = task_id.split(".")[0] if "." in task_id else task_id

    total_input += inp
    total_output += out
    total_cost += cost

    by_model[model]["input"] += inp
    by_model[model]["output"] += out
    by_model[model]["cost"] += cost

    by_track[track_letter]["input"] += inp
    by_track[track_letter]["output"] += out
    by_track[track_letter]["cost"] += cost

    by_task[task_id]["input"] += inp
    by_task[task_id]["output"] += out
    by_task[task_id]["cost"] += cost

if args.json:
    print(json.dumps({
        "period": {"from": from_date, "to": to_date},
        "total": {"input_tokens": total_input, "output_tokens": total_output, "cost": round(total_cost, 2)},
        "by_model": {k: {**v, "cost": round(v["cost"], 2)} for k, v in by_model.items()},
        "by_track": {k: {**v, "cost": round(v["cost"], 2)} for k, v in by_track.items()},
        "by_task": {k: {**v, "cost": round(v["cost"], 2)} for k, v in sorted(by_task.items(), key=lambda x: -x[1]["cost"])[:10]},
    }, indent=2))
    return

# Header
period_str = from_date if from_date == to_date else f"{from_date} to {to_date}"
print(f"\n  COST REPORT — {period_str}")
print(f"  {'=' * 50}")
print(f"\n  Total: {format_tokens(total_input)} input / {format_tokens(total_output)} output (${total_cost:.2f})")
print(f"  Records: {len(records)}")

# By Model
print(f"\n  By Model:")
for model, data in sorted(by_model.items(), key=lambda x: -x[1]["cost"]):
    short_name = model.replace("claude-", "")
    pct = (data["cost"] / total_cost * 100) if total_cost > 0 else 0
    print(f"    {short_name:20s} {format_tokens(data['input']):>6s} in / {format_tokens(data['output']):>6s} out  (${data['cost']:.2f}, {pct:.0f}%)")

# By Track
if len(by_track) > 1 or list(by_track.keys()) != ["unknown"]:
    track_names = {
        "A": "Backend", "B": "Frontend", "C": "DevOps", "D": "Security",
        "E": "Testing", "F": "Docs", "G": "DMS", "H": "Framework",
        "I": "UI Components", "J": "Memory", "K": "Workflow",
        "L": "Ext Testing", "M": "Ext Security", "N": "GTM",
    }
    print(f"\n  By Track:")
    for track, data in sorted(by_track.items(), key=lambda x: -x[1]["cost"]):
        name = track_names.get(track, track)
        pct = (data["cost"] / total_cost * 100) if total_cost > 0 else 0
        print(f"    {track} {name:20s} {format_tokens(data['input'] + data['output']):>6s}  ${data['cost']:.2f}  ({pct:.0f}%)")

# Top Tasks
top_tasks = sorted(by_task.items(), key=lambda x: -x[1]["cost"])[:5]
if top_tasks and top_tasks[0][0] != "unknown":
    print(f"\n  Top Tasks:")
    for task_id, data in top_tasks:
        print(f"    {task_id:12s} {format_tokens(data['input'] + data['output']):>6s}  ${data['cost']:.2f}")

print()

def display_budgets(args): """Display budget utilization.""" budgets = load_budgets() totals = load_running_totals()

if not budgets:
    print("\nNo budgets configured.")
    print("Set budgets with: TokenEconomicsService().set_budget(scope, amount)")
    return

if args.json:
    result = {}
    for scope, budget_data in budgets.items():
        limit = budget_data.get("limit", 0)
        consumed = totals.get(scope, {}).get("cost", 0)
        result[scope] = {
            "limit": limit,
            "consumed": round(consumed, 2),
            "utilization": round(consumed / limit, 4) if limit > 0 else 0,
        }
    print(json.dumps(result, indent=2))
    return

print(f"\n  BUDGET UTILIZATION")
print(f"  {'=' * 50}")
for scope, budget_data in sorted(budgets.items()):
    limit = budget_data.get("limit", 0)
    consumed = totals.get(scope, {}).get("cost", 0)
    utilization = consumed / limit if limit > 0 else 0
    bar_len = 20
    filled = int(utilization * bar_len)
    bar = "█" * min(filled, bar_len) + "░" * max(bar_len - filled, 0)
    status = "✓" if utilization < 0.80 else ("⚠" if utilization < 1.0 else "✗")
    print(f"    {scope:20s} ${consumed:>8.2f} / ${limit:>8.2f} ({utilization:.0%}) [{bar}] {status}")
print()

def display_models(args): """Display per-model pricing and usage.""" if args.json: print(json.dumps(DEFAULT_PRICING, indent=2)) return

print(f"\n  MODEL PRICING")
print(f"  {'=' * 50}")
print(f"    {'Model':25s} {'Input $/M':>10s} {'Output $/M':>12s}")
print(f"    {'-' * 47}")
for model, pricing in sorted(DEFAULT_PRICING.items()):
    short = model.replace("claude-", "")
    print(f"    {short:25s} ${pricing['input']:>8.2f}   ${pricing['output']:>9.2f}")
print()

def display_efficiency(records, args): """Display efficiency metrics.""" if not records: print("\nNo records for efficiency analysis.") return

total_input = sum(r.get("input_tokens", 0) for r in records)
total_output = sum(r.get("output_tokens", 0) for r in records)
total_cost = sum(r.get("cost", 0) for r in records)
num_records = len(records)
unique_tasks = len(set(r.get("task_id", "") for r in records))

output_ratio = total_output / total_input if total_input > 0 else 0
cost_per_task = total_cost / unique_tasks if unique_tasks > 0 else 0
avg_tokens_per_call = (total_input + total_output) / num_records if num_records > 0 else 0

if args.json:
    print(json.dumps({
        "cost_per_task": round(cost_per_task, 2),
        "tokens_per_call": round(avg_tokens_per_call),
        "output_input_ratio": round(output_ratio, 4),
        "unique_tasks": unique_tasks,
        "total_records": num_records,
    }, indent=2))
    return

print(f"\n  EFFICIENCY METRICS")
print(f"  {'=' * 50}")
print(f"    Cost per Task Completed:   ${cost_per_task:.2f}")
print(f"    Tokens per Record:         ~{avg_tokens_per_call:,.0f}")
print(f"    Output/Input Ratio:        {output_ratio:.1%}")
print(f"    Unique Tasks:              {unique_tasks}")
print(f"    Total Records:             {num_records}")
print()

def main(): args = parse_args()

today = date.today().isoformat()
from_date = args.from_date or today
to_date = args.to_date or today

if args.budgets:
    display_budgets(args)
    return

if args.models:
    display_models(args)
    return

records = load_records(from_date, to_date)
records = filter_records(records, track=args.track, task=args.task)

if args.efficiency:
    display_efficiency(records, args)
    return

display_summary(records, from_date, to_date, args)

if name == "main": main()