#!/usr/bin/env python3
"""Dynamic pass over local Claude Code transcripts -> one JSON document.

Usage: offload-scan --plugin <name> [--days N | --since D --until D]
                    [--root ~/.claude/projects] [--json] [--catalog PATH]

Measures what a plugin's skills and agents ACTUALLY cost in real sessions,
and how much of that cost bought no judgment: Mechanical Turn Ratio,
recurring tool-call n-grams, read amplification, retry density, judgment
density, fan-out multiplier, and the composite offload_value (design/spec.md
§5.2). Single streaming stdlib process, offline.

PRIVACY (FR-3.5): only normalized shapes and counts leave this process. Tool
arguments are masked to <path>/<n>/<sha>/<url>/<str> before they ever enter a
data structure that is emitted; Bash commands are reduced to their bare
executable/subcommand words. No raw transcript content, file content, command
argument, or user prose is ever printed.

MONEY MATH (FR-3.2): the pricing table, tier weighting, and cache multipliers
come from token-budget's bin/cc-tokens and are never reimplemented here. See
resolve_cc_tokens() for the resolution order and the note there about why this
imports the module rather than shelling out.
"""
import argparse
import glob
import importlib.machinery
import importlib.util
import json
import math
import os
import re
import sys
from collections import Counter, defaultdict
from datetime import datetime, timedelta, timezone

PLUGIN_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
DEFAULT_CATALOG = os.path.join(PLUGIN_ROOT, "references", "signals-catalog.md")
DEFAULT_ROOT = os.path.expanduser("~/.claude/projects")
CC_TOKENS_GLOB = os.path.expanduser(
    "~/.claude/plugins/cache/*/token-budget/*/bin/cc-tokens")


# --------------------------------------------------------------------------
# cc-tokens dependency (plan.md §4.3)
# --------------------------------------------------------------------------

def semver_key(text):
    m = re.match(r"^(\d+)\.(\d+)\.(\d+)", text)
    return tuple(int(g) for g in m.groups()) if m else (0, 0, 0)


def resolve_cc_tokens():
    """Locate cc-tokens: PATH first, then the plugin cache, else fail loudly.

    Imported as a module rather than shelled out with --json, because
    cc-tokens' subcommands aggregate by day/session/project/model and none of
    them can answer "weighted tokens for THIS set of (message.id, requestId)
    turns" — the attribution fields this pass filters on are not part of its
    CLI surface. Importing keeps the pricing table, the cache multipliers and
    the tier weighting in exactly one home (its Rec.cost / Rec.weighted are
    what compute every number here), which is the property plan.md §4.3
    actually cares about. The clean long-term fix is still
    `cc-tokens attribute --by skill|plugin --json` upstream.
    """
    path = None
    for d in os.environ.get("PATH", "").split(os.pathsep):
        cand = os.path.join(d, "cc-tokens")
        if os.path.isfile(cand) and os.access(cand, os.X_OK):
            path = cand
            break
    if path is None:
        cached = sorted(glob.glob(CC_TOKENS_GLOB),
                        key=lambda p: semver_key(os.path.basename(
                            os.path.dirname(os.path.dirname(p)))))
        if cached:
            path = cached[-1]
    if path is None:
        sys.exit(
            "error: cc-tokens not found on PATH and not in "
            "~/.claude/plugins/cache/*/token-budget/*/bin/cc-tokens.\n"
            "  offload-scan takes all token and cost arithmetic from "
            "token-budget's cc-tokens and never vendors a copy.\n"
            "  install: claude plugin install token-budget@oleks-local  "
            "(or add token-budget/bin to PATH)")

    spec = importlib.util.spec_from_loader(
        "cc_tokens", importlib.machinery.SourceFileLoader("cc_tokens", path))
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    for attr in ("Rec", "price_for", "parse_day"):
        if not hasattr(module, attr):
            sys.exit(f"error: {path} has no {attr!r} — incompatible cc-tokens; "
                     f"update token-budget or report this mismatch")
    return module, path


