diff --git a/.github/workflows/opc-lab.yml b/.github/workflows/opc-lab.yml index a481a31..aa7de8a 100644 --- a/.github/workflows/opc-lab.yml +++ b/.github/workflows/opc-lab.yml @@ -19,6 +19,8 @@ jobs: run: ./scripts/verify.sh - name: Verify OPC fixtures run: python3 labs/opc-company/tests/run.py + - name: Verify Chat Export fixtures + run: python3 labs/chat-export/tests/run.py - name: Parse OPC JSON run: | find labs/opc-company -name '*.json' -print0 | while IFS= read -r -d '' file; do diff --git a/README.en.md b/README.en.md index 9799dca..12f5404 100644 --- a/README.en.md +++ b/README.en.md @@ -26,6 +26,8 @@ If you are exploring how multiple long-lived owners/workers can share canonical If your agents can already keep working but keep interrupting you for the wrong reasons, see the experimental [Attention Inbox Lab](labs/attention-inbox/README.md). It keeps routine completions and internal waiting states quiet, and surfaces only concrete human decisions or meaningful risks. It is a derived attention layer over existing task/receipt state, not another queue or scheduler. +If you use several AI tools and want each of them to find what you discussed in the others, see the experimental [Chat Export Lab](labs/chat-export/README.md). It turns local Claude Code, Codex, Cursor and Grok Bot chat history into one Markdown format: read-only, incremental, with common credentials redacted. Chat logs stay retrieval hints; files remain the source of truth. + ## Try it inside one isolated project ```bash diff --git a/README.md b/README.md index 28678c1..1349477 100644 --- a/README.md +++ b/README.md @@ -26,6 +26,8 @@ QuietHarness 不会给每个请求套一层仪式，也不是要求你复制我 如果你的 Agent 已经能持续工作，但经常因为错误的 `waiting_for_user`、常规完成或重复状态打断你，可以看实验性的 [Attention Inbox Lab](labs/attention-inbox/README.md)。它只把真正的人工决策和有意义的风险送到注意力入口；任务事实、调度和执行仍留在现有系统里，不另造一套队列或控制面。 +如果你同时用几家 AI，想让它们都能翻到之前在别家聊过什么，可以看实验性的 [Chat Export Lab](labs/chat-export/README.md)。它把 Claude Code、Codex、Cursor、Grok Bot 的本地聊天记录导成同一种 Markdown，只读、增量、导出前去掉常见凭证；聊天记录只当线索，事实仍以文件为准。 + ## 先在一个隔离项目试用 ```bash @@ -276,6 +278,7 @@ docs/ # 我的系统架构、任务、收尾与日报 examples/first-success/ # 单客户端隔离试用与行为验收 examples/leo-system/ # 脱敏的真实结构示例 labs/attention-inbox/ # 实验性的 Agent 注意力协议与策略 +labs/chat-export/ # 实验性的多 AI 聊天记录导出 scripts/ # 盘点、安装、验证 tests/ # 隔离 HOME 冒烟测试 media/launch-v3/ # 系统地图与发布素材 diff --git a/labs/chat-export/README.md b/labs/chat-export/README.md new file mode 100644 index 0000000..dcf4e06 --- /dev/null +++ b/labs/chat-export/README.md @@ -0,0 +1,73 @@ +# Chat Export Lab + +把 Claude Code、Codex、Cursor、Grok Bot 的本地聊天记录导出成 Markdown，一会话一个文件，任何 AI 都能用 grep 或全文检索查到。 + +> 实验性模块，不是 QuietHarness 的安装前提。只用 Python 3 标准库，只读本地文件，不联网、不登录、不改任何 App 状态。 + +## 解决什么问题 + +同时用好几家 AI 时，聊天记录各存各的：Claude Code 是 jsonl，Codex 是 rollout 文件加 SQLite，Cursor 是两个 SQLite 库，Grok Bot 是 App 里的二进制 blob。你在 Claude 里讨论过的东西，Codex 查不到；换一家 AI，上下文就断了。 + +这个模块只做一件事：把它们统一成同一种 Markdown，放进同一个目录。之后用什么检索都行：`rg`、qmd、Obsidian，或者让 Agent 直接读。 + +## 先试一次 + +```bash +cd labs/chat-export + +# 1. 只看会导出多少，不写任何文件 +python3 scripts/export_all.py --dry-run + +# 2. 真正导出到 ~/.local/share/chat-export/<来源>/ +python3 scripts/export_all.py + +# 3. 查一下 +rg -l "上次那个部署问题" ~/.local/share/chat-export +``` + +重复运行只处理有变化的会话（按文件修改时间和大小记水位），可以放进你自己的定时任务。`--full` 忽略水位全部重导。 + +## 支持的来源 + +| 来源 | 读取位置（默认） | 单独运行 | +|---|---|---| +| Claude Code | `~/.claude/projects/*/*.jsonl` | `scripts/export_claude.py` | +| Codex | `~/.codex/thread_history_*.sqlite`，以及 `~/.codex/sessions/**/rollout-*.jsonl` | `scripts/export_codex.py` | +| Cursor | App：`globalStorage/state.vscdb`；CLI：`~/.cursor/chats/*/*/store.db` | `scripts/export_cursor.py` | +| Grok Bot（macOS 桌面 App） | `~/Library/Application Support/Grok Bot/sand-client-persistence/*.blob` | `scripts/export_grok.py` | + +本机没有的来源会自动跳过。每个脚本都能用 `--help` 查看参数，路径都可以覆盖，`--dest` 改输出目录。 + +## 留什么、丢什么 + +留：你发的消息，和 AI 给你看的文字回复。 + +丢：系统提示和注入的上下文（``、整份 AGENTS.md、环境信息）、思考过程、工具调用与结果、子 Agent 的旁支对话。不丢的话，每个会话都带一份 AGENTS.md，任何关键词一搜就命中全部会话。 + +导出前会把常见凭证替换成 ``：私钥、`sk-` 开头的 API key、GitHub / AWS / Slack token、Bearer 头、`*_TOKEN=` 这类赋值、超长 base64。这是尽力而为的规则匹配，**不能保证把所有敏感内容都清掉**。导出的目录和文件都只有你本人能读写（0700 / 0600），只适合留在本机，不要提交到仓库，也不要上传。 + +## 踩过的坑 + +- **Codex 新版一轮对话会写好几条回复**：先是「我先看看」这类进度，最后才是结论。只取第一条回复，所有结论都会丢。这里保留一轮里的全部回复。新版 Codex 自己的 `thread_history` 库直接记了每轮的提问和最终回答，存在时优先用它；即使 rollout 文件已被清理，结论也还在。 +- **同一个 Codex 会话会跨多个 rollout 文件**（继续、恢复对话时）。按会话 id 命名会互相覆盖，所以按 rollout 自己的 id 命名。 +- **Grok Bot 的 blob 里，你的消息和它的回复是两种结构**。只解析一种，就只看到半边对话。 +- **超过 6 万字符的一轮直接跳过**，通常是贴进去的日志或整份文件，切一半反而误导检索。 + +## 它不是什么 + +聊天记录只是线索，不是事实。决定、进展、下一步还是要写回任务卡、文档或仓库才算数；聊天记录和文件对不上时，以文件为准。这个模块让各家 AI 都能翻到「之前聊过什么」，但不替代这一层。 + +它也不是同步服务、记忆数据库或向量库：没有后台进程，不做语义合并，不往任何 AI 里写东西。 + +## 测试 + +```bash +python3 labs/chat-export/tests/run.py +# CHAT_EXPORT_TEST_OK +``` + +测试只用临时生成的假数据，不读你的真实聊天记录。 + +## English summary + +Exports local Claude Code, Codex, Cursor and Grok Bot (macOS app) chat history to one Markdown file per conversation under `~/.local/share/chat-export/`, so any agent can search past conversations with `rg` or a full-text index. Stdlib-only, read-only, incremental, with best-effort credential redaction. Run `python3 scripts/export_all.py --dry-run` first. Chat logs are retrieval hints, not the source of truth. diff --git a/labs/chat-export/scripts/common.py b/labs/chat-export/scripts/common.py new file mode 100644 index 0000000..e9e68e5 --- /dev/null +++ b/labs/chat-export/scripts/common.py @@ -0,0 +1,108 @@ +"""Shared pieces for the chat exporters: turn grouping, redaction, markdown rendering, watermark.""" +import json +import os +import re +from pathlib import Path + +# Exports are private conversations: every directory and file this process creates is owner-only. +os.umask(0o077) + +DEST_ROOT = Path.home() / ".local/share/chat-export" + +# A turn this large is almost always a pasted log or file; skip it rather than cut it mid-thought. +MAX_TURN_CHARS = 60_000 + +# Credentials are redacted before anything touches disk. Most specific patterns first. +REDACTIONS = [ + (re.compile(r"0x[a-fA-F0-9]{64}\b"), ""), + (re.compile(r"\bsk-[A-Za-z0-9_\-]{20,}"), ""), + (re.compile(r"\bghp_[A-Za-z0-9]{20,}"), ""), + (re.compile(r"\bgithub_pat_[A-Za-z0-9_]{20,}"), ""), + (re.compile(r"\bAKIA[0-9A-Z]{16}\b"), ""), + (re.compile(r"\bxox[baprs]-[A-Za-z0-9\-]{10,}"), ""), + (re.compile(r"(?i)\b(bearer)\s+[A-Za-z0-9._\-]{20,}"), r"\1 "), + (re.compile(r"(?i)\b([A-Z0-9_]*(?:SECRET|PASSWORD|PRIVATE_KEY|API_KEY|TOKEN)[A-Z0-9_]*)\s*=\s*\S+"), + r"\1="), + (re.compile(r"[A-Za-z0-9+/]{300,}={0,2}"), ""), +] + + +def redact(text): + for pattern, repl in REDACTIONS: + text = pattern.sub(repl, text) + return text + + +def group_turns(messages): + """messages: [(role, text)] in order. A user text opens a turn; every reply until the next user text + belongs to it. Keeping all replies matters: newer agents stream progress notes before the final answer.""" + turns, current = [], None + for role, text in messages: + text = (text or "").strip() + if not text: + continue + if role == "user": + if current and current["assistant"]: + turns.append(current) + current = {"user": text, "assistant": []} + elif role == "assistant" and current is not None: + current["assistant"].append(text) + if current and current["assistant"]: + turns.append(current) + return turns + + +def render(source, session_id, turns, *, title="", timestamp="", cwd="", source_file=""): + lines = [ + "---", + f"session_id: {session_id}", + f"source: {source}", + f"title: {json.dumps(title or '', ensure_ascii=False)}", + f"timestamp: {timestamp or ''}", + f"cwd: {cwd or ''}", + f"turns: {len(turns)}", + f"source_file: {source_file}", + "---", + "", + f"# {source} {title or str(session_id)[:8]}", + "", + ] + for i, turn in enumerate(turns, 1): + assistant = "\n\n".join(turn["assistant"]) + if len(turn["user"]) + len(assistant) > MAX_TURN_CHARS: + lines.append(f"## Turn {i} (skipped: exceeds {MAX_TURN_CHARS} chars)\n") + continue + lines += [f"## Turn {i}", "", "**User:**", "", redact(turn["user"]), "", + "**Assistant:**", "", redact(assistant), ""] + return "\n".join(lines) + + +def file_sig(*paths): + """mtime+size of each existing path: cheap change detection for the watermark.""" + return ":".join(f"{p.stat().st_mtime_ns}/{p.stat().st_size}" for p in paths if p.exists()) + + +class Watermark: + """Per-source {key: signature} file so reruns only touch what changed.""" + + def __init__(self, dest, full=False): + self.path = dest / ".watermark.json" + try: + self.state = {} if full else json.loads(self.path.read_text()) + except (OSError, json.JSONDecodeError): + self.state = {} + + def unchanged(self, key, sig): + return self.state.get(key) == sig + + def mark(self, key, sig): + self.state[key] = sig + + def save(self): + self.path.parent.mkdir(parents=True, exist_ok=True) + self.path.write_text(json.dumps(self.state, indent=0)) + # Tighten directories left world-readable by an earlier run; new ones already get 0700 from the umask. + # Only the export dir itself and the default root; never a parent the user pointed --dest into. + self.path.parent.chmod(0o700) + if self.path.parent.parent == DEST_ROOT: + DEST_ROOT.chmod(0o700) diff --git a/labs/chat-export/scripts/export_all.py b/labs/chat-export/scripts/export_all.py new file mode 100644 index 0000000..b76b31b --- /dev/null +++ b/labs/chat-export/scripts/export_all.py @@ -0,0 +1,20 @@ +#!/usr/bin/env python3 +"""Run every exporter whose source exists on this machine. Pass --dry-run to only report.""" +import importlib +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) + +EXPORTERS = ["export_claude", "export_codex", "export_cursor", "export_grok"] + + +def main(): + passthrough = [a for a in sys.argv[1:] if a in ("--dry-run", "--full")] + for name in EXPORTERS: + importlib.import_module(name).main(passthrough) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/labs/chat-export/scripts/export_claude.py b/labs/chat-export/scripts/export_claude.py new file mode 100644 index 0000000..860f462 --- /dev/null +++ b/labs/chat-export/scripts/export_claude.py @@ -0,0 +1,96 @@ +#!/usr/bin/env python3 +"""Export Claude Code sessions (~/.claude/projects/*/.jsonl) to one markdown file per session. + +Kept: what you typed and Claude's visible text replies. +Dropped: system reminders, thinking, tool calls/results, meta records and subagent sidechains. +""" +import argparse +import json +import re +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from common import DEST_ROOT, Watermark, file_sig, group_turns, render # noqa: E402 + +SRC = Path.home() / ".claude/projects" +DEST = DEST_ROOT / "claude" + +INJECTED = [re.compile(rf"<{tag}>.*?", re.S) for tag in ( + "system-reminder", "local-command-stdout", "local-command-caveat", "task-notification")] + + +def strip_injected(text): + for pattern in INJECTED: + text = pattern.sub("", text) + return text.strip() + + +def text_blocks(content): + if isinstance(content, str): + return content + if not isinstance(content, list): + return "" + return "\n".join(b["text"] for b in content + if isinstance(b, dict) and b.get("type") == "text" and isinstance(b.get("text"), str)) + + +def parse_session(path): + meta, messages = {}, [] + for line in path.open(encoding="utf-8", errors="replace"): + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue + rtype = rec.get("type") + if rtype == "custom-title" and isinstance(rec.get("customTitle"), str): + meta["title"] = rec["customTitle"] + continue + if rtype not in ("user", "assistant") or rec.get("isSidechain") or rec.get("isMeta"): + continue + if not meta.get("session_id"): + meta.update(session_id=rec.get("sessionId"), cwd=rec.get("cwd"), timestamp=rec.get("timestamp")) + text = text_blocks((rec.get("message") or {}).get("content")) + if rtype == "user": + text = strip_injected(text) # empty afterwards = tool result or pure injection + messages.append((rtype, text)) + return meta, group_turns(messages) + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--src", type=Path, default=SRC) + ap.add_argument("--dest", type=Path, default=DEST) + ap.add_argument("--full", action="store_true", help="ignore the watermark and reprocess everything") + ap.add_argument("--dry-run", action="store_true", help="report only, write nothing") + args = ap.parse_args(argv) + + mark = Watermark(args.dest, args.full) + files = sorted((p for p in args.src.glob("*/*.jsonl") if "subagents" not in p.parts), + key=lambda p: p.stat().st_mtime_ns) # a session copied across project dirs: newest wins + written = empty = 0 + for f in files: + sig = file_sig(f) + if mark.unchanged(str(f), sig): + continue + meta, turns = parse_session(f) + if turns: + if not args.dry_run: + args.dest.mkdir(parents=True, exist_ok=True) + date = (meta.get("timestamp") or "")[:10] or "undated" + (args.dest / f"{date}-{f.stem}.md").write_text(render( + "claude", meta.get("session_id") or f.stem, turns, title=meta.get("title", ""), + timestamp=meta.get("timestamp") or "", cwd=meta.get("cwd") or "", source_file=str(f)), encoding="utf-8") + written += 1 + else: + empty += 1 + mark.mark(str(f), sig) + if not args.dry_run: + mark.save() + print(f"claude: {written} written, {empty} empty, {len(files)} sessions seen" + + (" (dry run)" if args.dry_run else f" -> {args.dest}")) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/labs/chat-export/scripts/export_codex.py b/labs/chat-export/scripts/export_codex.py new file mode 100644 index 0000000..6c8445b --- /dev/null +++ b/labs/chat-export/scripts/export_codex.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +"""Export Codex sessions to one markdown file per conversation. + +Two sources, read-only: + 1. Codex's thread-history SQLite (newer Codex). It stores each turn's prompt and FINAL answer, + even after rollout files are pruned. Preferred when present. + 2. rollout-*.jsonl under ~/.codex/sessions, for sessions the history db does not cover. + +The trap this handles: newer Codex writes several assistant messages per turn (progress notes, +then the answer). An exporter that keeps only the first reply saves "let me look" and loses +every conclusion. +""" +import argparse +import datetime as dt +import hashlib +import json +import re +import sqlite3 +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from common import DEST_ROOT, Watermark, file_sig, group_turns, render # noqa: E402 + +CODEX_HOME = Path.home() / ".codex" +DEST = DEST_ROOT / "codex" + +# Codex folds plugin lists, the whole AGENTS.md and environment context into the first user +# message. Left in, every session carries a copy of AGENTS.md and floods keyword search. +BOILERPLATE = [re.compile(rf"<{tag}>.*?", re.S) for tag in ( + "recommended_plugins", "INSTRUCTIONS", "environment_context", "user_instructions")] +BOILERPLATE.append(re.compile(r"^#\s*AGENTS\.md instructions for .*$", re.M)) + + +def strip_boilerplate(text): + for pattern in BOILERPLATE: + text = pattern.sub("", text) + return text.strip() + + +def content_text(content): + if isinstance(content, str): + return content + if not isinstance(content, list): + return "" + return "\n".join(p["text"] for p in content if isinstance(p, dict) and isinstance(p.get("text"), str)) + + +def parse_rollout(path): + meta, messages = {}, [] + for line in path.open(encoding="utf-8", errors="replace"): + try: + rec = json.loads(line) + except json.JSONDecodeError: + continue + payload = rec.get("payload") + if not isinstance(payload, dict): + continue + if rec.get("type") == "session_meta" and not meta: + meta = {"session_id": payload.get("session_id") or payload.get("id"), "cwd": payload.get("cwd"), + "timestamp": payload.get("timestamp") or rec.get("timestamp")} + elif rec.get("type") == "response_item" and payload.get("type") == "message": + role = payload.get("role") # "developer" is the injected system prompt, not conversation + if role in ("user", "assistant"): + text = content_text(payload.get("content")) + messages.append((role, strip_boilerplate(text) if role == "user" else text)) + return meta, group_turns(messages) + + +def newest(pattern): + """Codex bumps the db file number on schema changes (state_5.sqlite ...): take the highest.""" + def version(p): + m = re.search(r"_(\d+)\.sqlite$", p.name) + return int(m.group(1)) if m else -1 + found = sorted(CODEX_HOME.glob(pattern), key=version) + return found[-1] if found else None + + +def ro(path): + return sqlite3.connect(f"file:{path}?mode=ro", uri=True) + + +def item_text(item_json): + try: + item = json.loads(item_json) + except (TypeError, json.JSONDecodeError): + return "" + return item["text"] if isinstance(item.get("text"), str) else content_text(item.get("content")) + + +def export_threads(dest, mark, dry_run, thread_db, state_db): + """Return the set of thread ids the history db covers, and how many files were written.""" + if not thread_db or not thread_db.exists(): + return set(), 0 + db = ro(thread_db) + # Cheap per-thread signature first; item bodies are read only for threads that changed. + sigs = {tid: f"{n}:{last}:{fin}" for tid, n, last, fin in db.execute( + "SELECT thread_id, count(*), max(rollout_ordinal), max(final_agent_item_id) FROM thread_turns " + "WHERE first_user_item_id IS NOT NULL AND final_agent_item_id IS NOT NULL GROUP BY thread_id")} + changed = [tid for tid, sig in sigs.items() if not mark.unchanged(f"thread:{tid}", sig)] + rows = [] + for i in range(0, len(changed), 500): + chunk = changed[i:i + 500] + rows += db.execute( + "SELECT t.thread_id, t.started_at, u.item_json, a.item_json FROM thread_turns t " + "JOIN thread_items u ON u.thread_id = t.thread_id AND u.item_id = t.first_user_item_id " + "JOIN thread_items a ON a.thread_id = t.thread_id AND a.item_id = t.final_agent_item_id " + f"WHERE t.thread_id IN ({','.join('?' * len(chunk))}) ORDER BY t.thread_id, t.rollout_ordinal", + chunk).fetchall() + info, children = {}, set() + if state_db and state_db.exists(): + sdb = ro(state_db) + info = {r[0]: {"title": r[1], "cwd": r[2], "created_at": r[3]} + for r in sdb.execute("SELECT id, title, cwd, created_at FROM threads")} + children = {r[0] for r in sdb.execute("SELECT child_thread_id FROM thread_spawn_edges")} + by_thread = {} + for tid, started, user_json, final_json in rows: + if tid in children: + continue # subagent threads: the parent's final answer already carries their result + user, final = strip_boilerplate(item_text(user_json)), item_text(final_json).strip() + if user and final: + by_thread.setdefault(tid, []).append({"user": user, "assistant": [final], "started": started or 0}) + written = 0 + for tid, turns in by_thread.items(): + meta = info.get(tid, {}) + created = meta.get("created_at") or turns[0]["started"] + if created and created > 1e11: + created /= 1000 # milliseconds + date = dt.datetime.fromtimestamp(created).strftime("%Y-%m-%d") if created else "undated" + if not dry_run: + (dest / "threads").mkdir(parents=True, exist_ok=True) + (dest / "threads" / f"{date}-{tid}.md").write_text(render( + "codex-thread", tid, turns, title=(meta.get("title") or "").replace("\n", " "), + timestamp=date, cwd=meta.get("cwd") or "", source_file=f"{thread_db}#{tid}"), encoding="utf-8") + written += 1 + for tid in changed: + mark.mark(f"thread:{tid}", sigs[tid]) + return set(sigs) - children, written + + +def rendered_source(path): + try: + m = re.search(r"^source_file: (.+)$", path.read_text(encoding="utf-8", errors="replace")[:2000], re.M) + except OSError: + return None + return m.group(1).strip() if m else None + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--src", type=Path, default=CODEX_HOME / "sessions", help="rollout directory") + ap.add_argument("--thread-db", type=Path, default=None, help="default: newest ~/.codex/thread_history_*.sqlite") + ap.add_argument("--state-db", type=Path, default=None, help="default: newest ~/.codex/state_*.sqlite") + ap.add_argument("--no-threads", action="store_true", help="only export rollout files") + ap.add_argument("--dest", type=Path, default=DEST) + ap.add_argument("--full", action="store_true", help="ignore the watermark and reprocess everything") + ap.add_argument("--dry-run", action="store_true", help="report only, write nothing") + args = ap.parse_args(argv) + + mark = Watermark(args.dest, args.full) + covered, thread_written = set(), 0 + if not args.no_threads: + covered, thread_written = export_threads( + args.dest, mark, args.dry_run, + args.thread_db or newest("thread_history_*.sqlite"), args.state_db or newest("state_*.sqlite")) + + files = sorted(args.src.rglob("rollout-*.jsonl")) + written = skipped = 0 + for f in files: + sig = file_sig(f) + if mark.unchanged(str(f), sig): + continue + meta, turns = parse_rollout(f) + if turns and meta.get("session_id") not in covered: + if not args.dry_run: + args.dest.mkdir(parents=True, exist_ok=True) + # One session id spans many rollout files (resume/continue): name by the rollout's own id, + # or siblings silently overwrite each other. + rollout_id = f.stem.split("-")[-1][:8] + date = (meta.get("timestamp") or "")[:10] or "undated" + out = args.dest / f"{date}-{(meta.get('session_id') or 'nosid')[:8]}-{rollout_id}.md" + if out.exists() and rendered_source(out) != str(f): + out = out.with_name(out.stem + "-" + hashlib.sha1(str(f).encode()).hexdigest()[:6] + ".md") + out.write_text(render("codex", meta.get("session_id") or f.stem, turns, + timestamp=meta.get("timestamp") or "", cwd=meta.get("cwd") or "", + source_file=str(f)), encoding="utf-8") + written += 1 + else: + skipped += 1 + mark.mark(str(f), sig) + if not args.dry_run: + mark.save() + print(f"codex: {thread_written} threads from history db, {written} rollouts written, " + f"{skipped} skipped (empty or covered), {len(files)} rollouts seen" + + (" (dry run)" if args.dry_run else f" -> {args.dest}")) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/labs/chat-export/scripts/export_cursor.py b/labs/chat-export/scripts/export_cursor.py new file mode 100644 index 0000000..0aa90a1 --- /dev/null +++ b/labs/chat-export/scripts/export_cursor.py @@ -0,0 +1,162 @@ +#!/usr/bin/env python3 +"""Export Cursor conversations (app composer + cursor-agent CLI) to one markdown file per conversation. + +Two local stores, both opened read-only: + app: /globalStorage/state.vscdb. composerData: lists bubble ids in order; + bubbleId:: holds type 1 (user) / type 2 (assistant) text. + cli: ~/.cursor/chats///store.db. Content-addressed blobs; the root blob named in + meta lists message hashes in order. +""" +import argparse +import datetime as dt +import json +import platform +import re +import sqlite3 +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from common import DEST_ROOT, Watermark, file_sig, group_turns, render # noqa: E402 + +if platform.system() == "Darwin": + APP_DB = Path.home() / "Library/Application Support/Cursor/User/globalStorage/state.vscdb" +else: + APP_DB = Path.home() / ".config/Cursor/User/globalStorage/state.vscdb" +CLI_ROOT = Path.home() / ".cursor/chats" +DEST = DEST_ROOT / "cursor" + +QUERY = re.compile(r"\s*(.*?)\s*", re.S) +INJECTED = re.compile(r"<(user_info|timestamp|system_reminder|system_notification|attached_files|rules)>.*?", re.S) + + +def ro(path): + return sqlite3.connect(f"file:{path}?mode=ro", uri=True) + + +def clean_user(text): + m = QUERY.search(text) + return (m.group(1) if m else INJECTED.sub("", text)).strip() + + +def content_text(content): + if isinstance(content, str): + return content + if not isinstance(content, list): + return "" + return "\n".join(p["text"] for p in content + if isinstance(p, dict) and p.get("type") == "text" and isinstance(p.get("text"), str)) + + +def iso(ms): + return dt.datetime.fromtimestamp(ms / 1000).isoformat(timespec="seconds") if ms else "" + + +def cli_conversation(store): + db = ro(store) + meta = json.loads(bytes.fromhex(db.execute("SELECT value FROM meta WHERE key='0'").fetchone()[0])) + blobs = dict(db.execute("SELECT id, data FROM blobs")) + root = bytes(blobs.get(meta.get("latestRootBlobId"), b"")) + ordered = sorted((pos, bid) for bid in blobs if (pos := root.find(bytes.fromhex(bid))) >= 0) + messages = [] + for _, bid in ordered: + data = bytes(blobs[bid]) + if not data.startswith(b"{"): + continue + try: + msg = json.loads(data) + except json.JSONDecodeError: + continue + role, text = msg.get("role"), content_text(msg.get("content")) + if role in ("user", "assistant"): + messages.append((role, clean_user(text) if role == "user" else text)) + return meta, group_turns(messages) + + +def export_cli(cli_root, dest, mark, dry_run): + written = 0 + for store in sorted(cli_root.glob("*/*/store.db")): + sig, key = file_sig(store, store.with_name("store.db-wal")), f"cli:{store}" + if mark.unchanged(key, sig): + continue + try: + meta, turns = cli_conversation(store) + except (sqlite3.Error, ValueError, TypeError, KeyError) as exc: + print(f" cli read error {store.parent.name}: {type(exc).__name__}", file=sys.stderr) + continue + if turns: + try: + cwd = json.loads((store.parent / "meta.json").read_text()).get("cwd", "") + except (OSError, json.JSONDecodeError): + cwd = "" + name = meta.get("name") if meta.get("name") not in (None, "New Agent") else "" + when = iso(meta.get("createdAt")) + if not dry_run: + (dest / "cli").mkdir(parents=True, exist_ok=True) + (dest / "cli" / f"{when[:10] or 'undated'}-{store.parent.name}.md").write_text(render( + "cursor-cli", store.parent.name, turns, title=name or "", timestamp=when, cwd=cwd, + source_file=str(store)), encoding="utf-8") + written += 1 + mark.mark(key, sig) + return written + + +def export_app(app_db, dest, mark, dry_run): + if not app_db.exists(): + return 0 + db = ro(app_db) + written = 0 + for key, value in db.execute("SELECT key, value FROM cursorDiskKV WHERE key LIKE 'composerData:%'"): + try: + data = json.loads(value) + except (TypeError, json.JSONDecodeError): + continue + headers = data.get("fullConversationHeadersOnly") or [] + cid = key.split(":", 1)[1] + sig = f"{len(headers)}:{data.get('lastUpdatedAt') or ''}" + if not headers or mark.unchanged(f"app:{cid}", sig): + continue + messages = [] + for h in headers: + row = db.execute("SELECT value FROM cursorDiskKV WHERE key=?", + (f"bubbleId:{cid}:{h.get('bubbleId')}",)).fetchone() + if not row: + continue + bubble = json.loads(row[0]) + role = {1: "user", 2: "assistant"}.get(bubble.get("type")) + if role: + text = bubble.get("text") or "" + messages.append((role, clean_user(text) if role == "user" else text)) + turns = group_turns(messages) + if turns: + when = iso(data.get("createdAt")) + if not dry_run: + (dest / "app").mkdir(parents=True, exist_ok=True) + (dest / "app" / f"{when[:10] or 'undated'}-{cid}.md").write_text(render( + "cursor-app", cid, turns, title=data.get("name") or "", timestamp=when, + source_file=f"{app_db}#{cid}"), encoding="utf-8") + written += 1 + mark.mark(f"app:{cid}", sig) + return written + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--app-db", type=Path, default=APP_DB) + ap.add_argument("--cli-root", type=Path, default=CLI_ROOT) + ap.add_argument("--dest", type=Path, default=DEST) + ap.add_argument("--full", action="store_true", help="ignore the watermark and reprocess everything") + ap.add_argument("--dry-run", action="store_true", help="report only, write nothing") + args = ap.parse_args(argv) + mark = Watermark(args.dest, args.full) + cli = export_cli(args.cli_root, args.dest, mark, args.dry_run) + app = export_app(args.app_db, args.dest, mark, args.dry_run) + if not args.dry_run: + mark.save() + print(f"cursor: {app} app + {cli} cli conversations written" + + (" (dry run)" if args.dry_run else f" -> {args.dest}")) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/labs/chat-export/scripts/export_grok.py b/labs/chat-export/scripts/export_grok.py new file mode 100644 index 0000000..1b0914f --- /dev/null +++ b/labs/chat-export/scripts/export_grok.py @@ -0,0 +1,117 @@ +#!/usr/bin/env python3 +"""Export Grok Bot desktop-app conversations (macOS) to one markdown file per conversation. + +Grok Bot has no export API, but the desktop app keeps a plaintext replica of each conversation in +~/Library/Application Support/Grok Bot/sand-client-persistence/.blob. The slice named +`...transcript.replicas.` is the conversation. Read-only; never touches the app. + +Scope: only what this machine's app replica holds (not grok.com web chats). + +Each blob has a binary header followed by JSON objects. Both kinds are needed, or you only see +half the conversation: + {"kind":"message", "role":"user", "content":"..."} your messages + {"kind":"send-message", "message":{"type":"text","content":"..."}} Grok's replies +""" +import argparse +import base64 +import datetime as dt +import json +import re +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent)) +from common import DEST_ROOT, Watermark, file_sig, group_turns, render # noqa: E402 + +BLOB_DIR = Path.home() / "Library/Application Support/Grok Bot/sand-client-persistence" +DEST = DEST_ROOT / "grok" +MARKER = ".transcript.replicas." +START = re.compile(r'\{"kind":\s*"(?:message|send-message)"') + + +def decode_name(path): + name = path.name[:-5] if path.name.endswith(".blob") else path.name + try: + return base64.b32decode(name.upper() + "=" * (-len(name) % 8)).decode("utf-8", "replace") + except ValueError: + return "" + + +def balanced_end(raw, start): + """Index of the brace closing the JSON object that opens at `start`, string-aware.""" + depth, in_str, esc = 0, False, False + for i in range(start, len(raw)): + c = raw[i] + if in_str: + if esc: + esc = False + elif c == "\\": + esc = True + elif c == '"': + in_str = False + elif c == '"': + in_str = True + elif c == "{": + depth += 1 + elif c == "}": + depth -= 1 + if depth == 0: + return i + return len(raw) - 1 + + +def extract_messages(path): + raw = path.read_bytes().decode("utf-8", "replace") + msgs = [] + for m in START.finditer(raw): + try: + obj = json.loads(raw[m.start():balanced_end(raw, m.start()) + 1]) + except json.JSONDecodeError: + continue + if obj.get("kind") == "send-message": + role, body = "assistant", (obj.get("message") or {}).get("content") + else: + role, body = obj.get("role"), obj.get("content") + if isinstance(body, str) and body.strip(): + msgs.append({"role": role, "content": body, "ts": obj.get("timestampMs")}) + msgs.sort(key=lambda x: x["ts"] or 0) + return msgs + + +def main(argv=None): + ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + ap.add_argument("--blob-dir", type=Path, default=BLOB_DIR) + ap.add_argument("--dest", type=Path, default=DEST) + ap.add_argument("--full", action="store_true", help="ignore the watermark and reprocess everything") + ap.add_argument("--dry-run", action="store_true", help="report only, write nothing") + args = ap.parse_args(argv) + + mark = Watermark(args.dest, args.full) + blobs = [(b, decode_name(b)) for b in sorted(args.blob_dir.glob("*.blob"))] + blobs = [(b, name) for b, name in blobs if MARKER in name] + written = 0 + for blob, name in blobs: + sig = file_sig(blob) + if mark.unchanged(str(blob), sig): + continue + messages = extract_messages(blob) + turns = group_turns([(m["role"], m["content"]) for m in messages]) + if turns: + conv_id = name.rsplit(MARKER, 1)[1] + first = next((m["ts"] for m in messages if m["ts"]), None) + when = dt.datetime.fromtimestamp(first / 1000).isoformat(timespec="seconds") if first else "" + if not args.dry_run: + args.dest.mkdir(parents=True, exist_ok=True) + (args.dest / f"{when[:10] or 'undated'}-{conv_id}.md").write_text(render( + "grok-bot", conv_id, turns, timestamp=when, source_file=str(blob)), encoding="utf-8") + written += 1 + mark.mark(str(blob), sig) + if not args.dry_run: + mark.save() + print(f"grok: {written} written, {len(blobs)} conversations seen" + + (" (dry run)" if args.dry_run else f" -> {args.dest}")) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/labs/chat-export/tests/run.py b/labs/chat-export/tests/run.py new file mode 100644 index 0000000..45e5475 --- /dev/null +++ b/labs/chat-export/tests/run.py @@ -0,0 +1,136 @@ +#!/usr/bin/env python3 +"""Synthetic-fixture tests for the chat exporters. Python 3 stdlib only; never reads your real chats.""" +import base64 +import contextlib +import io +import json +import sqlite3 +import sys +import tempfile +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "scripts")) +import export_claude # noqa: E402 +import export_codex # noqa: E402 +import export_cursor # noqa: E402 +import export_grok # noqa: E402 + +KEY = "sk-" + "a" * 30 + + +def expect(condition, message): + if not condition: + raise AssertionError(message) + + +def run(module, argv): + with contextlib.redirect_stdout(io.StringIO()): + expect(module.main(argv) == 0, f"{module.__name__} exited non-zero") + + +def only_md(dest): + files = list(dest.rglob("*.md")) + expect(len(files) == 1, f"expected one markdown file in {dest}, got {len(files)}") + return files[0].read_text() + + +def jsonl(path, records): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text("\n".join(json.dumps(r, ensure_ascii=False) for r in records) + "\n") + + +def test_claude(tmp): + jsonl(tmp / "src/proj/s1.jsonl", [ + {"type": "user", "sessionId": "s1", "timestamp": "2026-10-01T00:00:00Z", "cwd": "/w", + "message": {"content": "injectedHow do I ship?"}}, + {"type": "assistant", "message": {"content": [{"type": "thinking", "thinking": "hidden"}, + {"type": "text", "text": f"Use the key {KEY}"}]}}, + {"type": "user", "message": {"content": [{"type": "tool_result", "content": "tool noise"}]}}, + {"type": "assistant", "isSidechain": True, "message": {"content": [{"type": "text", "text": "subagent"}]}}, + {"type": "assistant", "message": {"content": [{"type": "text", "text": "Then run tests."}]}}, + ]) + run(export_claude, ["--src", str(tmp / "src"), "--dest", str(tmp / "out")]) + text = only_md(tmp / "out") + expect("How do I ship?" in text and "Then run tests." in text, "claude: user prompt and both replies kept") + for dropped in ("injected", "hidden", "tool noise", "subagent", KEY): + expect(dropped not in text, f"claude: {dropped!r} should not be exported") + # Watermark: an unchanged rerun writes nothing. + out = io.StringIO() + with contextlib.redirect_stdout(out): + export_claude.main(["--src", str(tmp / "src"), "--dest", str(tmp / "out")]) + expect("0 written" in out.getvalue(), "claude: unchanged session must not be rewritten") + md = next((tmp / "out").glob("*.md")) + expect(md.stat().st_mode & 0o077 == 0, "claude: exported file must be owner-only") + expect((tmp / "out").stat().st_mode & 0o077 == 0, "claude: export dir must be owner-only") + + +def test_codex_rollout_keeps_final_answer(tmp): + msg = lambda role, text: {"type": "response_item", "payload": {"type": "message", "role": role, # noqa: E731 + "content": [{"type": "text", "text": text}]}} + jsonl(tmp / "sessions/rollout-2026-10-01-abcdef12.jsonl", [ + {"type": "session_meta", "payload": {"id": "sess-1", "cwd": "/w", "timestamp": "2026-10-01T00:00:00Z"}}, + msg("developer", "system prompt"), + msg("user", "envFix the bug"), + msg("assistant", "Let me look first."), + msg("assistant", "Final: the bug was an off-by-one."), + ]) + run(export_codex, ["--src", str(tmp / "sessions"), "--no-threads", "--dest", str(tmp / "out")]) + text = only_md(tmp / "out") + expect("Final: the bug was an off-by-one." in text, "codex: final answer after a progress note must survive") + expect("system prompt" not in text and "env" not in text, "codex: scaffolding stripped") + + +def test_codex_thread_history(tmp): + db = sqlite3.connect(tmp / "thread_history_1.sqlite") + db.executescript( + "CREATE TABLE thread_turns(thread_id, rollout_ordinal, started_at, first_user_item_id, final_agent_item_id);" + "CREATE TABLE thread_items(thread_id, item_id, item_json);") + db.execute("INSERT INTO thread_turns VALUES('t1', 1, 1790000000, 'u1', 'a1')") + db.execute("INSERT INTO thread_items VALUES('t1', 'u1', ?)", (json.dumps({"text": "Plan the release"}),)) + db.execute("INSERT INTO thread_items VALUES('t1', 'a1', ?)", (json.dumps({"text": "Release plan: three steps"}),)) + db.commit() + run(export_codex, ["--src", str(tmp / "none"), "--thread-db", str(tmp / "thread_history_1.sqlite"), + "--state-db", str(tmp / "missing.sqlite"), "--dest", str(tmp / "out")]) + text = only_md(tmp / "out") + expect("Plan the release" in text and "Release plan: three steps" in text, "codex: thread history exported") + + +def test_cursor_app(tmp): + db = sqlite3.connect(tmp / "state.vscdb") + db.execute("CREATE TABLE cursorDiskKV(key, value)") + db.execute("INSERT INTO cursorDiskKV VALUES('composerData:c1', ?)", (json.dumps({ + "name": "Refactor", "createdAt": 1790000000000, "lastUpdatedAt": 1, + "fullConversationHeadersOnly": [{"bubbleId": "b1"}, {"bubbleId": "b2"}]}),)) + db.execute("INSERT INTO cursorDiskKV VALUES('bubbleId:c1:b1', ?)", (json.dumps( + {"type": 1, "text": "osRename the module"}),)) + db.execute("INSERT INTO cursorDiskKV VALUES('bubbleId:c1:b2', ?)", (json.dumps({"type": 2, "text": "Renamed."}),)) + db.commit() + run(export_cursor, ["--app-db", str(tmp / "state.vscdb"), "--cli-root", str(tmp / "none"), + "--dest", str(tmp / "out")]) + text = only_md(tmp / "out") + expect("Rename the module" in text and "Renamed." in text and "user_info" not in text, "cursor: app chat exported") + + +def test_grok(tmp): + name = "sand.client.slice.account.x.transcript.replicas.conv-1234abcd" + blob = tmp / "blobs" / (base64.b32encode(name.encode()).decode().rstrip("=").lower() + ".blob") + blob.parent.mkdir(parents=True) + body = (json.dumps({"kind": "message", "role": "user", "content": "Question", "timestampMs": 1790000000000}) + + json.dumps({"kind": "send-message", "message": {"type": "text", "content": f"Answer {KEY}"}, + "timestampMs": 1790000001000})) + blob.write_bytes(b"\x00\x01header" + body.encode()) + run(export_grok, ["--blob-dir", str(tmp / "blobs"), "--dest", str(tmp / "out")]) + text = only_md(tmp / "out") + expect("Question" in text and "Answer" in text and KEY not in text, "grok: both sides exported, key redacted") + + +def main(): + for test in (test_claude, test_codex_rollout_keeps_final_answer, test_codex_thread_history, + test_cursor_app, test_grok): + with tempfile.TemporaryDirectory() as tmp: + test(Path(tmp)) + print("CHAT_EXPORT_TEST_OK") + + +if __name__ == "__main__": + main()