"""Codex session adapter.

Codex (OpenAI) stores each session as a JSONL "rollout" at
`~/.codex/sessions/<YYYY>/<MM>/<DD>/rollout-*.jsonl`. Lines are wrapper objects
`{type, timestamp, payload}`. The types we use:

  session_meta              -> payload.cwd, payload.timestamp, payload.model_provider
  turn_context              -> payload.model (the concrete model id)
  event_msg/user_message    -> payload.message|text (first one = session title)
  event_msg/agent_message   -> payload.message|text
  response_item/message     -> payload.role + payload.content ([{type,text}] or str)

This adapter exposes the same shape session-search uses for Claude transcripts
(cwd, title, plus a (role, text) iterator for grep), so routing is a drop-in.
"""
from __future__ import annotations

import json
from pathlib import Path
from typing import Iterator

CODEX_ROOT = Path.home() / ".codex" / "sessions"


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


def iter_turns(path: Path) -> Iterator[tuple[str, str]]:
    """Yield (role, text) for the conversational turns in a rollout.

    Codex records each turn twice: a clean `event_msg` (user_message /
    agent_message) and a lower-level `response_item/message` that also carries
    system/developer preamble. We use the clean channel and fall back to
    response_item only if a rollout has no event_msg turns at all.
    """
    event_turns: list[tuple[str, str]] = []
    item_turns: list[tuple[str, str]] = []
    try:
        with path.open(encoding="utf-8", errors="replace") as f:
            for line in f:
                try:
                    e = json.loads(line)
                except json.JSONDecodeError:
                    continue
                p = e.get("payload") or {}
                if not isinstance(p, dict):
                    continue
                typ, ptyp = e.get("type"), p.get("type")
                if typ == "event_msg" and ptyp == "user_message":
                    txt = p.get("message") or p.get("text") or ""
                    if isinstance(txt, str) and txt:
                        event_turns.append(("user", txt))
                elif typ == "event_msg" and ptyp == "agent_message":
                    txt = p.get("message") or p.get("text") or ""
                    if isinstance(txt, str) and txt:
                        event_turns.append(("assistant", txt))
                elif typ == "response_item" and ptyp == "message":
                    txt = _content_text(p.get("content"))
                    if txt and p.get("role") in ("user", "assistant"):
                        item_turns.append((p.get("role"), txt))
    except OSError:
        return
    yield from (event_turns or item_turns)


def peek(path: Path) -> tuple[str, str]:
    """Return (cwd, first_user_line) for a rollout, reading only the head."""
    cwd = ""
    first_user = ""
    try:
        with path.open(encoding="utf-8", errors="replace") as f:
            for i, line in enumerate(f):
                if i > 120 and cwd and first_user:
                    break
                try:
                    e = json.loads(line)
                except json.JSONDecodeError:
                    continue
                p = e.get("payload") or {}
                if not isinstance(p, dict):
                    continue
                if not cwd and e.get("type") == "session_meta":
                    cwd = p.get("cwd", "") or ""
                if (not first_user and e.get("type") == "event_msg"
                        and p.get("type") == "user_message"):
                    txt = p.get("message") or p.get("text") or ""
                    if isinstance(txt, str) and txt.strip():
                        first_user = txt.strip().splitlines()[0]
    except OSError:
        pass
    return cwd, first_user


def iter_sessions() -> Iterator[dict]:
    """Yield session records shaped like session-search's Claude records."""
    if not CODEX_ROOT.is_dir():
        return
    for jsonl in CODEX_ROOT.glob("*/*/*/rollout-*.jsonl"):
        try:
            st = jsonl.stat()
        except OSError:
            continue
        cwd, title = peek(jsonl)
        yield {
            "kind": "codex",
            "mtime": st.st_mtime,
            "size": st.st_size,
            "title": title,
            "cwd": cwd,
            "path": str(jsonl),
        }


def _usage_from_last(lt: dict) -> dict:
    """Map a Codex last_token_usage record to the Anthropic-style usage dict the
    token-doctor pipeline expects (uncached input + cache_read split; OpenAI has
    no separate cache-write metric).
    """
    inp = int(lt.get("input_tokens") or 0)
    cached = int(lt.get("cached_input_tokens") or 0)
    out = int(lt.get("output_tokens") or 0) + int(lt.get("reasoning_output_tokens") or 0)
    return {
        "input": max(inp - cached, 0),
        "output": out,
        "cache_read": cached,
        "cache_creation": 0,
    }


