#!/usr/bin/env python3
"""Agentmole scanner — 本地扫描你与 AI 的协作历史，产出统计 JSON。

信任声明（请审查）：
  - 本文件从头到尾没有任何网络访问代码（无 socket / urllib / requests / subprocess 出网）。
  - 只读取：~/.claude/projects/**（transcripts）、~/.claude/history.jsonl、
            ~/.codex/sessions/**（rollouts）、~/.codex/history.jsonl、
            ~/.openclaw/agents/*/sessions/**、
            ~/.claude/CLAUDE.md、~/.codex/AGENTS.md、~/.claude/skills、~/.claude/commands、~/.codex/prompts
  - 只写入：--workdir 指定的工作目录（默认 ~/.agentmole/work/）。
  - 依赖：Python 3.9+ 标准库，无第三方包。

用法：
  python3 scan.py scan [--workdir DIR]     # 全量扫描 → report-data.json + 语料分片
  python3 scan.py redact <json文件>        # 对 JSON 里所有字符串值做正则脱敏（原地重写）

口径说明（报告里引用的数字如何得来，全部可复算）：
  - human_msgs        = 你亲手输入的消息条数（Claude transcripts 人类消息 + Codex history）
  - agent_out_tokens  = 主线程 + subagent 输出 token 合计
  - cache_hit_rate    = cache_read / (cache_read + fresh_in)
  - leverage_ratio    = agent_out_tokens / (你输入的总字符数 / 3)   # 字符/3 ≈ 混合中英 token 估算
  - night_owl_index   = 本地时区 00:00–05:59 的人类消息占比
  - correction_rate   = 命中纠偏模式的消息占比（启发式关键词，v1 口径）
  - rules_taught      = ~/.claude/CLAUDE.md + ~/.codex/AGENTS.md + ~/.claude/projects/*/memory/*.md 非空内容行数
  - taming_curve.rules_cumulative = 累计出现的品味/流程声明消息数（启发式）
  - agent_out_tokens_estimated    = Claude transcript 蒸发期外推估算（窗口内平均每消息产出 ×
                                    全史打字消息数 + Codex 实测）；estimation_note 记口径
  - session_turns_avg             = 人类消息总数 / 有人类输入的 session 数
  - skills_installed              = ~/.claude/skills/*/ + ~/.claude/commands/*.md + ~/.codex/prompts/*.md 计数
"""
import json, os, sys, glob, re
from collections import defaultdict, Counter
from datetime import datetime, timezone

SCHEMA_VERSION = 3

# 已确认存在但尚未支持解析的 harness（探测到会在摘要中提示，配方接入后移入正式扫描）。
# 考古结论（2026-07-06）：opencode 本机数据是单日批量实验（1.2 万 session 同日同目录），
# 并入会污染作息/消息口径故不接；cursor 气泡无逐条时间戳只够计数，缓接；
# gemini/qwen 数据量为零；Manus 云端无本地数据不可接。
HARNESS_PROBES = {
    "cursor": "~/.cursor",
    "gemini_cli": "~/.gemini",
    "opencode": "~/.local/share/opencode",
    "qwen_code": "~/.qwen",
}

OPENCLAW_AGENTS = os.path.expanduser("~/.openclaw/agents")

CLAUDE_PROJECTS = os.path.expanduser("~/.claude/projects")
CLAUDE_HISTORY = os.path.expanduser("~/.claude/history.jsonl")
CODEX_SESSIONS = os.path.expanduser("~/.codex/sessions")
CODEX_HISTORY = os.path.expanduser("~/.codex/history.jsonl")
RULE_FILES = [os.path.expanduser("~/.claude/CLAUDE.md"), os.path.expanduser("~/.codex/AGENTS.md")]

SKIP_PREFIXES = ("<command-name>", "<local-command-caveat>", "<local-command-stdout",
                 "<system-reminder>", "[Request interrupted", "<task-notification>")

# ---------- 启发式模式（v1 口径，报告页脚注明是启发式） ----------
RE_CORRECTION = re.compile(
    r"^(不对|不是这|不是的|不行|别这|错了|又错|等等|停|你没|你怎么|回滚|撤销|重来|改回)"
    r"|^(no[,.\s]|wrong|stop|don't|undo|revert|that'?s not|not what i)"
    r"|怎么又|还是错|还是不对|你又|并没有|没解决|没修好|白改|改坏|又坏", re.I)
RE_POLITE = re.compile(r"请|麻烦|谢谢|辛苦|拜托|please|thank", re.I)
RE_TASTE = re.compile(
    r"不应该|永远|从不|必须|禁止|原则|规范|风格|我喜欢|我讨厌|以后都|每次都|记住|以后不要"
    r"|always|never|must|prefer|convention|principle|from now on|rule of thumb", re.I)

# 用途分类（v1 启发式：关键词投票取最高，报告脚注注明口径）
USAGE_BUCKETS = [
    ("design", re.compile(r"设计|样式|配色|字体|排版|美观|好看|太丑|视觉|动效|布局|画一|画个|图标|素材|封面|海报"
                          r"|logo|icon|ui|ux|css|hero|banner|mockup|figma", re.I)),
    ("coding", re.compile(r"代码|函数|报错|编译|测试|重构|实现|接口|数据库|部署|类型|依赖|脚本|修复|回滚"
                          r"|bug|crash|error|fix|npm|python|git|commit|api|log|diff|lint|typescript|refactor", re.I)),
    ("research", re.compile(r"调研|搜索|查一下|查查|了解一下|对比|竞品|资料|分析一下|原理|方案比较|市场"
                            r"|research|survey|benchmark", re.I)),
    ("management", re.compile(r"咋样了|进度|状态|情况如何|继续|推进|监控|盯着|验收|安排|排期|优先级|汇报"
                              r"|status|progress", re.I)),
]

def usage_split(corpus):
    counts = Counter()
    for r in corpus:
        t = r["text"][:400]
        votes = {name: len(pat.findall(t)) for name, pat in USAGE_BUCKETS}
        best = max(votes, key=votes.get)
        counts[best if votes[best] > 0 else "other"] += 1
    n = max(1, len(corpus))
    return {k: round(counts.get(k, 0) / n, 4) for k in ("coding", "design", "research", "management", "other")}