# --------------------------------------------------------------------------
# signals-catalog.md loader (same fenced-block contract as plugin-inventory,
# but tolerant of glob patterns like list_* in the tool sections)
# --------------------------------------------------------------------------

SCAN_SECTIONS = {"mechanical-tool-calls", "read-tool-calls", "read-only-bash",
                 "scan-thresholds"}


def load_scan_catalog(path):
    with open(path, encoding="utf-8") as f:
        text = f.read()
    sections = {}
    for m in re.finditer(r"^##\s+([a-z-]+)\s*$", text, re.MULTILINE):
        name = m.group(1)
        if name not in SCAN_SECTIONS:
            continue
        block = re.search(r"```\S*\n(.*?)```", text[m.end():], re.DOTALL)
        if not block:
            continue
        body = block.group(1)
        if name == "scan-thresholds":
            values = {}
            for raw in body.split("\n"):
                line = raw.split("#", 1)[0].strip()
                if ":" in line:
                    k, v = line.split(":", 1)
                    values[k.strip()] = float(v.strip())
            sections[name] = values
        else:
            items = set()
            for raw in body.split("\n"):
                for item in raw.split(","):
                    item = item.strip()
                    if re.match(r"^[A-Za-z_*][A-Za-z0-9_ *-]*$", item):
                        items.add(item)
            sections[name] = items
    missing = SCAN_SECTIONS - sections.keys()
    if missing:
        sys.exit(f"error: {path} is missing section(s): {', '.join(sorted(missing))}")
    return sections


def matches_any(name, patterns):
    for pat in patterns:
        if pat == name:
            return True
        if "*" in pat and re.fullmatch(re.escape(pat).replace(r"\*", ".*"), name):
            return True
    return False


# --------------------------------------------------------------------------
# masking (FR-3.5) — nothing that has not been through here may be emitted
# --------------------------------------------------------------------------

HEX_RE = re.compile(r"^[0-9a-fA-F]{7,}$")
URL_RE = re.compile(r"^[a-z][a-z0-9+.-]*://", re.IGNORECASE)
PATH_RE = re.compile(r"^~?/|^\.{1,2}/|^[\w.-]+/[\w./-]+$")
NUM_RE = re.compile(r"^-?\d+(\.\d+)?$")
BARE_WORD_RE = re.compile(r"^[a-z][a-z0-9_-]*$")


def mask_value(value):
    if isinstance(value, bool):
        return "<bool>"
    if isinstance(value, (int, float)):
        return "<n>"
    if isinstance(value, dict):
        return "<obj>"
    if isinstance(value, list):
        return "<list>"
    if value is None:
        return "<null>"
    text = str(value).strip()
    if not text:
        return "<empty>"
    if URL_RE.match(text):
        return "<url>"
    if NUM_RE.match(text):
        return "<n>"
    if HEX_RE.match(text):
        return "<sha>"
    if PATH_RE.match(text) and " " not in text:
        return "<path>"
    return "<str>"


def mask_bash(command):
    """Reduce a shell command to its executable/subcommand words only.

    Keeps bare lowercase words (`git`, `status`, `kubectl`, `get`) and masks
    everything else, so `cat /etc/secrets/token` becomes `cat <path>` and
    `curl https://…` becomes `curl <url>`. Argument literals never survive.
    """
    tokens = str(command).strip().split()
    kept = []
    for tok in tokens[:3]:
        if BARE_WORD_RE.match(tok):
            kept.append(tok)
        else:
            kept.append(mask_value(tok))
            break
    return " ".join(kept) if kept else "<empty>"


def short_tool_name(name):
    """mcp__plugin_cluster_gitea-tools__issue_read -> issue_read."""
    if name.startswith("mcp__") and "__" in name[5:]:
        return name.rsplit("__", 1)[-1]
    return name


def tool_signature(block):
    name = short_tool_name(block.get("name") or "?")
    args = block.get("input")
    if not isinstance(args, dict) or not args:
        return f"{name}()"
    parts = []
    for key in sorted(args):
        val = args[key]
        if name in ("Bash", "BashOutput") and key in ("command", "cmd"):
            parts.append(f"{key}={mask_bash(val)}")
        else:
            parts.append(f"{key}={mask_value(val)}")
    return f"{name}({', '.join(parts)})"