def session_meta(path: Path) -> dict:
    """Return {cwd, model, start_ts} for a rollout."""
    cwd, model, start_ts = "", "", ""
    try:
        with path.open(encoding="utf-8", errors="replace") as f:
            for i, line in enumerate(f):
                if i > 200 and cwd and model:
                    break
                try:
                    e = json.loads(line)
                except json.JSONDecodeError:
                    continue
                p = e.get("payload") or {}
                if not isinstance(p, dict):
                    continue
                if e.get("type") == "session_meta":
                    cwd = cwd or p.get("cwd", "") or ""
                    start_ts = start_ts or p.get("timestamp", "") or e.get("timestamp", "")
                elif e.get("type") == "turn_context" and p.get("model"):
                    model = model or p["model"]
    except OSError:
        pass
    return {"cwd": cwd, "model": model, "start_ts": start_ts}


def codex_turns(path: Path, model_hint: str = "") -> list[dict]:
    """Parse a rollout into turn dicts matching the token-doctor/task-profile
    shape: {role, ts, text, tool_calls, [model, usage]}.

    Usage is attached from each `token_count` event's per-response
    `last_token_usage` to the assistant turn it belongs to.
    """
    model = model_hint
    turns: list[dict] = []
    pending_tools: list[str] = []
    try:
        with path.open(encoding="utf-8", errors="replace") as f:
            for line in f:
                try:
                    e = json.loads(line)
                except json.JSONDecodeError:
                    continue
                p = e.get("payload") or {}
                if not isinstance(p, dict):
                    continue
                typ, ptyp = e.get("type"), p.get("type")
                ts = e.get("timestamp")
                if typ == "turn_context" and p.get("model"):
                    model = p["model"]
                elif typ == "event_msg" and ptyp == "user_message":
                    txt = p.get("message") or p.get("text") or ""
                    if isinstance(txt, str) and txt:
                        turns.append({"role": "user", "ts": ts, "text": txt, "tool_calls": []})
                elif typ == "event_msg" and ptyp == "agent_message":
                    txt = p.get("message") or p.get("text") or ""
                    turns.append({"role": "assistant", "ts": ts,
                                  "text": txt if isinstance(txt, str) else "",
                                  "tool_calls": pending_tools})
                    pending_tools = []
                elif typ == "response_item" and ptyp == "function_call":
                    name = p.get("name") or p.get("tool_name")
                    if name:
                        pending_tools.append(name)
                elif typ == "event_msg" and ptyp == "token_count":
                    lt = (p.get("info") or {}).get("last_token_usage") or {}
                    if not lt:
                        continue
                    target = next((t for t in reversed(turns)
                                   if t["role"] == "assistant" and "usage" not in t), None)
                    if target is None:
                        target = {"role": "assistant", "ts": ts, "text": "", "tool_calls": []}
                        turns.append(target)
                    if pending_tools:
                        target["tool_calls"] = target.get("tool_calls", []) + pending_tools
                        pending_tools = []
                    target["model"] = model or "gpt-5"
                    target["usage"] = _usage_from_last(lt)
    except OSError:
        return []
    return turns

#!/usr/bin/env python3
"""Find past Claude Code (CLI) and Claude Cowork (desktop) sessions on this Mac.

Reads transcripts directly from disk. Works for any user , paths come from $HOME.
See ../SKILL.md for when to use this.
"""
from __future__ import annotations

import argparse
import json
import re
import sys
from datetime import datetime
from pathlib import Path
from typing import Iterator

sys.path.insert(0, str(Path(__file__).parent))
import codex_sessions  # noqa: E402
from host_platform import CLAUDE, CODEX, degrade, detect_platform  # noqa: E402

HOME = Path.home()
CODE_ROOT = HOME / ".claude" / "projects"
COWORK_ROOT = HOME / "Library" / "Application Support" / "Claude" / "local-agent-mode-sessions"


def _extract_text(entry: dict) -> str:
    """Pull user/assistant text out of a transcript entry. Skip tool calls/results/system noise."""
    t = entry.get("type")
    if t not in ("user", "assistant"):
        return ""
    msg = entry.get("message") or {}
    content = msg.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":
                txt = c.get("text") or ""
                if txt:
                    parts.append(txt)
        return "\n".join(parts)
    return ""


def _peek_code(jsonl: Path) -> tuple[str, str]:
    """Return (cwd, first_user_line) for a Claude Code transcript."""
    cwd = ""
    first_user = ""
    try:
        with jsonl.open(encoding="utf-8", errors="replace") as f:
            for i, line in enumerate(f):
                if i > 80 and cwd and first_user:
                    break
                try:
                    e = json.loads(line)
                except json.JSONDecodeError:
                    continue
                if not cwd and isinstance(e.get("cwd"), str):
                    cwd = e["cwd"]
                if not first_user and e.get("type") == "user":
                    txt = _extract_text(e)
                    if txt:
                        first_user = txt.strip().splitlines()[0]
    except OSError:
        pass
    return cwd, first_user