# 等值 API 成本（美元/百万 token，公开牌价近似；口径=如果全按 API 计价）
PRICE_TABLE = [  # (模型名子串, in, out, cache_read)
    ("opus", 15.0, 75.0, 1.5),
    ("sonnet", 3.0, 15.0, 0.3),
    ("haiku", 0.8, 4.0, 0.08),
    ("fable", 20.0, 100.0, 2.0),
    ("gpt", 1.25, 10.0, 0.125),
    ("", 3.0, 15.0, 0.3),  # 兜底
]

def price_of(model):
    m = (model or "").lower()
    for key, i, o, c in PRICE_TABLE:
        if key in m:
            return i, o, c
    return PRICE_TABLE[-1][1:]

# ---------- 脱敏（生成时做，PRD §3.1 规则 2） ----------
REDACT_PATTERNS = [
    (re.compile(r"sk-[A-Za-z0-9_\-]{16,}"), "[REDACTED-KEY]"),
    (re.compile(r"(ghp|gho|ghu|ghs|github_pat)_[A-Za-z0-9_]{16,}"), "[REDACTED-KEY]"),
    (re.compile(r"xox[baprs]-[A-Za-z0-9\-]{10,}"), "[REDACTED-KEY]"),
    (re.compile(r"AKIA[0-9A-Z]{16}"), "[REDACTED-KEY]"),
    (re.compile(r"AIza[0-9A-Za-z_\-]{30,}"), "[REDACTED-KEY]"),
    (re.compile(r"eyJ[A-Za-z0-9_\-]{20,}\.[A-Za-z0-9_\-]{20,}\.[A-Za-z0-9_\-]{10,}"), "[REDACTED-JWT]"),
    (re.compile(r"(?i)(api[_\-]?key|token|secret|passw(or)?d|credential)\s*[:=]\s*['\"]?[A-Za-z0-9_\-\./+]{8,}"), r"\1=[REDACTED]"),
    (re.compile(r"(?i)bearer\s+[A-Za-z0-9_\-\.=/+]{16,}"), "Bearer [REDACTED]"),
    (re.compile(r"[A-Za-z0-9._%+\-]+@[A-Za-z0-9.\-]+\.[A-Za-z]{2,}"), "[REDACTED-EMAIL]"),
    (re.compile(r"\b(?:\d{1,3}\.){3}\d{1,3}\b"), "[REDACTED-IP]"),
    (re.compile(r"(https?://[^\s\"']*[?&](?:key|token|sig|secret|password)=)[^\s\"'&]+", re.I), r"\1[REDACTED]"),
    (re.compile(r"/Users/[A-Za-z0-9_\-]+"), "~"),
    (re.compile(r"/home/[A-Za-z0-9_\-]+"), "~"),
]

def redact_text(s):
    for pat, rep in REDACT_PATTERNS:
        s = pat.sub(rep, s)
    return s

def redact_value(v):
    if isinstance(v, str):
        return redact_text(v)
    if isinstance(v, list):
        return [redact_value(x) for x in v]
    if isinstance(v, dict):
        return {k: redact_value(x) for k, x in v.items()}
    return v

# ---------- 工具 ----------
def text_of(content):
    if isinstance(content, str):
        return content
    if isinstance(content, list):
        return "\n".join(c.get("text", "") for c in content
                         if isinstance(c, dict) and c.get("type") == "text")
    return ""

def to_local(ts_iso):
    try:
        return datetime.fromisoformat(ts_iso.replace("Z", "+00:00")).astimezone()
    except Exception:
        return None

def log(msg):
    print(msg, file=sys.stderr)

