#!/usr/bin/env python3
# Copyright (c) 2026 Craig M. Brown. All rights reserved.
"""workforce_extract.py — local extractor for the BlindOracle Agent Workforce report.

Run this on YOUR machine. It reads your Claude Code transcripts (and, if you ask, your
git log, a decision export and your .claude/ instruction counts) and writes a small
JSON summary of per-session COUNTS. You then send that summary, not your transcripts,
to `ops.agent-workforce-report` ($25) or `ops.agent-workforce-diagnostic` ($0.25):

    python3 workforce_extract.py --transcripts ~/.claude/projects/<project> --days 30 \\
        [--git-repo PATH] [--decisions FILE.jsonl] [--instructions-dir .claude] \\
        --out summary.json
    python3 workforce_extract.py --self-check summary.json     # what is in it, and its size

What the summary contains
  * per session: a salted hash key, interactive|autonomous, first/last timestamp,
    human turns, agent-active seconds, wait seconds (<= 30 min), the count and seconds
    of waits over 30 min, assistant word count, meta/direct edit counts and skill
    invocation counts by skill name (`--hash-skill-names` hashes those too);
  * optional `git_log`: per commit a salted hash of the sha, timestamp, agent|human
    class (Claude co-author trailer), is_fix, is_revert, salted hashes of the file
    paths, insertions and deletions;
  * optional `decisions`: created_ts, resolved_ts, status only;
  * optional `instructions`: counts of rules, skills, commands, memory files and
    CLAUDE.md lines.

What it never contains: message text, prompts, tool inputs or outputs, file paths in
clear, session ids, user names, commit subjects or shas in clear. The salt is random
per run and is never written, so the hashes cannot be joined to anything outside the
file. This script makes no network calls and imports nothing outside the Python
standard library; read it before you run it — it is one file.

The turn walk is the same as the report's own (`workforce_report._scan_session`,
method fleet_workforce_meter/2): human prompt -> agent entries until the next prompt
(active) -> gap to the next prompt (wait); gaps over 30 minutes are excluded. A
session counts if its first timestamp falls inside the window.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import re
import secrets
import subprocess
import sys
import time
from array import array
from collections import Counter
from datetime import datetime, timedelta, timezone
from pathlib import Path

EXTRACTOR_VERSION = "1.0.0"
SCHEMA = "bo.workforce_summary/1"
METHOD = "fleet_workforce_meter/2 turn walk (workforce_report._scan_session)"

SESSION_FIELDS = [
    "session_key", "entrypoint", "first_ts", "last_ts", "human_turns", "active_s", "wait_s",
    "waits_over_30min", "waits_over_30min_s", "assistant_words", "meta_edits", "direct_edits",
    "skills",
]
COMMIT_FIELDS = ["sha", "ts", "author", "is_fix", "is_revert", "files", "insertions", "deletions"]
HASH_HEX = 16  # 64-bit salted hashes: collision-free at transcript/commit scale
MAX_SKILL_NAME = 120        # longer skill ids are sent hashed (the report accepts <= 128 chars)
MAX_FILES_PER_COMMIT = 20_000  # the report's cap; larger commits keep the first 20,000 paths

# --- mirrored from workforce_report.py / scripts/fleet_workforce_meter.py -------------
WAIT_EXCLUDE_S = 30 * 60
META_PATH_RE = re.compile(
    r"(^|/)(\.claude|configs?|rules?|skills?|hooks?|prompts?|critics?|rubrics)(/|$)|prompt|critic",
    re.I,
)
FIX_SUBJECT_RE = re.compile(r"^(revert|fix|hotfix|bugfix)\b|\bfix\(", re.I)
REVERT_SUBJECT_RE = re.compile(r"^revert\b", re.I)
_LOCAL_CMD_RE = re.compile(r"<command-name>/?(clear|compact|exit|quit|help|config|status|cost)</command-name>", re.I)
_EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc)
_US = timedelta(microseconds=1)


def _parse_ts(s):
    if not s or not isinstance(s, str):
        return None
    try:
        s2 = s.replace("Z", "+00:00")
        d = datetime.fromisoformat(s2)
        if d.tzinfo is None:
            d = d.replace(tzinfo=timezone.utc)
        return d
    except Exception:
        return None


def _words(text) -> int:
    if not isinstance(text, str) or not text:
        return 0
    return len(text.split())


def _is_prompt_text(text: str) -> bool:
    t = (text or "").strip()
    if not t:
        return False
    if t.startswith("<local-command-stdout>") or t.startswith("<local-command-caveat>"):
        return False
    if _LOCAL_CMD_RE.search(t) and re.search(r"<command-args>\s*</command-args>", t):
        return False
    return True


def _is_human_turn(rec: dict) -> bool:
    if rec.get("type") != "user" or rec.get("isSidechain"):
        return False
    if rec.get("promptSource") == "system" or rec.get("isMeta"):
        return False
    msg = rec.get("message") or {}
    content = msg.get("content")
    if isinstance(content, str):
        return _is_prompt_text(content)
    if isinstance(content, list):
        has_text = False
        for b in content:
            if not isinstance(b, dict):
                continue
            if b.get("type") == "tool_result":
                return False
            if b.get("type") == "text" and _is_prompt_text(str(b.get("text", ""))):
                has_text = True
        return has_text
    return False


def _is_agent_trailer_text(text: str) -> bool:
    t = (text or "").lower()
    return "co-authored-by: claude" in t or "noreply@anthropic.com" in t


# --------------------------------------------------------------------------- helpers
def _us(d: datetime) -> int:
    """Exact integer microseconds since the epoch (UTC instant)."""
    return (d - _EPOCH) // _US


def _iso_us(us: int) -> str:
    d = _EPOCH + timedelta(microseconds=us)
    spec = "milliseconds" if us % 1000 == 0 else "microseconds"
    return d.isoformat(timespec=spec).replace("+00:00", "Z")


class Hasher:
    """Salted, truncated sha256. The salt is random per run and never written out."""

    def __init__(self, salt: bytes | None = None):
        self._salt = salt if salt is not None else secrets.token_bytes(32)

    def __call__(self, value: str) -> str:
        return hashlib.sha256(self._salt + str(value).encode("utf-8", "replace")).hexdigest()[:HASH_HEX]


class _Session:
    """Streaming equivalent of workforce_report._scan_session for one sessionId: keeps
    (timestamp, kind) per event and the counters, never the record."""

    __slots__ = ("ts", "kind", "entry", "human_turns", "assistant_words", "skills",
                 "meta_edits", "direct_edits", "aborted")

    def __init__(self):
        self.ts = array("q")
        self.kind = bytearray()  # 1 = human, 0 = agent
        self.entry = Counter()
        self.human_turns = 0
        self.assistant_words = 0
        self.skills: dict = {}
        self.meta_edits = 0
        self.direct_edits = 0
        self.aborted = False

    def add(self, rec: dict) -> None:
        """Same statement order as _scan_session's loop body; an exception aborts the
        session exactly as `except Exception: parse_errors += 1; return out` does."""
        if self.aborted:
            return
        try:
            ts = _parse_ts(rec.get("timestamp"))
            if ts is None:
                return
            ep = rec.get("entrypoint")
            if ep:
                self.entry[ep] += 1
            if _is_human_turn(rec):
                self.ts.append(_us(ts))
                self.kind.append(1)
                self.human_turns += 1
                return
            self.ts.append(_us(ts))
            self.kind.append(0)
            if rec.get("type") == "assistant":
                msg = rec.get("message") or {}
                content = msg.get("content") if isinstance(msg, dict) else None
                for b in content if isinstance(content, list) else []:
                    if not isinstance(b, dict):
                        continue
                    bt = b.get("type")
                    if bt == "text":
                        self.assistant_words += _words(b.get("text"))
                    elif bt == "tool_use":
                        name = b.get("name")
                        inp = b.get("input") or {}
                        if not isinstance(inp, dict):
                            inp = {}
                        if name == "Skill":
                            sk = inp.get("skill")
                            if sk:
                                sk = str(sk)
                                self.skills[sk] = self.skills.get(sk, 0) + 1
                        elif name in ("Edit", "Write", "MultiEdit", "NotebookEdit"):
                            fp = str(inp.get("file_path") or inp.get("notebook_path") or "")
                            if fp:
                                if META_PATH_RE.search(fp):
                                    self.meta_edits += 1
                                else:
                                    self.direct_edits += 1
        except Exception:
            self.aborted = True

    def finish(self) -> dict | None:
        """Per-session aggregates, or None when there is nothing to report (aborted or no
        timestamped events) — those sessions are counted, never emitted."""
        n = len(self.ts)
        if self.aborted or not n:
            return None
        order = sorted(range(n), key=self.ts.__getitem__)  # stable, like events.sort(key=ts)
        ts = [self.ts[i] for i in order]
        kind = [self.kind[i] for i in order]
        entrypoint = self.entry.most_common(1)[0][0] if self.entry else None
        out = {"first_us": ts[0], "last_us": ts[-1], "interactive": entrypoint == "cli",
               "human_turns": self.human_turns, "active_s": 0.0, "wait_s": 0.0,
               "wait_excluded_s": 0.0, "wait_excluded_n": 0}
        i = 0
        if entrypoint == "cli" and self.human_turns:
            while i < n:
                if kind[i] != 1:
                    i += 1
                    continue
                t = ts[i]
                j = i + 1
                last_agent = t
                while j < n and kind[j] != 1:
                    last_agent = ts[j]
                    j += 1
                out["active_s"] += max(0.0, (last_agent - t) / 10**6)
                if j < n:
                    gap = (ts[j] - last_agent) / 10**6
                    if gap > WAIT_EXCLUDE_S:
                        out["wait_excluded_s"] += gap
                        out["wait_excluded_n"] += 1
                    elif gap > 0:
                        out["wait_s"] += gap
                i = j
        else:
            prev = ts[0]
            for t in ts[1:]:
                gap = (t - prev) / 10**6
                if gap > WAIT_EXCLUDE_S:
                    out["wait_excluded_s"] += gap
                    out["wait_excluded_n"] += 1
                elif gap > 0:
                    out["active_s"] += gap
                prev = t
        return out


def _is_turn_line(line: str) -> bool:
    # Cheap prefilter (the parent meter's): a record whose top-level type is user or
    # assistant always contains one of these substrings; json.loads decides.
    return ('"type":"user"' in line or '"type": "user"' in line
            or '"type":"assistant"' in line or '"type": "assistant"' in line)


class Extraction:
    """Accumulates records into sessions keyed by sessionId (the report's grouping)."""

    def __init__(self, hasher: Hasher, hash_skill_names: bool = False):
        self.hasher = hasher
        self.hash_skill_names = hash_skill_names
        self.sessions: dict = {}
        self.counts = Counter()

    def add_record(self, rec) -> None:
        if not isinstance(rec, dict):
            self.counts["records_not_objects"] += 1
            return
        if rec.get("type") not in ("user", "assistant"):
            self.counts["records_skipped_non_turn"] += 1
            return
        sid = rec.get("sessionId") or rec.get("session_id") or "_"
        s = self.sessions.get(str(sid))
        if s is None:
            s = self.sessions[str(sid)] = _Session()
        s.add(rec)
        self.counts["records_scored"] += 1

    def add_file(self, path: Path) -> None:
        with open(path, "r", errors="ignore") as fh:
            for line in fh:
                if not _is_turn_line(line):
                    continue
                try:
                    rec = json.loads(line)
                except Exception:
                    self.counts["line_parse_errors"] += 1
                    continue
                self.add_record(rec)

    def rows(self, start_us: int | None, end_us: int | None) -> list:
        rows = []
        for sid, s in self.sessions.items():
            agg = s.finish()
            if agg is None:
                self.counts["sessions_without_timestamps"] += 1
                if s.aborted:
                    self.counts["session_scan_errors"] += 1
                continue
            if (start_us is not None and agg["first_us"] < start_us) or \
                    (end_us is not None and agg["first_us"] > end_us):
                self.counts["sessions_outside_window"] += 1
                continue
            skills: dict = {}
            for k, v in s.skills.items():
                if self.hash_skill_names or len(k) > MAX_SKILL_NAME:
                    k = "h:" + self.hasher("skill:" + k)
                skills[k] = skills.get(k, 0) + v
            rows.append([
                self.hasher("session:" + sid),
                "interactive" if agg["interactive"] else "autonomous",
                _iso_us(agg["first_us"]), _iso_us(agg["last_us"]),
                agg["human_turns"], agg["active_s"], agg["wait_s"],
                agg["wait_excluded_n"], agg["wait_excluded_s"],
                s.assistant_words, s.meta_edits, s.direct_edits, dict(skills),
            ])
        self.counts["sessions_in_window"] = len(rows)
        return rows


# --------------------------------------------------------------------------- optional blocks
def git_commits(repo: Path, ref: str, start: datetime, end: datetime, hasher: Hasher) -> dict:
    """Per-commit aggregates from `git log` (merges excluded). Subjects and bodies are read
    only to set is_fix / is_revert / author class, then dropped."""
    fmt = "%x1e%H%x1f%ad%x1f%s%x1f%B%x1f"
    raw = subprocess.run(
        ["git", "-C", str(repo), "log", ref, f"--since={start.isoformat()}", f"--until={end.isoformat()}",
         "--no-merges", "--numstat", "--date=iso-strict", f"--format={fmt}"],
        capture_output=True, text=True, timeout=600, check=True).stdout
    return git_commits_from_text(raw, hasher, ref=ref)


def git_commits_from_text(raw: str, hasher: Hasher, ref: str = "") -> dict:
    commits = []
    truncated = 0
    for rec in raw.split("\x1e"):
        if not rec.strip():
            continue
        parts = rec.split("\x1f")
        if len(parts) < 4:
            continue
        sha, ad, subj, body = parts[0].strip(), parts[1].strip(), parts[2], parts[3]
        numstat = parts[4] if len(parts) > 4 else ""
        ts = _parse_ts(ad)
        files, ins, dels = [], 0, 0
        for line in numstat.splitlines():
            p = line.split("\t")
            if len(p) == 3:
                a, d, f = p
                files.append(hasher("file:" + f))
                ins += int(a) if a.isdigit() else 0
                dels += int(d) if d.isdigit() else 0
        if len(files) > MAX_FILES_PER_COMMIT:
            files, truncated = files[:MAX_FILES_PER_COMMIT], truncated + 1
        commits.append([
            hasher("commit:" + sha), _iso_us(_us(ts)) if ts else None,
            "agent" if _is_agent_trailer_text(body) else "human",
            bool(FIX_SUBJECT_RE.search(subj)), bool(REVERT_SUBJECT_RE.search(subj)),
            files, ins, dels,
        ])
    out = {"ref": ref, "commit_fields": COMMIT_FIELDS, "commits": commits}
    if truncated:
        out["commits_with_file_list_truncated"] = truncated
    return out


def load_decisions(path: Path, start: datetime, end: datetime) -> dict:
    """Keeps created_ts, resolved_ts and status only. Accepts JSONL or a JSON list."""
    text = path.read_text(errors="ignore")
    items_in = []
    stripped = text.lstrip()
    if stripped.startswith("["):
        loaded = json.loads(stripped)
        items_in = loaded if isinstance(loaded, list) else []
    else:
        for line in text.splitlines():
            if line.strip():
                try:
                    items_in.append(json.loads(line))
                except Exception:
                    continue
    items, outside, unparseable = [], 0, 0
    for d in items_in:
        if not isinstance(d, dict):
            unparseable += 1
            continue
        cts = _parse_ts(d.get("created_ts"))
        if cts is None:
            unparseable += 1
            continue
        if cts < start or cts > end:
            outside += 1
            continue
        rts = _parse_ts(d.get("resolved_ts"))
        status = re.sub(r"[^a-z0-9_\-]", "", str(d.get("status") or "").lower())[:40]
        items.append({"created_ts": _iso_us(_us(cts)), "resolved_ts": _iso_us(_us(rts)) if rts else None,
                      "status": status})
    return {"items": items, "excluded_outside_window": outside, "unparseable": unparseable}


def count_instructions(claude_dir: Path, memory_dir: Path | None) -> dict:
    """Same counts as the parent meter: rule files, SKILL.md files, command files, memory
    files and CLAUDE.md lines (CLAUDE.md beside the .claude directory)."""
    rules = claude_dir / "rules"
    skills = claude_dir / "skills"
    commands = claude_dir / "commands"
    claude_md = claude_dir.parent / "CLAUDE.md"
    out = {
        "rules": sum(1 for _ in rules.rglob("*.md")) if rules.is_dir() else 0,
        "skills": sum(1 for _ in skills.rglob("SKILL.md")) if skills.is_dir() else 0,
        "commands": sum(1 for _ in commands.glob("*.md")) if commands.is_dir() else 0,
        "memory_files": (sum(1 for _ in memory_dir.glob("*.md"))
                         if memory_dir is not None and memory_dir.is_dir() else 0),
        "claude_md_lines": 0,
    }
    if claude_md.exists():
        with open(claude_md, errors="ignore") as fh:
            out["claude_md_lines"] = sum(1 for _ in fh)
    return out


# --------------------------------------------------------------------------- summary
def own_sha256() -> str:
    try:
        return hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
    except Exception:
        return ""


def build_summary(rows: list, counts: dict, *, window_start: str | None, window_end: str | None,
                  window_days: float | None, skill_names: str, git_log=None, decisions=None,
                  instructions=None, transcript_scope: str = "") -> dict:
    return {
        "schema": SCHEMA,
        "extractor_version": EXTRACTOR_VERSION,
        "extractor_sha256": own_sha256(),
        "method": METHOD,
        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds").replace("+00:00", "Z"),
        "contains_message_text": False,
        "hashing": "sha256(per-run random salt + value)[:16]; the salt is not in this file",
        "skill_names": skill_names,
        "transcript_scope": transcript_scope,
        "window": {"start": window_start, "end": window_end, "days": window_days},
        "counts": {k: int(v) for k, v in sorted(counts.items())},
        "session_fields": SESSION_FIELDS,
        "sessions": rows,
        "git_log": git_log,
        "decisions": decisions,
        "instructions": instructions,
    }


def summarize_records(records, *, window_days: float | None = None, hash_skill_names: bool = False,
                      salt: bytes | None = None) -> dict:
    """Library form used by the parity tests: the same aggregation the CLI does, over records
    already in memory, with no window filter (every session counts)."""
    ex = Extraction(Hasher(salt), hash_skill_names=hash_skill_names)
    for rec in records:
        ex.add_record(rec)
    rows = ex.rows(None, None)
    first = min((r[2] for r in rows), default=None, key=lambda s: _us(_parse_ts(s)))
    last = max((r[3] for r in rows), default=None, key=lambda s: _us(_parse_ts(s)))
    return build_summary(rows, ex.counts, window_start=first, window_end=last, window_days=window_days,
                         skill_names="hashed" if hash_skill_names else "clear", transcript_scope="records")


def self_check(path: Path) -> int:
    raw = path.read_bytes()
    s = json.loads(raw)
    print(f"file: {path}  size: {len(raw):,} bytes ({len(raw) / 1024 / 1024:.2f} MB)")
    print(f"schema: {s.get('schema')}  extractor {s.get('extractor_version')} sha256 {s.get('extractor_sha256')}")
    w = s.get("window") or {}
    print(f"window: {w.get('start')} .. {w.get('end')}  days={w.get('days')}")
    print(f"contains_message_text: {s.get('contains_message_text')}  skill names: {s.get('skill_names')}")
    print(f"sessions: {len(s.get('sessions') or [])}  fields: {', '.join(s.get('session_fields') or [])}")
    for k, v in (s.get("counts") or {}).items():
        print(f"  {k}: {v}")
    g = s.get("git_log")
    print(f"git_log: {len(g['commits'])} commits, fields {', '.join(g['commit_fields'])}"
          if g else "git_log: not included")
    d = s.get("decisions")
    print(f"decisions: {len(d['items'])} items (created_ts, resolved_ts, status)" if d else "decisions: not included")
    print(f"instructions: {s.get('instructions') or 'not included'}")
    skills = Counter()
    for row in s.get("sessions") or []:
        for k, v in (row[-1] or {}).items():
            skills[k] += v
    print(f"skill names that will be sent ({len(skills)}): {', '.join(sorted(skills)) or 'none'}")
    longest = 0
    for row in s.get("sessions") or []:
        for v in row:
            if isinstance(v, str):
                longest = max(longest, len(v))
    print(f"longest string in any session row: {longest} chars (timestamps and hashes only)")
    return 0


def main(argv=None) -> int:
    ap = argparse.ArgumentParser(description=__doc__.split("\n\n")[0],
                                 formatter_class=argparse.RawDescriptionHelpFormatter)
    ap.add_argument("--transcripts", type=Path, help="~/.claude/projects/<project> directory")
    ap.add_argument("--days", type=float, default=30.0, help="window ending now (default 30)")
    ap.add_argument("--include-subagents", action="store_true",
                    help="also read <session>/subagents/*.jsonl (pooled into the parent session)")
    ap.add_argument("--git-repo", type=Path, help="repository to summarise with git log")
    ap.add_argument("--git-ref", default="HEAD", help="ref for git log (default HEAD)")
    ap.add_argument("--decisions", type=Path, help="JSONL/JSON list of {created_ts, resolved_ts, status}")
    ap.add_argument("--instructions-dir", type=Path, help="your .claude directory (counts only)")
    ap.add_argument("--memory-dir", type=Path,
                    help="memory directory to count (default <transcripts>/memory)")
    ap.add_argument("--hash-skill-names", action="store_true", help="send salted hashes, not skill names")
    ap.add_argument("--out", type=Path, help="where to write the summary JSON")
    ap.add_argument("--self-check", type=Path, metavar="SUMMARY", help="describe an existing summary")
    ap.add_argument("--quiet", action="store_true")
    a = ap.parse_args(argv)

    if a.self_check:
        return self_check(a.self_check)
    if not a.transcripts or not a.out:
        ap.error("--transcripts and --out are required (or use --self-check SUMMARY)")
    if not a.transcripts.is_dir():
        ap.error(f"not a directory: {a.transcripts}")
    if not 0 < a.days <= 3660:
        ap.error("--days must be between 0 and 3660")

    t0 = time.time()
    end = datetime.now(timezone.utc)
    start = end - timedelta(days=a.days)
    hasher = Hasher()
    ex = Extraction(hasher, hash_skill_names=a.hash_skill_names)
    files = sorted(a.transcripts.glob("*.jsonl"))
    if a.include_subagents:
        files += sorted(a.transcripts.glob("*/subagents/*.jsonl"))
    for k, p in enumerate(files):
        ex.counts["files_seen"] += 1
        try:
            if p.stat().st_mtime < start.timestamp():
                ex.counts["files_skipped_older_than_window"] += 1
                continue
            ex.add_file(p)
            ex.counts["files_scanned"] += 1
        except OSError:
            ex.counts["files_unreadable"] += 1
        if not a.quiet and k % 500 == 0:
            print(f"  {k}/{len(files)} files ({time.time() - t0:.0f}s)", file=sys.stderr)
    rows = ex.rows(_us(start), _us(end))

    git_block = decisions_block = instructions_block = None
    if a.git_repo:
        git_block = git_commits(a.git_repo, a.git_ref, start, end, hasher)
    if a.decisions:
        decisions_block = load_decisions(a.decisions, start, end)
    if a.instructions_dir:
        mem = a.memory_dir if a.memory_dir else a.transcripts / "memory"
        instructions_block = count_instructions(a.instructions_dir, mem)

    summary = build_summary(
        rows, ex.counts, window_start=_iso_us(_us(start)), window_end=_iso_us(_us(end)),
        window_days=a.days, skill_names="hashed" if a.hash_skill_names else "clear",
        git_log=git_block, decisions=decisions_block, instructions=instructions_block,
        transcript_scope="top-level *.jsonl" + (" + subagents" if a.include_subagents else ""))
    data = json.dumps(summary, separators=(",", ":"))
    a.out.write_text(data)
    if not a.quiet:
        print(f"wrote {a.out}: {len(data):,} bytes, {len(rows)} sessions, "
              f"{time.time() - t0:.1f}s", file=sys.stderr)
    return 0


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