def _cowork_title(audit: Path) -> str:
    session_dir = audit.parent
    meta_file = session_dir.parent / f"{session_dir.name}.json"
    if not meta_file.is_file():
        return ""
    try:
        meta = json.loads(meta_file.read_text(encoding="utf-8", errors="replace"))
    except (json.JSONDecodeError, OSError):
        return ""
    return (meta.get("title") or meta.get("name") or (meta.get("initialMessage") or "")[:120] or "").strip()


def iter_sessions(kind: str) -> Iterator[dict]:
    if kind in ("code", "all") and CODE_ROOT.is_dir():
        for project_dir in CODE_ROOT.iterdir():
            if not project_dir.is_dir():
                continue
            for jsonl in project_dir.glob("*.jsonl"):
                try:
                    st = jsonl.stat()
                except OSError:
                    continue
                cwd, title = _peek_code(jsonl)
                yield {
                    "kind": "code",
                    "mtime": st.st_mtime,
                    "size": st.st_size,
                    "title": title,
                    "cwd": cwd,
                    "path": str(jsonl),
                }
    if kind in ("cowork", "all") and COWORK_ROOT.is_dir():
        for audit in COWORK_ROOT.rglob("local_*/audit.jsonl"):
            if "skills-plugin" in audit.parts:
                continue
            try:
                st = audit.stat()
            except OSError:
                continue
            yield {
                "kind": "cowork",
                "mtime": st.st_mtime,
                "size": st.st_size,
                "title": _cowork_title(audit),
                "cwd": "",
                "path": str(audit),
            }


def grep_session(path: Path, pattern: re.Pattern, max_snippets: int) -> list[str]:
    """Return up to max_snippets short excerpts where pattern matched."""
    snippets: list[str] = []
    try:
        with path.open(encoding="utf-8", errors="replace") as f:
            for line in f:
                if len(snippets) >= max_snippets:
                    break
                try:
                    e = json.loads(line)
                except json.JSONDecodeError:
                    continue
                txt = _extract_text(e)
                if not txt:
                    continue
                for m in pattern.finditer(txt):
                    start = max(0, m.start() - 60)
                    end = min(len(txt), m.end() + 60)
                    snip = txt[start:end].replace("\n", " ").strip()
                    role = e.get("type", "?")
                    snippets.append(f"[{role}] …{snip}…")
                    if len(snippets) >= max_snippets:
                        break
    except OSError:
        pass
    return snippets


def grep_codex(path: Path, pattern: re.Pattern, max_snippets: int) -> list[str]:
    """grep_session equivalent for a Codex rollout (uses the codex adapter)."""
    snippets: list[str] = []
    for role, txt in codex_sessions.iter_turns(path):
        if len(snippets) >= max_snippets:
            break
        for m in pattern.finditer(txt):
            start = max(0, m.start() - 60)
            end = min(len(txt), m.end() + 60)
            snip = txt[start:end].replace("\n", " ").strip()
            snippets.append(f"[{role}] …{snip}…")
            if len(snippets) >= max_snippets:
                break
    return snippets


def fmt_size(b: float) -> str:
    for u in ("B", "K", "M", "G"):
        if b < 1024:
            return f"{b:.0f}{u}"
        b /= 1024
    return f"{b:.0f}T"


def parse_date(s: str) -> float:
    return datetime.strptime(s, "%Y-%m-%d").timestamp()