# ---------- Claude Code 扫描（源自已验证的试点脚本） ----------
def scan_claude():
    sessions, corpus = [], []
    if not os.path.isdir(CLAUDE_PROJECTS):
        return sessions, corpus
    files = glob.glob(os.path.join(CLAUDE_PROJECTS, "*", "*.jsonl"))
    # 新版把 subagent transcript 存在 <proj>/<session>/subagents/ 下——token 是真实用量必须计入；
    # 文件内自带 isSidechain 标记，走同一套解析即可，只是不计为独立 session
    sub_files = glob.glob(os.path.join(CLAUDE_PROJECTS, "*", "*", "subagents", "*.jsonl"))
    files = files + sub_files
    log(f"[claude] scanning {len(files)} session files (incl. {len(sub_files)} subagent transcripts)")
    for i, path in enumerate(files):
        if i % 500 == 0 and i:
            log(f"  {i}/{len(files)}")
        is_subagent = os.sep + "subagents" + os.sep in path
        rel = os.path.relpath(path, CLAUDE_PROJECTS)
        proj = rel.split(os.sep)[0]
        s = {"first_ts": None, "last_ts": None, "human_msgs": 0, "assistant_calls": 0,
             "in_tok": 0, "out_tok": 0, "cache_read": 0,
             "side_in_tok": 0, "side_out_tok": 0, "side_cache_read": 0, "sidechain_calls": 0,
             "interruptions": 0, "models": Counter()}
        seen = set()
        try:
            with open(path, "r", errors="replace") as f:
                for line in f:
                    if len(line) < 10:
                        continue
                    try:
                        d = json.loads(line)
                    except Exception:
                        continue
                    ts = d.get("timestamp")
                    if ts:
                        if s["first_ts"] is None or ts < s["first_ts"]:
                            s["first_ts"] = ts
                        if s["last_ts"] is None or ts > s["last_ts"]:
                            s["last_ts"] = ts
                    t = d.get("type")
                    if t == "user":
                        if d.get("isMeta") or d.get("isSidechain") or is_subagent:
                            continue
                        msg = d.get("message", {})
                        txt = text_of(msg.get("content"))
                        if not txt or txt.startswith(SKIP_PREFIXES):
                            if txt.startswith("[Request interrupted"):
                                s["interruptions"] += 1  # 你打断 agent 的次数
                            continue
                        origin = d.get("origin", {})
                        kind = origin.get("kind") if isinstance(origin, dict) else None
                        is_human = (kind == "human") or (d.get("promptSource") == "typed")
                        if kind is None and d.get("promptSource") is None:
                            content = msg.get("content")
                            if isinstance(content, list) and any(
                                    isinstance(c, dict) and c.get("type") == "tool_result" for c in content):
                                continue
                            is_human = True
                        if is_human:
                            s["human_msgs"] += 1
                            corpus.append({"src": "claude", "ts": ts, "project": proj, "text": txt[:8000]})
                    elif t == "assistant":
                        msg = d.get("message", {})
                        usage = msg.get("usage")
                        mid = msg.get("id") or d.get("requestId") or d.get("uuid")
                        if not usage or mid in seen:
                            continue
                        seen.add(mid)
                        itok = (usage.get("input_tokens") or 0) + (usage.get("cache_creation_input_tokens") or 0)
                        otok = usage.get("output_tokens") or 0
                        cr = usage.get("cache_read_input_tokens") or 0
                        if d.get("isSidechain"):
                            s["sidechain_calls"] += 1
                            s["side_in_tok"] += itok
                            s["side_out_tok"] += otok
                            s["side_cache_read"] += cr
                        else:
                            s["assistant_calls"] += 1
                            s["in_tok"] += itok
                            s["out_tok"] += otok
                            s["cache_read"] += cr
                        if msg.get("model"):
                            s["models"][msg["model"]] += 1
        except Exception as e:
            log(f"ERR {path}: {e}")
            continue
        if s["human_msgs"] or s["assistant_calls"] or s["sidechain_calls"]:
            s["models"] = dict(s["models"])
            s["is_subagent"] = is_subagent
            if is_subagent:
                # subagent 文件里的 assistant 行有的没带 isSidechain 标——统一归入 side 口径
                s["sidechain_calls"] += s["assistant_calls"]
                s["side_in_tok"] += s["in_tok"]
                s["side_out_tok"] += s["out_tok"]
                s["side_cache_read"] += s["cache_read"]
                s["assistant_calls"] = 0
                s["in_tok"] = s["out_tok"] = s["cache_read"] = 0
                s["human_msgs"] = 0
            sessions.append(s)
    return sessions, corpus

# ---------- Codex 扫描（流式 + 子串预过滤，源自试点脚本） ----------
def scan_codex():
    sessions = []
    if not os.path.isdir(CODEX_SESSIONS):
        return sessions
    files = sorted(glob.glob(os.path.join(CODEX_SESSIONS, "*", "*", "*", "rollout-*.jsonl")))
    log(f"[codex] scanning {len(files)} rollout files")
    for i, path in enumerate(files):
        if i % 1000 == 0 and i:
            log(f"  {i}/{len(files)}")
        rec = {"session_id": None, "originator": None, "cwd": None, "model": None, "effort": None,
               "first_ts": None, "last_ts": None, "user_msgs": 0,
               "in_tok": 0, "cached_tok": 0, "out_tok": 0, "turns": 0}
        last_total = None
        try:
            with open(path, "r", errors="replace") as f:
                for line in f:
                    if not ('"session_meta"' in line or '"token_count"' in line
                            or '"user_message"' in line or '"turn_context"' in line
                            or '"task_started"' in line):
                        continue
                    try:
                        d = json.loads(line)
                    except Exception:
                        continue
                    ts = d.get("timestamp")
                    if ts:
                        if rec["first_ts"] is None:
                            rec["first_ts"] = ts
                        rec["last_ts"] = ts
                    t, p = d.get("type"), d.get("payload", {})
                    if t == "session_meta":
                        rec["session_id"] = p.get("session_id") or p.get("id")
                        rec["originator"] = p.get("originator")
                        rec["cwd"] = p.get("cwd")
                    elif t == "turn_context" and rec["model"] is None:
                        rec["model"] = p.get("model")
                        rec["effort"] = p.get("effort")
                    elif t == "event_msg":
                        pt = p.get("type")
                        if pt == "token_count":
                            tot = (p.get("info") or {}).get("total_token_usage")
                            if tot:
                                last_total = tot
                        elif pt == "user_message":
                            rec["user_msgs"] += 1
                        elif pt == "task_started":
                            rec["turns"] += 1
        except Exception as e:
            log(f"ERR {path}: {e}")
            continue
        if last_total:
            rec["in_tok"] = last_total.get("input_tokens", 0) or 0
            rec["cached_tok"] = last_total.get("cached_input_tokens", 0) or 0
            rec["out_tok"] = last_total.get("output_tokens", 0) or 0
        sessions.append(rec)
    return sessions

def claude_history_corpus(existing_corpus):
    """补上已被 30 天清理蒸发的历史：~/.claude/history.jsonl 保存全部打字记录。
    与 transcript 语料按（分钟桶 + 文本前 50 字）去重，只补缺失的行。"""
    rows = []
    if not os.path.exists(CLAUDE_HISTORY):
        return rows
    seen = set()
    for r in existing_corpus:
        dt = to_local(r["ts"])
        if dt:
            seen.add((dt.strftime("%Y-%m-%d %H:%M"), r["text"][:50]))
    for l in open(CLAUDE_HISTORY, errors="replace"):
        try:
            d = json.loads(l)
        except Exception:
            continue
        t = (d.get("display") or "").strip()
        if not t or t.startswith("/"):
            continue
        dt = datetime.fromtimestamp(d["timestamp"] / 1000, tz=timezone.utc)
        key = (dt.astimezone().strftime("%Y-%m-%d %H:%M"), t[:50])
        if key in seen:
            continue
        seen.add(key)
        proj = os.path.basename(d.get("project") or "?") or "?"
        rows.append({"src": "claude", "ts": dt.isoformat(), "project": proj, "text": t[:8000]})
    return rows