# --------------------------------------------------------------------------
# turn model
# --------------------------------------------------------------------------

class Turn:
    """One de-duplicated attributed assistant turn, plus its classification
    inputs. `rec` is a cc-tokens Rec — all token/cost arithmetic lives there."""

    __slots__ = ("rec", "session", "ts", "skill", "agent", "plugin", "sidechain",
                 "sigs", "tool_names", "text_words", "output_tokens",
                 "after_error", "asks_user", "pulled_tokens", "referenced_tokens")

    def __init__(self, rec):
        self.rec = rec
        self.session = rec.session
        self.ts = rec.ts
        self.skill = None
        self.agent = None
        self.plugin = None
        self.sidechain = bool(rec.sidechain)
        self.sigs = []
        self.tool_names = []
        self.text_words = 0
        self.output_tokens = rec.out
        self.after_error = False
        self.asks_user = False
        self.pulled_tokens = 0
        self.referenced_tokens = 0

    @property
    def group(self):
        return self.agent or self.skill or f"{self.plugin}:(unattributed)"

    @property
    def kind(self):
        return "agent" if self.agent else "skill"


def est_tokens(text):
    """~4 chars per token. Only ever applied to material that stays local."""
    return max(1, len(text) // 4)


def content_text(blocks):
    out = []
    if isinstance(blocks, str):
        return blocks
    if not isinstance(blocks, list):
        return ""
    for b in blocks:
        if isinstance(b, dict) and b.get("type") == "text":
            out.append(str(b.get("text") or ""))
    return "\n".join(out)


def result_text(block):
    content = block.get("content")
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        parts = []
        for c in content:
            if isinstance(c, dict) and c.get("type") == "text":
                parts.append(str(c.get("text") or ""))
        return "\n".join(parts)
    return ""


WORD_RE = re.compile(r"[A-Za-z0-9_./-]{3,}")


def word_set(text, cap=4000):
    return set(WORD_RE.findall(text)[:cap])


# --------------------------------------------------------------------------
# streaming scan
# --------------------------------------------------------------------------

def normalize_skill(value, plugin):
    if not value:
        return None
    return value if ":" in value else f"{plugin}:{value}"


def scan(root, plugin, since, until, cc, catalog):
    """Streams every transcript once. Yields nothing; returns
    (turns, session_totals, sessions_with_target).

    Memory is O(turns attributed to the target + sessions), not O(all turns):
    unattributed turns are counted and dropped immediately.
    """
    read_tools = catalog["read-tool-calls"]
    turns = []
    session_turns = Counter()
    session_hit = set()
    session_span = {}

    floor = since.timestamp() - 86400 if since else None
    seen = set()

    for path in sorted(glob.iglob(os.path.join(root, "**", "*.jsonl"),
                                  recursive=True)):
        try:
            if floor is not None and os.stat(path).st_mtime < floor:
                continue
        except OSError:
            continue
        try:
            fh = open(path, "r", errors="replace")
        except OSError:
            continue

        pending = {}
        order = []
        # Rolling state, valid only within one file (== one session's append
        # log, so file order is chronological).
        last_error = False
        pending_reads = []
        tool_use_names = {}

        with fh:
            for line in fh:
                if '"type"' not in line:
                    continue
                try:
                    o = json.loads(line)
                except (ValueError, TypeError):
                    continue
                kind = o.get("type")

                if kind == "user":
                    blocks = (o.get("message") or {}).get("content")
                    if not isinstance(blocks, list):
                        continue
                    for b in blocks:
                        if not isinstance(b, dict) or b.get("type") != "tool_result":
                            continue
                        if b.get("is_error"):
                            last_error = True
                        name = tool_use_names.pop(b.get("tool_use_id"), None)
                        if not name or not matches_any(name, read_tools):
                            continue
                        text = result_text(b)
                        if text and len(pending_reads) < 32:
                            pending_reads.append((est_tokens(text), word_set(text)))
                    continue

                if kind != "assistant":
                    continue

                msg = o.get("message") or {}
                usage = msg.get("usage") or {}
                if not usage or msg.get("model") == "<synthetic>":
                    continue
                try:
                    when = datetime.fromisoformat(
                        str(o.get("timestamp")).replace("Z", "+00:00"))
                except (ValueError, TypeError):
                    continue
                if since and when < since:
                    continue
                if until and when > until:
                    continue

                session = o.get("sessionId") or os.path.basename(path)[:-6]
                session_turns[session] += 1
                span = session_span.get(session)
                if span is None:
                    session_span[session] = [when, when]
                else:
                    if when < span[0]:
                        span[0] = when
                    if when > span[1]:
                        span[1] = when

                # Registered for every assistant turn, attributed or not: the
                # read is issued by turn N and consumed by turn N+1, and only
                # one of the two need belong to the target.
                if len(tool_use_names) > 256:
                    tool_use_names.clear()
                for b in msg.get("content") or []:
                    if isinstance(b, dict) and b.get("type") == "tool_use":
                        tool_use_names[b.get("id")] = short_tool_name(b.get("name") or "?")

                attr_plugin = o.get("attributionPlugin")
                attr_agent = o.get("attributionAgent")
                attr_skill = o.get("attributionSkill")
                mine = (attr_plugin == plugin
                        or (attr_agent or "").startswith(plugin + ":")
                        or (attr_skill or "").startswith(plugin + ":"))
                if not mine:
                    # Still consumed: a tool error or a read that precedes an
                    # unattributed turn belongs to that turn, not to ours.
                    last_error = False
                    pending_reads = []
                    continue
                session_hit.add(session)

                cw = usage.get("cache_creation") or {}
                cw5 = int(cw.get("ephemeral_5m_input_tokens") or 0)
                cw1 = int(cw.get("ephemeral_1h_input_tokens") or 0)
                total_cw = int(usage.get("cache_creation_input_tokens") or 0)
                if not (cw5 or cw1) and total_cw:
                    cw1 = total_cw

                rec = cc.Rec(
                    ts=when, model=msg.get("model"), session=session,
                    cwd=o.get("cwd") or "", sidechain=bool(o.get("isSidechain")),
                    inp=int(usage.get("input_tokens") or 0), cw5=cw5, cw1=cw1,
                    cr=int(usage.get("cache_read_input_tokens") or 0),
                    out=int(usage.get("output_tokens") or 0))

                key = (msg.get("id"), o.get("requestId"))
                if key == (None, None):
                    key = ("_anon", o.get("uuid") or len(order))

                turn = pending.get(key)
                if turn is None:
                    turn = Turn(rec)
                    turn.plugin = attr_plugin or plugin
                    turn.skill = normalize_skill(attr_skill, turn.plugin)
                    turn.agent = attr_agent
                    turn.after_error = last_error
                    pending[key] = turn
                    order.append(key)
                else:
                    # Same message, later streaming snapshot: max each field.
                    # (The cache fields never vary; output_tokens grows.)
                    for f in ("inp", "cw5", "cw1", "cr", "out"):
                        setattr(turn.rec, f,
                                max(getattr(turn.rec, f), getattr(rec, f)))
                    turn.output_tokens = turn.rec.out

                for b in msg.get("content") or []:
                    if not isinstance(b, dict):
                        continue
                    if b.get("type") == "tool_use":
                        name = short_tool_name(b.get("name") or "?")
                        turn.tool_names.append(name)
                        turn.sigs.append(tool_signature(b))
                        if name == "AskUserQuestion":
                            turn.asks_user = True
                    elif b.get("type") == "text":
                        turn.text_words += len(str(b.get("text") or "").split())

                if pending_reads:
                    words = word_set(content_text(msg.get("content")))
                    for pulled, result_words in pending_reads:
                        turn.pulled_tokens += pulled
                        overlap = words & result_words
                        turn.referenced_tokens += est_tokens(" ".join(overlap))
                    pending_reads = []
                last_error = False

        for key in order:
            if key in seen:
                continue
            seen.add(key)
            turns.append(pending[key])

    return turns, session_turns, session_hit, session_span


# --------------------------------------------------------------------------
# classification (spec §5.2.1, §5.2.4, §5.2.5)
# --------------------------------------------------------------------------

def is_mechanical_tool(name, sig, catalog):
    if name in ("Bash", "BashOutput"):
        head = sig.split("cmd=", 1)[-1].split("command=", 1)[-1].rstrip(")")
        words = [w for w in head.split() if BARE_WORD_RE.match(w)]
        if not words:
            return False
        return (matches_any(words[0], catalog["read-only-bash"])
                or matches_any(" ".join(words[:2]), catalog["read-only-bash"]))
    return matches_any(name, catalog["mechanical-tool-calls"])


def classify(turn, catalog):
    th = catalog["scan-thresholds"]
    if turn.after_error:
        return "retry"
    if turn.tool_names:
        if (turn.output_tokens < th["mechanical_output_tokens_max"]
                and all(is_mechanical_tool(n, s, catalog)
                        for n, s in zip(turn.tool_names, turn.sigs))):
            return "mechanical"
        return "other"
    if turn.output_tokens >= th["judgment_output_tokens_min"] or turn.asks_user:
        return "judgment"
    return "other"


# --------------------------------------------------------------------------
# invocation reconstruction (plan.md §4.1 step 4)
# --------------------------------------------------------------------------

def reconstruct_invocations(turns, idle_gap_minutes):
    """Contiguous runs of attributed turns in one session and one group,
    split on an idle gap. Returns [(invocation_id, group, [turns])]."""
    by_session = defaultdict(list)
    for t in turns:
        by_session[t.session].append(t)

    invocations = []
    gap = timedelta(minutes=idle_gap_minutes)
    for session, group_turns in sorted(by_session.items()):
        group_turns.sort(key=lambda t: t.ts)
        current = []
        for t in group_turns:
            if current and (t.group != current[-1].group
                            or t.ts - current[-1].ts > gap):
                invocations.append((f"{session}#{len(invocations)}",
                                    current[0].group, current))
                current = []
            current.append(t)
        if current:
            invocations.append((f"{session}#{len(invocations)}",
                                current[0].group, current))
    return invocations


# --------------------------------------------------------------------------
# n-gram mining (spec §5.2.2)
# --------------------------------------------------------------------------

def is_subsequence(short, long_):
    n = len(short)
    return any(long_[i:i + n] == short for i in range(len(long_) - n + 1))


def mine_ngrams(invocations, weighted_of, min_len, max_len, min_invocations,
                top=5):
    """Ordered signature sequences per invocation; n-grams of length >= min_len
    recurring in >= min_invocations distinct invocations. Only maximal n-grams
    are reported — a 3-gram fully inside a kept 5-gram with no extra
    recurrences is the same finding stated twice."""
    seqs = []
    for inv_id, _group, turns in invocations:
        seq = []
        for t in turns:
            for sig in t.sigs:
                seq.append((sig, t))
        if len(seq) >= min_len:
            seqs.append((inv_id, seq))

    found = {}
    for n in range(min_len, max_len + 1):
        counts = defaultdict(lambda: [0, set(), set()])
        for inv_id, seq in seqs:
            for i in range(len(seq) - n + 1):
                gram = tuple(s for s, _ in seq[i:i + n])
                entry = counts[gram]
                entry[0] += 1
                entry[1].add(inv_id)
                for _, t in seq[i:i + n]:
                    entry[2].add(id(t))
        for gram, (occurrences, invs, turn_ids) in counts.items():
            if len(invs) >= min_invocations:
                found[gram] = (occurrences, len(invs), turn_ids)

    kept = []
    for gram in sorted(found, key=lambda g: (-len(g), -found[g][0])):
        occurrences, invs, turn_ids = found[gram]
        if any(len(gram) < len(k) and is_subsequence(list(gram), list(k))
               and found[k][1] >= invs for k in (kk for kk, _ in kept)):
            continue
        kept.append((gram, (occurrences, invs, turn_ids)))

    turn_weight = {id(t): weighted_of(t)
                   for _i, _g, ts in invocations for t in ts}
    out = []
    for gram, (occurrences, invs, turn_ids) in kept:
        waste = int(sum(turn_weight.get(tid, 0) for tid in turn_ids))
        out.append({"sig": list(gram), "recurrences": occurrences,
                    "invocations": invs, "waste": waste})
    out.sort(key=lambda g: (-g["waste"], -g["recurrences"]))
    return out[:top]


# --------------------------------------------------------------------------
# aggregation (spec §5.2.7)
# --------------------------------------------------------------------------

def summarize_group(name, kind, turns, invocations, plugin_weighted,
                    inline_mean_ctx, catalog):
    """fanout_multiplier is mean context per SIDECHAIN turn over mean context
    per inline turn of the same plugin — literally "a step in here costs N x
    the same step inline". It is 1.0 when the group has no sidechain turns,
    and it can legitimately come out BELOW 1: a subagent spawned with a tight
    brief carries less context than a marathon main session, which is the
    fan-out done right. Reported unclamped; it is a diagnostic, not an input
    to offload_value."""
    th = catalog["scan-thresholds"]
    n = len(turns)
    weighted = sum(t.rec.weighted for t in turns)
    kinds = Counter(classify(t, catalog) for t in turns)
    mechanical = [t for t in turns if classify(t, catalog) == "mechanical"]
    offload_waste = sum(t.rec.weighted for t in mechanical)

    pulled = sum(t.pulled_tokens for t in turns)
    referenced = sum(t.referenced_tokens for t in turns)
    read_amp = round(pulled / referenced, 2) if referenced else (
        float(pulled) if pulled else 0.0)

    sidechain = [t for t in turns if t.sidechain]
    if sidechain and inline_mean_ctx:
        mean_ctx = sum(t.rec.ctx for t in sidechain) / len(sidechain)
        fanout = round(mean_ctx / inline_mean_ctx, 2)
    else:
        fanout = 1.0

    ngrams = mine_ngrams(
        invocations, lambda t: t.rec.weighted,
        int(th["ngram_min_len"]), int(th["ngram_max_len"]),
        int(th["ngram_min_invocations"]))

    judgment_density = round(kinds["judgment"] / n, 3) if n else 0.0
    dominant = max((g["recurrences"] for g in ngrams), default=0)
    repetition_factor = 1 + math.log2(dominant) if dominant > 1 else 1.0
    offload_value = int(offload_waste * (1 - judgment_density) * repetition_factor)

    return {
        ("agent" if kind == "agent" else "skill"): name,
        "invocations": len(invocations),
        "turns": n,
        "sidechain_turns": len(sidechain),
        "weighted_tokens": int(weighted),
        "cost_estimate_usd": round(sum(t.rec.cost for t in turns), 2),
        "share_of_plugin": round(weighted / plugin_weighted, 3) if plugin_weighted else 0.0,
        "mtr": round(len(mechanical) / n, 3) if n else 0.0,
        "judgment_density": judgment_density,
        "retry_density": round(kinds["retry"] / n, 3) if n else 0.0,
        "offload_waste": int(offload_waste),
        "read_amplification": read_amp,
        "fanout_multiplier": fanout,
        "repetition_factor": round(repetition_factor, 3),
        "ngrams": ngrams,
        "offload_value": offload_value,
    }


# --------------------------------------------------------------------------
# main
# --------------------------------------------------------------------------

def main():
    ap = argparse.ArgumentParser(prog="offload-scan")
    ap.add_argument("--plugin", required=True, help="plugin name to attribute against")
    ap.add_argument("--days", type=int, help="window size in days back from now (default 30)")
    ap.add_argument("--since", help="inclusive lower bound YYYY-MM-DD")
    ap.add_argument("--until", help="inclusive upper bound YYYY-MM-DD")
    ap.add_argument("--root", default=DEFAULT_ROOT, help="transcripts root")
    ap.add_argument("--catalog", default=DEFAULT_CATALOG, help="path to signals-catalog.md")
    ap.add_argument("--json", action="store_true", help="compact single-line JSON")
    args = ap.parse_args()

    cc, cc_path = resolve_cc_tokens()
    catalog = load_scan_catalog(args.catalog)

    root = os.path.expanduser(args.root)
    if not os.path.isdir(root):
        sys.exit(f"error: no such directory: {root}")

    if args.since or args.until:
        since = cc.parse_day(args.since) if args.since else None
        until = cc.parse_day(args.until) if args.until else None
        if until:
            until += timedelta(days=1) - timedelta(microseconds=1)
    else:
        days = args.days if args.days else 30
        since = datetime.now(timezone.utc) - timedelta(days=days)
        until = None

    turns, session_turns, session_hit, session_span = scan(
        root, args.plugin, since, until, cc, catalog)

    th = catalog["scan-thresholds"]
    invocations = reconstruct_invocations(turns, th["idle_gap_minutes"])
    plugin_weighted = sum(t.rec.weighted for t in turns)

    inline = [t for t in turns if not t.sidechain]
    inline_mean_ctx = (sum(t.rec.ctx for t in inline) / len(inline)) if inline else 0.0

    by_group = defaultdict(list)
    for t in turns:
        by_group[(t.group, t.kind)].append(t)
    invs_by_group = defaultdict(list)
    for inv in invocations:
        invs_by_group[inv[1]].append(inv)

    skills, agents = [], []
    for (name, kind), group_turns in by_group.items():
        row = summarize_group(name, kind, group_turns, invs_by_group[name],
                              plugin_weighted, inline_mean_ctx, catalog)
        (agents if kind == "agent" else skills).append(row)
    skills.sort(key=lambda r: -r["offload_value"])
    agents.sort(key=lambda r: -r["offload_value"])

    candidate_turns = sum(c for s, c in session_turns.items() if s in session_hit)
    spans = [session_span[s] for s in session_hit if s in session_span]
    window_since = min((s[0] for s in spans), default=since)
    window_until = max((s[1] for s in spans), default=until)

    doc = {
        "window": {
            "since": window_since.isoformat() if window_since else None,
            "until": window_until.isoformat() if window_until else None,
            "requested_since": since.isoformat() if since else None,
            "requested_until": until.isoformat() if until else None,
            "sessions": len(session_hit),
        },
        "plugin": args.plugin,
        "tool_versions": {"cc_tokens_path": cc_path},
        "coverage": {
            "attributed_turns": len(turns),
            "candidate_turns": candidate_turns,
            "ratio": round(len(turns) / candidate_turns, 3) if candidate_turns else 0.0,
            "caveat": "hook-driven and unattributed agent turns excluded",
        },
        "totals": {
            "weighted_tokens": int(plugin_weighted),
            "cost_estimate_usd": round(sum(t.rec.cost for t in turns), 2),
            "invocations": len(invocations),
            "mechanical_share": round(
                sum(1 for t in turns if classify(t, catalog) == "mechanical")
                / len(turns), 3) if turns else 0.0,
            # reconstruct_invocations() only sees already-attributed turns and
            # splits on a wall-clock idle gap; it has no signal for unrelated
            # activity between two genuinely separate same-skill calls, so
            # back-to-back invocations closer together than the gap collapse
            # into one reported invocation. Counts here are a LOWER BOUND on
            # the true invocation count, and per-invocation figures derived
            # from them (e.g. cost/invocation) are correspondingly inflated.
            "invocation_caveat": (
                "invocation counts are a lower bound: two same-skill calls "
                f"within {int(th['idle_gap_minutes'])} minutes of each other on the wall "
                "clock are merged into one reported invocation"),
        },
        "skills": skills,
        "agents": agents,
    }

    print(json.dumps(doc) if args.json else json.dumps(doc, indent=2))


if __name__ == "__main__":
    main()