def main() -> int:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument("--kind", choices=("code", "cowork", "all"), default="all")
    p.add_argument("--since", help="YYYY-MM-DD (inclusive)")
    p.add_argument("--until", help="YYYY-MM-DD (inclusive)")
    p.add_argument("--title", help="substring, case-insensitive")
    p.add_argument("--cwd", help="substring match against code session cwd")
    p.add_argument("--grep", help="regex searched against transcript body (case-insensitive)")
    p.add_argument("--snippets", type=int, default=2, help="matching snippets per session (default 2)")
    p.add_argument("-n", "--limit", type=int, default=50, help="max sessions to output (0 = all)")
    p.add_argument("--json", action="store_true")
    args = p.parse_args()

    # Route by host platform. Claude Code reads its own transcripts; Codex reads
    # ~/.codex/sessions rollouts; Antigravity has no parseable local store.
    platform = detect_platform()
    if platform not in (CLAUDE, CODEX):
        degrade("session-search", platform)

    since_ts = parse_date(args.since) if args.since else None
    until_ts = parse_date(args.until) + 86400 if args.until else None
    title_q = args.title.lower() if args.title else None
    cwd_q = args.cwd.lower() if args.cwd else None
    grep_re = re.compile(args.grep, re.IGNORECASE) if args.grep else None

    if platform == CODEX:
        sessions = list(codex_sessions.iter_sessions())
        grep_fn = grep_codex
    else:
        sessions = list(iter_sessions(args.kind))
        grep_fn = grep_session
    sessions.sort(key=lambda s: s["mtime"], reverse=True)

    results = []
    for s in sessions:
        if since_ts is not None and s["mtime"] < since_ts:
            continue
        if until_ts is not None and s["mtime"] > until_ts:
            continue
        if title_q and title_q not in (s["title"] or "").lower():
            continue
        if cwd_q and cwd_q not in (s["cwd"] or "").lower():
            continue
        if grep_re is not None:
            snips = grep_fn(Path(s["path"]), grep_re, args.snippets)
            if not snips:
                continue
            s = dict(s, snippets=snips)
        results.append(s)
        if args.limit > 0 and len(results) >= args.limit:
            break

    if args.json:
        json.dump(results, sys.stdout, indent=2, default=str)
        sys.stdout.write("\n")
        return 0

    if not results:
        print("(no matching sessions)")
        return 0

    print(f"{'WHEN':<17} {'KIND':<7} {'SIZE':>6}  LABEL")
    print("-" * 100)
    for s in results:
        ts = datetime.fromtimestamp(s["mtime"]).strftime("%Y-%m-%d %H:%M")
        if s["kind"] in ("code", "codex"):
            label = s["title"] or "(no user message yet)"
            if s["cwd"]:
                label = f"{label}  ·  {s['cwd']}"
        else:
            label = s["title"] or "(untitled cowork session)"
        print(f"{ts:<17} {s['kind']:<7} {fmt_size(s['size']):>6}  {label[:120]}")
        print(f"{'':<17} path: {s['path']}")
        for snip in s.get("snippets", []):
            print(f"{'':<17}   {snip[:180]}")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

"""Host-platform detection for the ai-adoption skills.

One mechanism, shared by every script that reads agent session history. An
identical copy ships in each skill's scripts/ dir so it travels with the script
under all install shapes (Claude Code native, Codex flat, Antigravity nested).

Resolution order (first hit wins):
  1. AI_FIRST_PLATFORM env var, if set to a known platform (explicit override).
  2. The "platform" field stamped into .techwolf-plugin.json by install.sh.
     The installer knows the target IDE (--ide codex|antigravity), so this is
     deterministic for Codex/Antigravity installs.
  3. Fallback: if ~/.claude exists, assume Claude Code. Claude Code uses the
     native plugin system and never runs install.sh, so it is never stamped.
  4. Default: "claude".

Per-platform session-data reality (see each skill's SKILL.md):
  - claude       Claude Code (~/.claude/projects) + Cowork transcripts. Full.
  - codex        ~/.codex/sessions/**/rollout-*.jsonl, plaintext JSONL with
                 cwd, model, token usage, and turns. Parseable (session-search
                 routes here).
  - antigravity  IDE conversations are AEAD-encrypted at rest
                 (~/.gemini/antigravity/conversations/*.pb); the unencrypted CLI
                 store (~/.gemini/antigravity-cli/conversations/*.db) carries no
                 parseable turn/token content. No honest analysis path -> degrade.
"""
from __future__ import annotations

import json
import os
from pathlib import Path

CLAUDE = "claude"
CODEX = "codex"
ANTIGRAVITY = "antigravity"
_VALID = {CLAUDE, CODEX, ANTIGRAVITY}


def _from_stamp() -> str | None:
    # scripts/ -> skill root holds .techwolf-plugin.json (written by install.sh).
    here = Path(__file__).resolve()
    for d in (here.parent, here.parent.parent, here.parent.parent.parent):
        stamp = d / ".techwolf-plugin.json"
        if not stamp.is_file():
            continue
        try:
            platform = json.loads(stamp.read_text(encoding="utf-8")).get("platform")
        except (json.JSONDecodeError, OSError):
            return None
        if isinstance(platform, str) and platform.lower() in _VALID:
            return platform.lower()
        return None
    return None


def detect_platform() -> str:
    env = os.environ.get("AI_FIRST_PLATFORM", "").strip().lower()
    if env in _VALID:
        return env
    stamped = _from_stamp()
    if stamped:
        return stamped
    return CLAUDE