# ---------- OpenClaw 扫描 ----------
RE_OC_META = re.compile(r"^(Conversation info|Sender) \(untrusted metadata\):")
# OpenClaw 会把自动化触发（heartbeat/cron/会话管理）伪装成 user 消息——实测占 7 成，
# 必须剔除，否则用户会"被说了"上千句话
OC_AUTOMATION_PREFIXES = (
    "Read HEARTBEAT.md", "A new session was started via", "Pre-compaction memory flush",
    "[cron:", "[Queued messages", "reply exactly:", "Continue where you left off",
)
RE_OC_TS_PREFIX = re.compile(r"^\[[A-Z][a-z]{2} \d{4}-\d{2}-\d{2} [\d:]+ GMT[+\-]\d+\]\s*")

def _openclaw_user_text(txt):
    """剥掉 openclaw user 消息的外壳：IM 元数据包裹 / 媒体占位 / System 注入 / 自动化触发。"""
    if txt.startswith("[media attached") or txt.startswith("System:"):
        return ""
    if txt.startswith(OC_AUTOMATION_PREFIXES):
        return ""
    if RE_OC_META.match(txt):  # IM 渠道消息：真实文本在最后一个 ``` 围栏后
        txt = txt.rsplit("```", 1)[-1]
        txt = "\n".join(l for l in txt.splitlines() if l and not l.startswith("[message_id"))
    txt = RE_OC_TS_PREFIX.sub("", txt.strip())  # 去掉 IM 转发的时间戳前缀
    if txt.startswith(OC_AUTOMATION_PREFIXES):
        return ""
    return txt.strip()

def scan_openclaw():
    sessions, corpus = [], []
    if not os.path.isdir(OPENCLAW_AGENTS):
        return sessions, corpus
    files = glob.glob(os.path.join(OPENCLAW_AGENTS, "*", "sessions", "*.jsonl*"))
    log(f"[openclaw] scanning {len(files)} session files")
    for path in files:
        agent = path.split(os.sep)[-3]
        s = {"first_ts": None, "last_ts": None, "human_msgs": 0, "assistant_calls": 0,
             "in_tok": 0, "out_tok": 0, "cache_read": 0, "models": Counter()}
        try:
            with open(path, errors="replace") as f:
                for line in f:
                    try:
                        d = json.loads(line)
                    except Exception:
                        continue
                    if d.get("type") != "message":
                        continue
                    ts = d.get("timestamp")
                    if ts:
                        if s["first_ts"] is None:
                            s["first_ts"] = ts
                        s["last_ts"] = ts
                    m = d.get("message", {})
                    if m.get("role") == "user":
                        txt = _openclaw_user_text(text_of(m.get("content")))
                        if txt:
                            s["human_msgs"] += 1
                            corpus.append({"src": "openclaw", "ts": ts,
                                           "project": agent, "text": txt[:8000]})
                    elif m.get("role") == "assistant":
                        u = m.get("usage") or {}
                        if u:
                            s["assistant_calls"] += 1
                            s["in_tok"] += (u.get("input") or 0) + (u.get("cacheWrite") or 0)
                            s["out_tok"] += u.get("output") or 0
                            s["cache_read"] += u.get("cacheRead") or 0
                        if m.get("model"):
                            s["models"][m["model"]] += 1
        except Exception as e:
            log(f"ERR {path}: {e}")
            continue
        if s["human_msgs"] or s["assistant_calls"]:
            s["models"] = dict(s["models"])
            sessions.append(s)
    return sessions, corpus

def codex_corpus(codex_sessions):
    rows = []
    if not os.path.exists(CODEX_HISTORY):
        return rows
    cwd_of = {s["session_id"]: s["cwd"] for s in codex_sessions if s.get("session_id") and s.get("cwd")}
    for l in open(CODEX_HISTORY, errors="replace"):
        try:
            d = json.loads(l)
        except Exception:
            continue
        t = (d.get("text") or "").strip()
        if not t or t.startswith("/"):
            continue
        ts = datetime.fromtimestamp(d["ts"], tz=timezone.utc).isoformat()
        cwd = cwd_of.get(d.get("session_id"), "?")
        rows.append({"src": "codex", "ts": ts, "project": os.path.basename(cwd) if cwd != "?" else "?",
                     "text": t[:8000]})
    return rows

# ---------- 统计与画像 ----------
def clamp(x):
    return max(0.0, min(1.0, x))

def longest_streak(days):
    if not days:
        return 0
    from datetime import date, timedelta
    ds = sorted(date.fromisoformat(d) for d in days)
    best = cur = 1
    for a, b in zip(ds, ds[1:]):
        cur = cur + 1 if (b - a).days == 1 else 1
        best = max(best, cur)
    return best

ZH_STOP = set(("这个 那个 就是 可以 需要 然后 现在 我们 一下 一个 什么 怎么 这样 那样 还是 已经 是不是 不是 直接 应该 里面 上面 下面 时候 如果 所有 全部 但是 因为 所以 其他 或者 继续 开始 使用 这里 那里 出来 进去 东西 地方 感觉 觉得 知道 看看 看一下 弄一下 "
                "没有 这些 可能 一些 为什么 还有 其实 每个 好的 不要 不用 不能 还有 或是 以及 并且 而且 就行 即可 时候 之后 之前 先把 再把 要求 情况 方式 进行 处理 相关 对应 确保 完成 实现 支持 根据 通过 无法 有点 比较 非常 真的 到底 起来 下来 上去").split())
EN_STOP = set("the and for you this that with not are was have has can will your from what when how все all use just like make sure need want does did been being them they their there here would should could into onto over under out about after before again then than very much more most some any each was were its it's don doesn didn isn aren won can't i'm we're you're let's".split())

RE_PASTE_PLACEHOLDER = re.compile(r"\[pasted text[^\]]*\]|\[image[^\]]*\]", re.I)

def clean_text(t):
    """去掉 harness 的粘贴占位符（'[Pasted text #1 +58 lines]'），它不是用户说的话。"""
    return RE_PASTE_PLACEHOLDER.sub(" ", t).strip()