_DEGRADE = {
    CODEX: (
        "this analysis is not available on Codex yet.\n"
        "  Codex sessions (~/.codex/sessions) are parseable, but this skill does not\n"
        "  read them yet. Run it under Claude Code. (session-search already supports\n"
        "  Codex.)"
    ),
    ANTIGRAVITY: (
        "session analysis is not available on Antigravity.\n"
        "  Antigravity stores IDE conversations encrypted at rest\n"
        "  (~/.gemini/antigravity/conversations/*.pb, AEAD), and its unencrypted CLI\n"
        "  store carries no parseable turn/token content. There is no honest local\n"
        "  data path to analyse. Run this skill under Claude Code."
    ),
}


def degrade(skill: str, platform: str | None = None) -> None:
    """Print a clear, platform-specific 'not available' message and exit 0.

    Degrading is not an error: the skill simply isn't available on this host.
    """
    platform = platform or detect_platform()
    print(f"{skill}: {_DEGRADE.get(platform, f'not available on {platform}.')}")
    raise SystemExit(0)


def require_claude(skill: str) -> str:
    """Return the platform if Claude; otherwise degrade. For skills that only
    support Claude transcripts today (token-doctor, task-profile)."""
    platform = detect_platform()
    if platform == CLAUDE:
        return platform
    degrade(skill, platform)

#!/usr/bin/env python3
"""Print a single Claude Code / Cowork or Codex session as readable turns.

Takes the transcript path reported by find_sessions.py (a .jsonl file). Codex
rollouts (~/.codex/sessions/.../rollout-*.jsonl) are detected by path and read
through the Codex adapter.
See ../SKILL.md for when to use this.
"""
from __future__ import annotations

import argparse
import json
import re
import sys
from datetime import datetime
from pathlib import Path

sys.path.insert(0, str(Path(__file__).parent))
import codex_sessions  # noqa: E402


def _is_codex(path: Path) -> bool:
    parts = path.parts
    return ".codex" in parts and "sessions" in parts or path.name.startswith("rollout-")


def _extract_text(entry: dict) -> str:
    t = entry.get("type")
    if t not in ("user", "assistant"):
        return ""
    msg = entry.get("message") or {}
    content = msg.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":
                txt = c.get("text") or ""
                if txt:
                    parts.append(txt)
        return "\n".join(parts)
    return ""


def iter_turns(path: Path):
    with path.open(encoding="utf-8", errors="replace") as f:
        for line in f:
            try:
                e = json.loads(line)
            except json.JSONDecodeError:
                continue
            txt = _extract_text(e)
            if not txt.strip():
                continue
            ts = e.get("timestamp") or ""
            yield {"role": e.get("type"), "ts": ts, "text": txt}


def fmt_ts(ts: str) -> str:
    if not ts:
        return ""
    try:
        return datetime.fromisoformat(ts.replace("Z", "+00:00")).strftime("%Y-%m-%d %H:%M")
    except ValueError:
        return ts


def main() -> int:
    p = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    p.add_argument("path", help="transcript .jsonl path (from find_sessions.py)")
    p.add_argument("--grep", help="only print turns matching regex (plus --context surrounding turns)")
    p.add_argument("--context", type=int, default=1, help="surrounding turns when using --grep (default 1)")
    p.add_argument("--tail", type=int, help="print only the last N turns")
    p.add_argument("--head", type=int, help="print only the first N turns")
    p.add_argument("--max-chars", type=int, default=4000, help="truncate each turn to N chars (default 4000)")
    args = p.parse_args()

    path = Path(args.path)
    if not path.is_file():
        print(f"not found: {path}", file=sys.stderr)
        return 1

    if _is_codex(path):
        turns = [{"role": role, "ts": "", "text": txt}
                 for role, txt in codex_sessions.iter_turns(path) if txt.strip()]
    else:
        turns = list(iter_turns(path))

    if args.grep:
        pat = re.compile(args.grep, re.IGNORECASE)
        keep = set()
        for i, t in enumerate(turns):
            if pat.search(t["text"]):
                for j in range(max(0, i - args.context), min(len(turns), i + args.context + 1)):
                    keep.add(j)
        turns = [t for i, t in enumerate(turns) if i in keep]

    if args.head:
        turns = turns[: args.head]
    if args.tail:
        turns = turns[-args.tail :]

    if not turns:
        print("(no matching turns)")
        return 0

    print(f"# {path}")
    print(f"# {len(turns)} turns\n")
    for t in turns:
        header = f"## {t['role']}"
        ts = fmt_ts(t["ts"])
        if ts:
            header += f"  ·  {ts}"
        print(header)
        body = t["text"]
        if len(body) > args.max_chars:
            body = body[: args.max_chars] + f"\n… [truncated at {args.max_chars} chars]"
        print(body)
        print()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())