def wordcloud_terms(corpus, top_n=80):
    counts = Counter()
    for r in corpus:
        t = clean_text(r["text"][:300]).lower()
        for w in re.findall(r"[a-z][a-z0-9_\-\.]{2,24}", t):
            if w not in EN_STOP:
                counts[w] += 1
        for run in re.findall(r"[一-鿿]{2,}", t):
            for n in (2, 3):
                for i in range(len(run) - n + 1):
                    counts[run[i:i + n]] += 1
    # 去掉被更长高频词覆盖的短词（"报告页" 吃掉 "报告"）；停用词的子串也一并排除（"为什"⊂"为什么"）
    grams = [(g, c) for g, c in counts.items()
             if c >= 3 and g not in ZH_STOP and not any(g in sw for sw in ZH_STOP if len(sw) > len(g))]
    grams.sort(key=lambda kv: (-len(kv[0]), -kv[1]))
    kept = []
    for g, c in grams:
        covered = any(g in kg and c <= kc * 1.2 for kg, kc in kept)
        if not covered:
            kept.append((g, c))
    kept.sort(key=lambda kv: -kv[1])
    return [{"text": g, "weight": c} for g, c in kept[:top_n]]

def signature_phrases(corpus, top_n=10):
    c = Counter()
    for r in corpus:
        t = clean_text(r["text"])
        if 1 < len(t) <= 30:
            c[t] += 1
    return [{"text": t, "count": n} for t, n in c.most_common(top_n) if n >= 5]

ARCHETYPES = [
    # (id, 中文名, 英文名, tagline, 评分函数)
    ("root-cause-tyrant", "根因主义暴君", "The Root-Cause Tyrant",
     "不修根因不许收工，礼貌是什么，能吃吗",
     lambda d: d["correction_density"] * 0.6 + (1 - d["politeness"]) * 0.4),
    ("one-am-orchestrator", "凌晨一点的编排者", "The 1 A.M. Orchestrator",
     "全世界睡着之后，你的 agent 军团才开始上班",
     lambda d: d["night_shift"] * 0.6 + d["automation_leverage"] * 0.4),
    ("status-bomber", "咋样了轰炸机", "The Status Bomber",
     "你最常输入的不是需求，是催促",
     lambda d: d["repetition"] * 0.7 + (1 - d["msg_length"]) * 0.3),
    ("gentle-pair", "温柔的结对者", "The Gentle Pair-Programmer",
     "对 AI 说请和谢谢的人，运气不会太差",
     lambda d: d["politeness"] * 0.7 + (1 - d["correction_density"]) * 0.3),
    ("free-range-rancher", "放养大师", "The Free-Range Rancher",
     "把任务丢出去就去睡觉，醒来验收",
     lambda d: d["automation_leverage"] * 0.7 + (1 - d["correction_density"]) * 0.3),
    ("microscope-foreman", "显微镜下的监工", "The Microscope Foreman",
     "每一行 diff 都要亲自过目",
     lambda d: (1 - d["automation_leverage"]) * 0.5 + d["correction_density"] * 0.3 + d["msg_length"] * 0.2),
    ("taste-legislator", "品味立法者", "The Taste Legislator",
     "你不是在用 AI，你是在给 AI 立法",
     lambda d: d["taste_signal_density"] * 0.7 + d["rules_norm"] * 0.3),
    ("dissertation-client", "论文型甲方", "The Dissertation Client",
     "一条消息三千字，需求文档级 prompt",
     lambda d: d["msg_length"] * 0.7 + (1 - d["frequency"]) * 0.3),
    ("telegraph-operator", "电报员", "The Telegraph Operator",
     "四个字能说清的事绝不用五个字",
     lambda d: (1 - d["msg_length"]) * 0.6 + d["frequency"] * 0.4),
    ("dual-wield-arbiter", "双修仲裁者", "The Dual-Wield Arbiter",
     "让两家模型互相 review，你坐收渔利",
     lambda d: d["duality"]),
    ("daylight-sprinter", "白日冲刺手", "The Daylight Sprinter",
     "作息健康得不像这个行业的人",
     lambda d: (1 - d["night_shift"]) * 0.5 + d["streak_norm"] * 0.5),
    ("beast-tamer", "驯兽师", "The Beast Tamer",
     "纠偏率一路下降——你把 AI 驯明白了",
     lambda d: d["taming"] * 0.7 + d["rules_norm"] * 0.3),
]

def compute(claude_sessions, codex_sessions, openclaw_sessions, corpus):
    corpus.sort(key=lambda r: r["ts"])
    local_rows = []
    for r in corpus:
        dt = to_local(r["ts"])
        if dt:
            local_rows.append((dt, r))

    # --- 基础量 ---
    c_out = sum(s["out_tok"] + s["side_out_tok"] for s in claude_sessions)
    c_in = sum(s["in_tok"] + s["side_in_tok"] for s in claude_sessions)
    c_cache = sum(s["cache_read"] + s["side_cache_read"] for s in claude_sessions)
    x_out = sum(s["out_tok"] for s in codex_sessions)
    x_fresh = sum(max(0, s["in_tok"] - s["cached_tok"]) for s in codex_sessions)
    x_cache = sum(s["cached_tok"] for s in codex_sessions)
    oc_out = sum(s["out_tok"] for s in openclaw_sessions)
    oc_in = sum(s["in_tok"] for s in openclaw_sessions)
    oc_cache = sum(s["cache_read"] for s in openclaw_sessions)

    out_tok = c_out + x_out + oc_out
    fresh_in = c_in + x_fresh + oc_in
    cache_read = c_cache + x_cache + oc_cache
    human_chars = sum(len(r["text"]) for r in corpus)

    hour_hist = [0] * 24
    weekday_hist = [0] * 7
    daily = Counter()
    monthly_msgs = Counter()
    for dt, r in local_rows:
        hour_hist[dt.hour] += 1
        weekday_hist[dt.weekday()] += 1
        daily[dt.date().isoformat()] += 1
        monthly_msgs[dt.strftime("%Y-%m")] += 1

    models = Counter()
    monthly_out = Counter()
    monthly_sess = Counter()
    for s in claude_sessions:
        for m, n in s["models"].items():
            models[m] += n
        if s["first_ts"]:
            mo = s["first_ts"][:7]
            monthly_out[mo] += s["out_tok"] + s["side_out_tok"]
            if not s.get("is_subagent"):
                monthly_sess[mo] += 1
    for s in codex_sessions:
        if s.get("model"):
            models[s["model"]] += max(1, s["turns"])
        if s["first_ts"]:
            mo = s["first_ts"][:7]
            monthly_out[mo] += s["out_tok"]
            monthly_sess[mo] += 1
    for s in openclaw_sessions:
        for m, n in s["models"].items():
            models[m] += n
        if s["first_ts"]:
            mo = s["first_ts"][:7]
            monthly_out[mo] += s["out_tok"]
            monthly_sess[mo] += 1

    n_msgs = len(corpus) or 1
    night = sum(1 for dt, _ in local_rows if dt.hour < 6) / max(1, len(local_rows))
    corrections = sum(1 for r in corpus if RE_CORRECTION.search(r["text"][:60]))
    polite = sum(1 for r in corpus if RE_POLITE.search(r["text"]))
    taste_rows = [r for r in corpus if RE_TASTE.search(r["text"])]
    streak = longest_streak(daily.keys())
    rules_taught = 0
    rule_paths = list(RULE_FILES) + glob.glob(os.path.expanduser("~/.claude/projects/*/memory/*.md"))
    for p in rule_paths:
        if os.path.exists(p):
            rules_taught += sum(1 for l in open(p, errors="replace")
                                if l.strip() and not l.strip().startswith("#"))

    # 已安装 skill/command 数（口径：claude skills 目录 + claude commands + codex prompts）
    skills_installed = (len(glob.glob(os.path.expanduser("~/.claude/skills/*/")))
                        + len(glob.glob(os.path.expanduser("~/.claude/commands/*.md")))
                        + len(glob.glob(os.path.expanduser("~/.codex/prompts/*.md"))))

    # 平均每 session 人类消息数（只计交互式 session：codex 的 exec 自动派发不算，
    # 否则 2.9 万个机器 rollout 会把分母冲垮）
    active_sessions = (sum(1 for s in claude_sessions if s["human_msgs"])
                       + sum(1 for s in codex_sessions
                             if s["user_msgs"] and s.get("originator") != "codex_exec")
                       + sum(1 for s in openclaw_sessions if s["human_msgs"]))
    session_turns_avg = round(n_msgs / active_sessions, 1) if active_sessions else None

    # 阵营只看 Claude vs Codex 两大家（openclaw 等自建 harness 不参与站队）
    claude_msgs = sum(1 for r in corpus if r["src"] == "claude")
    codex_msgs = sum(1 for r in corpus if r["src"] == "codex")
    camp_total = max(1, claude_msgs + codex_msgs)
    minority = min(claude_msgs, codex_msgs) / camp_total
    camp = "dual" if minority >= 0.15 else ("claude" if claude_msgs >= codex_msgs else "codex")

    human_tok_est = max(1, human_chars // 3)
    leverage = out_tok / human_tok_est
    correction_rate = corrections / n_msgs
    avg_chars = human_chars / n_msgs
    span_days = max(1, len(daily))

    # --- 新维度（owner 2026-07-06 晚）：用途分类 / 等值成本 / 并行度 / 打断 / 周末 / 外包率 ---
    split = usage_split(corpus)

    cost = 0.0
    for s in claude_sessions:
        model = max(s["models"], key=s["models"].get) if s["models"] else ""
        pi, po, pc = price_of(model)
        cost += ((s["in_tok"] + s["side_in_tok"]) * pi + (s["out_tok"] + s["side_out_tok"]) * po
                 + (s["cache_read"] + s["side_cache_read"]) * pc) / 1e6
    for s in codex_sessions:
        pi, po, pc = price_of(s.get("model"))
        cost += (max(0, s["in_tok"] - s["cached_tok"]) * pi + s["out_tok"] * po + s["cached_tok"] * pc) / 1e6
    for s in openclaw_sessions:
        model = max(s["models"], key=s["models"].get) if s["models"] else ""
        pi, po, pc = price_of(model)
        cost += (s["in_tok"] * pi + s["out_tok"] * po + s["cache_read"] * pc) / 1e6

    hour_projects = defaultdict(set)
    weekend = 0
    for dt, r in local_rows:
        hour_projects[dt.strftime("%Y-%m-%d %H")].add(r["project"])
        if dt.weekday() >= 5:
            weekend += 1
    parallel_peak = max((len(v) for v in hour_projects.values()), default=0)

    # effort 档位分布（目前只有 Codex 暴露该指标；按 rollout 计数）
    effort_split = dict(Counter(s["effort"] for s in codex_sessions if s.get("effort")).most_common())

    interruptions = sum(s.get("interruptions", 0) for s in claude_sessions)
    main_calls = sum(s["assistant_calls"] for s in claude_sessions)
    side_calls = sum(s["sidechain_calls"] for s in claude_sessions)
    sidechain_share = round(side_calls / max(1, main_calls + side_calls), 4)

    def norm_proj(p):
        # claude transcript 目录名是路径的连字符编码（"-Users-k-MyPlayground-Weaver"）；归一到项目短名
        if "-MyPlayground-" in p:
            return p.split("-MyPlayground-")[-1]
        if p.startswith("-Users-"):
            return p.rsplit("-", 1)[-1]
        return p
    proj_counter = Counter(norm_proj(r["project"]) for r in corpus if r["project"] != "?")
    project_top = [{"project": p, "share": round(n / n_msgs, 4)} for p, n in proj_counter.most_common(5)]

    # 磨合曲线：分月纠偏率 + 累计规则声明
    months = sorted(monthly_msgs)
    curve = []
    rules_cum = 0
    taste_by_month = Counter(to_local(r["ts"]).strftime("%Y-%m") for r in taste_rows if to_local(r["ts"]))
    corr_by_month = Counter(to_local(r["ts"]).strftime("%Y-%m") for r in corpus
                            if RE_CORRECTION.search(r["text"][:60]) and to_local(r["ts"]))
    for mo in months:
        rules_cum += taste_by_month.get(mo, 0)
        curve.append({"month": mo,
                      "correction_rate": round(corr_by_month.get(mo, 0) / max(1, monthly_msgs[mo]), 4),
                      "rules_cumulative": rules_cum})

    # 驯化：前半段 vs 后半段纠偏率下降幅度
    half = len(curve) // 2
    if half >= 1:
        first = sum(c["correction_rate"] for c in curve[:half]) / half
        second = sum(c["correction_rate"] for c in curve[half:]) / (len(curve) - half)
        taming = clamp((first - second) / max(first, 1e-6))
    else:
        taming = 0.0

    sig = signature_phrases(corpus)
    top_repeat = sig[0]["count"] if sig else 0

    dims = {
        "automation_leverage": clamp((len(str(int(leverage))) - 1) / 5),  # log10 量级归一
        "correction_density": clamp(correction_rate / 0.15),
        "night_shift": clamp(night / 0.4),
        "msg_length": clamp(avg_chars / 400),
        "politeness": clamp(polite / n_msgs / 0.15),
        "taste_signal_density": clamp(len(taste_rows) / n_msgs / 0.08),
        # 以下为选型用隐藏维度，不进报告雷达图
        "repetition": clamp(top_repeat / 150),
        "frequency": clamp(n_msgs / span_days / 60),
        "duality": clamp(minority / 0.35),
        "streak_norm": clamp(streak / 120),  # 45 天就饱和会让重度用户全变"冲刺手"，上限放宽
        "rules_norm": clamp(rules_taught / 150),
        "taming": taming,
    }
    aid, zh, en, tagline, _ = max(ARCHETYPES, key=lambda a: a[4](dims))

    badges = []
    longest_msg = max((len(r["text"]) for r in corpus), default=0)
    if longest_msg >= 3000:
        badges.append({"id": "dissertation", "name": "万字甲方", "desc": f"单条消息最长 {longest_msg} 字",
                       "evidence": f"{longest_msg} chars"})
    if top_repeat >= 6:
        badges.append({"id": "same-msg-repeat", "name": "复读机", "desc": f"同一句话发了 {top_repeat} 次",
                       "evidence": sig[0]["text"]})
    hit = max((m for m in (1000, 5000, 10000, 20000, 50000) if n_msgs >= m), default=0)
    if hit:
        badges.append({"id": f"msg-{hit}", "name": f"第 {hit:,} 条消息", "desc": f"累计亲手输入 {n_msgs:,} 条消息",
                       "evidence": str(n_msgs)})
    if camp == "dual":
        badges.append({"id": "dual-wield", "name": "跨模型仲裁者", "desc": "同时驱使 Claude 与 Codex",
                       "evidence": f"claude {claude_msgs} / codex {codex_msgs}"})
    if streak >= 14:
        badges.append({"id": "streak", "name": f"连跑 {streak} 天", "desc": "最长连续活跃天数",
                       "evidence": str(streak)})
    if night >= 0.25:
        badges.append({"id": "night-owl", "name": "夜行动物", "desc": f"{night:.0%} 的消息发生在 0–6 点",
                       "evidence": f"{night:.2f}"})
    if cache_read and cache_read / max(1, cache_read + fresh_in) >= 0.9:
        badges.append({"id": "cache-lord", "name": "缓存领主", "desc": "cache 命中率超过 90%",
                       "evidence": f"{cache_read/(cache_read+fresh_in):.2%}"})

    sources = []
    if claude_sessions:
        sources.append("claude_code")
    if codex_sessions:
        sources.append("codex")
    if openclaw_sessions:
        sources.append("openclaw")

    # --- per-source 对比（各 agent 使用频率 / 产出 / 各自吐槽位） ---
    per_source = {}
    src_map = {"claude": ("claude_code", c_out, sum(1 for s in claude_sessions if not s.get("is_subagent"))),
               "codex": ("codex", x_out, len(codex_sessions)),
               "openclaw": ("openclaw", oc_out, len(openclaw_sessions))}
    for src, (key, out, n_sess) in src_map.items():
        rows = [r for r in corpus if r["src"] == src]
        if not rows:
            continue
        days = {dt.date().isoformat() for dt, r in local_rows if r["src"] == src}
        per_source[key] = {
            "human_msgs": len(rows), "sessions": n_sess, "out_tokens": out,
            "active_days": len(days), "share": round(len(rows) / n_msgs, 4),
            "roast": None,  # 深挖模式由 host agent 撰写（基于真实数字）
        }

    # --- Claude 蒸发期外推 ---
    # transcript（含 token 用量）只存活 ~cleanupPeriodDays 天，但 history.jsonl 打字记录是全史。
    # 外推口径：存活窗口内 每条人类消息平均产出 token × 全史 claude 消息数 + codex 实测全量。
    transcript_msgs = sum(s["human_msgs"] for s in claude_sessions)
    claude_all_msgs = per_source.get("claude_code", {}).get("human_msgs", 0)
    est_total = None
    est_note = None
    if transcript_msgs and claude_all_msgs > transcript_msgs * 1.15:
        tokens_per_msg = c_out / transcript_msgs
        est_claude = int(tokens_per_msg * claude_all_msgs)
        est_total = est_claude + x_out + oc_out
        est_note = (f"Claude Code transcript 仅存活最近窗口（{transcript_msgs} 条消息对应 {c_out:,} tokens），"
                    f"更早月份已被 30 天清理策略蒸发；按全史 {claude_all_msgs} 条打字记录 × 窗口内平均每消息产出外推。"
                    f"其余来源为全史实测值。")

    data = {
        "schema_version": SCHEMA_VERSION,
        "generated_at": datetime.now(timezone.utc).isoformat(timespec="seconds"),
        "handle": None,
        "sources": sources,
        "period": {"from": corpus[0]["ts"][:10] if corpus else None,
                   "to": corpus[-1]["ts"][:10] if corpus else None},
        "mode": "fast",
        "stats": {
            "sessions": sum(1 for s in claude_sessions if not s.get("is_subagent")) + len(codex_sessions) + len(openclaw_sessions),
            "active_days": len(daily),
            "human_msgs": n_msgs,
            "agent_out_tokens": out_tok,
            "fresh_in_tokens": fresh_in,
            "cache_read_tokens": cache_read,
            "cache_hit_rate": round(cache_read / max(1, cache_read + fresh_in), 4),
            # 全口径总处理量 = 输出 + 新鲜输入 + cache 重读；与 ccusage 等工具的口径对齐，
            # 避免用户拿其他统计工具对比时以为我们算少了
            "total_processed_tokens": out_tok + fresh_in + cache_read,
            "api_calls": (sum(s["assistant_calls"] + s["sidechain_calls"] for s in claude_sessions)
                          + sum(s["assistant_calls"] for s in openclaw_sessions)),
            "leverage_ratio": round(leverage, 1),
            "longest_streak_days": streak,
            "night_owl_index": round(night, 4),
            "correction_rate": round(correction_rate, 4),
            "rules_taught": rules_taught,
            "hour_histogram": hour_hist,
            "weekday_histogram": weekday_hist,
            "daily_activity": dict(daily),
            "monthly": [{"month": mo, "human_msgs": monthly_msgs[mo],
                         "out_tokens": monthly_out.get(mo, 0), "sessions": monthly_sess.get(mo, 0)}
                        for mo in months],
            "models": dict(models.most_common(12)),
            "camp": camp,
            "session_turns_avg": session_turns_avg,
            "skills_installed": skills_installed,
            "agent_out_tokens_estimated": est_total,
            "estimation_note": est_note,
            "usage_split": split,
            "api_cost_estimated_usd": round(cost),
            "api_cost_note": "按各模型公开 API 牌价对实测 token 估算（输入/输出/cache 分开计价）；订阅制实际支出远低于此——这就是杠杆",
            "parallel_peak_projects_hour": parallel_peak,
            "weekend_share": round(weekend / max(1, len(local_rows)), 4),
            "interruptions": interruptions,
            "sidechain_share": sidechain_share,
            "project_top": project_top,
            "effort_split": effort_split,
        },
        "per_source": per_source,
        "archetype": {
            "id": aid, "name": zh, "name_en": en, "tagline": tagline,
            "dimensions": {k: round(dims[k], 3) for k in
                           ("automation_leverage", "correction_density", "night_shift",
                            "msg_length", "politeness", "taste_signal_density")},
        },
        "narrative": None,
        "quotes": {"harshest": None, "signature": sig, "roast": None},
        "moments": [],
        "diagnosis": [],
        "wordcloud": {"words": wordcloud_terms(corpus)},
        "taming_curve": curve,
        "badges": badges,
        "creeds": [],
        "distilled_skills": [],
    }
    return data

def write_chunks(corpus, workdir, n=10):
    """把语料按时间均分为 n 片，供深挖模式的并行 subagent 使用。"""
    if not corpus:
        return []
    size = (len(corpus) + n - 1) // n
    manifest = []
    for i in range(0, len(corpus), size):
        chunk = corpus[i:i + size]
        name = f"corpus-chunk-{i//size+1:02d}.jsonl"
        with open(os.path.join(workdir, name), "w") as f:
            for r in chunk:
                f.write(json.dumps(r, ensure_ascii=False) + "\n")
        manifest.append({"file": name, "rows": len(chunk),
                         "from": chunk[0]["ts"][:10], "to": chunk[-1]["ts"][:10]})
    return manifest

def main():
    if len(sys.argv) < 2 or sys.argv[1] not in ("scan", "redact"):
        print(__doc__)
        sys.exit(1)
    if sys.argv[1] == "redact":
        path = sys.argv[2]
        with open(path) as f:
            data = json.load(f)
        data = redact_value(data)
        with open(path, "w") as f:
            json.dump(data, f, ensure_ascii=False, indent=1)
        log(f"redacted in place: {path}")
        return

    workdir = os.path.expanduser("~/.agentmole/work")
    if "--workdir" in sys.argv:
        workdir = sys.argv[sys.argv.index("--workdir") + 1]
    os.makedirs(workdir, exist_ok=True)

    claude_sessions, claude_corpus = scan_claude()
    claude_corpus += claude_history_corpus(claude_corpus)
    codex_sessions = scan_codex()
    openclaw_sessions, openclaw_rows = scan_openclaw()
    corpus = claude_corpus + codex_corpus(codex_sessions) + openclaw_rows
    if not corpus:
        log("未找到任何 Claude Code / Codex / OpenClaw 历史数据。")
        sys.exit(2)

    data = compute(claude_sessions, codex_sessions, openclaw_sessions, corpus)
    # 脱敏在生成时做：写盘前对全部字符串值过一遍正则脱敏
    data = redact_value(data)
    out = os.path.join(workdir, "report-data.json")
    with open(out, "w") as f:
        json.dump(data, f, ensure_ascii=False, indent=1)

    corpus.sort(key=lambda r: r["ts"])
    manifest = write_chunks(corpus, workdir)
    with open(os.path.join(workdir, "chunks.json"), "w") as f:
        json.dump(manifest, f, ensure_ascii=False, indent=1)

    detected = [name for name, p in HARNESS_PROBES.items() if os.path.isdir(os.path.expanduser(p))]
    s = data["stats"]
    log(f"done → {out}")
    print(json.dumps({
        "workdir": workdir,
        "human_msgs": s["human_msgs"], "sessions": s["sessions"],
        "active_days": s["active_days"], "agent_out_tokens": s["agent_out_tokens"],
        "agent_out_tokens_estimated": s["agent_out_tokens_estimated"],
        "archetype": data["archetype"]["name"],
        "chunks": len(manifest),
        "deep_mode_recommended": s["human_msgs"] >= 2000,
        "detected_unsupported_harnesses": detected,
    }, ensure_ascii=False, indent=1))

if __name__ == "__main__":
    main()