codex_transfer.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""codex_transfer.py — Codex CLI 会话迁移工具

功能：
  1. 根据 `codex resume <session-id>` 命令串 / session id（或前缀）/ 文件路径，
     在 ~/.codex/ 下定位对应的会话 JSONL 文件；
  2. 解析该 JSONL（session_meta / event_msg / response_item / turn_context /
     world_state / compacted 六类记录），还原按 turn 组织的完整对话流；
  3. 渲染成一份 Markdown（或 JSON）交接文档，供另一个 Agent（如 Claude Code）
     阅读并继续完成任务。

用法示例：
  python codex_transfer.py "codex resume 019fcc57-5522-7781-a797-6ef8c4b3dde5"
  python codex_transfer.py 019fcc57 --level summary
  python codex_transfer.py latest -o out.md
  python codex_transfer.py --list 20

结构依据：docs/01-meta-and-envelope.md、02-event-msg.md、03-response-item.md
仅依赖 Python 标准库（3.10+）。
"""
from __future__ import annotations

import argparse
import json
import re
import sqlite3
import sys
from dataclasses import dataclass, field, asdict
from datetime import datetime
from pathlib import Path

CODEX_HOME = Path.home() / ".codex"
SESSIONS_DIR = CODEX_HOME / "sessions"
ARCHIVED_DIR = CODEX_HOME / "archived_sessions"
SQLITE_PATH = CODEX_HOME / "state_5.sqlite"

UUID_RE = re.compile(
    r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-"
    r"[0-9a-fA-F]{4}-[0-9a-fA-F]{12}"
)

# ======================================================================
# 数据结构
# ======================================================================


@dataclass
class Message:
    """一条对话消息（用户输入或 Agent 输出）。"""

    line_no: int
    role: str  # 'user' | 'agent-commentary' | 'agent-final'
    text: str


@dataclass
class ToolCall:
    """一次工具调用及其输出（response_item 的 *_call / *_call_output 对）。"""

    line_no: int
    kind: str  # 'exec' | 'wait' | 其他工具名
    call_id: str
    command: str | None = None  # exec 时提取出的 shell 命令
    workdir: str | None = None
    raw_input: str = ""  # 原始 input / arguments
    output: str = ""  # 拼接后的输出文本
    ok: bool | None = None  # 从输出头部 "Script completed/failed" 判断


@dataclass
class Patch:
    """一次 apply_patch 结果（event_msg 的 patch_apply_end）。"""

    line_no: int
    files: list[tuple[str, str]]  # [(绝对路径, 'add'|'update'), ...]
    stdout: str
    details: dict = field(default_factory=dict)  # path -> {'type', 'content'|'unified_diff'}


@dataclass
class CompactionNotice:
    """上下文压缩事件标记（摘要全文存于 Session.compactions）。"""

    line_no: int
    index: int  # 第几次压缩（从 1 开始）


@dataclass
class Turn:
    turn_id: str
    started_at: int | None = None  # unix 秒
    duration_ms: int | None = None
    items: list = field(default_factory=list)  # Message/ToolCall/Patch/CompactionNotice


@dataclass
class Session:
    path: Path
    session_id: str
    created_utc: str
    cwd: str
    cli_version: str
    model: str = ""
    turns: list[Turn] = field(default_factory=list)
    compactions: list[dict] = field(default_factory=list)  # [{'index','message','line_no'}]
    tokens_total: dict | None = None


# ======================================================================
# 1. 会话文件定位
# ======================================================================


def _strip_extended_prefix(p: str) -> str:
    """剥离 Windows 扩展路径前缀 \\\\?\\。"""
    return p[4:] if p.startswith("\\\\?\\") else p


def query_sqlite_path(session_id: str) -> Path | None:
    """从 state_5.sqlite 的 threads 表查 rollout_path（codex resume 的权威索引）。"""
    if not SQLITE_PATH.exists():
        return None
    uri = "file:" + str(SQLITE_PATH).replace("\\", "/") + "?mode=ro"
    try:
        con = sqlite3.connect(uri, uri=True)
        row = con.execute(
            "SELECT rollout_path FROM threads WHERE id = ?", (session_id,)
        ).fetchone()
        con.close()
    except sqlite3.Error:
        return None
    if not row:
        return None
    p = Path(_strip_extended_prefix(row[0]))
    return p if p.is_file() else None


def glob_session_file(pattern_id: str) -> list[Path]:
    """按 id 子串在 sessions/ 与 archived_sessions/ 下 glob。"""
    matches = []
    for root in (SESSIONS_DIR, ARCHIVED_DIR):
        if root.is_dir():
            matches.extend(root.glob(f"**/rollout-*{pattern_id}*.jsonl"))
    return sorted(set(matches), key=lambda p: p.stat().st_mtime, reverse=True)


def latest_session_path() -> Path | None:
    """最近更新的会话文件（优先 sqlite，回退 glob mtime）。"""
    if SQLITE_PATH.exists():
        uri = "file:" + str(SQLITE_PATH).replace("\\", "/") + "?mode=ro"
        try:
            con = sqlite3.connect(uri, uri=True)
            row = con.execute(
                "SELECT rollout_path FROM threads WHERE archived = 0 "
                "ORDER BY updated_at_ms DESC LIMIT 1"
            ).fetchone()
            con.close()
            if row:
                p = Path(_strip_extended_prefix(row[0]))
                if p.is_file():
                    return p
        except sqlite3.Error:
            pass
    matches = glob_session_file("")
    return matches[0] if matches else None


def resolve_session_path(target: str) -> Path:
    """把用户输入（各种形态）解析为会话 JSONL 的绝对路径。"""
    # 形态 1：直接的文件路径
    p = Path(target)
    if p.is_file():
        return p.resolve()

    # 形态 2：latest
    if target.strip().lower() in ("latest", ":latest"):
        found = latest_session_path()
        if found:
            return found
        sys.exit("[错误] 未找到任何会话文件。")

    # 形态 3：含完整 UUID（如 `codex resume <uuid>` 整串）
    m = UUID_RE.search(target)
    if m:
        sid = m.group(0).lower()
        found = query_sqlite_path(sid)
        if found:
            return found
        matches = glob_session_file(sid)
        if matches:
            return matches[0]
        sys.exit(f"[错误] 找不到 session {sid} 对应的 JSONL（sqlite 与 glob 均未命中）。")

    # 形态 4：id 前缀（如 019fcc57）
    prefix = target.strip().lower()
    if re.fullmatch(r"[0-9a-f]{4,32}", prefix):
        matches = glob_session_file(prefix)
        if not matches:
            sys.exit(f"[错误] 没有文件名包含 id 前缀 {prefix!r} 的会话。")
        if len(matches) > 1:
            print(
                f"[提示] 前缀 {prefix!r} 命中 {len(matches)} 个会话，取最近更新：",
                file=sys.stderr,
            )
            for cand in matches[:5]:
                print(f"  - {cand}", file=sys.stderr)
        return matches[0]

    sys.exit(f"[错误] 无法从输入解析 session 标识：{target!r}")


def list_sessions(n: int) -> None:
    """列出最近 n 个会话（供挑选）。"""
    if SQLITE_PATH.exists():
        uri = "file:" + str(SQLITE_PATH).replace("\\", "/") + "?mode=ro"
        try:
            con = sqlite3.connect(uri, uri=True)
            rows = con.execute(
                "SELECT id, updated_at_ms, archived, title, first_user_message "
                "FROM threads ORDER BY updated_at_ms DESC LIMIT ?",
                (n,),
            ).fetchall()
            con.close()
            print(f"{'SESSION ID':36}  {'最近活动':16}  {'归档':4}  标题 / 首条用户消息")
            for sid, updated_ms, archived, title, first_msg in rows:
                ts = datetime.fromtimestamp(updated_ms / 1000).strftime("%m-%d %H:%M")
                label = (title or "") or (first_msg or "")
                label = re.sub(r"\s+", " ", label)[:48]
                print(f"{sid}  {ts:16}  {'是' if archived else '':4}  {label}")
            return
        except sqlite3.Error as e:
            print(f"[警告] sqlite 读取失败（{e}），回退到文件列表。", file=sys.stderr)
    matches = glob_session_file("")[:n]
    for p in matches:
        ts = datetime.fromtimestamp(p.stat().st_mtime).strftime("%m-%d %H:%M")
        print(f"{p.stem.removeprefix('rollout-')}  {ts}  {p}")


# ======================================================================
# 2. JSONL 解析
# ======================================================================


_JS_STR = r'"((?:[^"\\]|\\.)*)"'  # 匹配一个 JS/JSON 字符串字面量（含转义）


def _unescape_js(s: str) -> str:
    """还原 JS 字符串字面量中的转义序列（与 JSON 字符串转义兼容）。"""
    try:
        return json.loads('"' + s + '"')
    except json.JSONDecodeError:
        return s


def describe_tool_input(kind: str, raw: str) -> tuple[str, str | None]:
    """把工具调用的原始 input 解析成 (人类可读文本, workdir)。

    实测 input 形态（见 docs/03-response-item.md 的补充统计）：
      1. tools.update_plan({plan:[{step:"...",status:"..."}, ...]})   —— 计划更新
      2. tools.shell_command({command:"...", workdir:"..."})          —— key 可无引号，可多次调用
      3. const cmds = [["标签","命令","workdir"], ...]                 —— 批量元组
      4. const patch = "*** Begin Patch\\n..."                         —— apply_patch
      5. function_call 的 arguments 是 JSON 字符串（如 wait）
    """
    # 1. update_plan：提取步骤清单
    if "tools.update_plan(" in raw:
        steps = re.findall(
            rf"step\s*:\s*{_JS_STR}\s*,\s*status\s*:\s*{_JS_STR}", raw
        )
        if steps:
            lines = [f"- [{st}] {_unescape_js(s)}" for s, st in steps]
            return "更新计划：\n" + "\n".join(lines), None
    # 2. shell_command（兼容 JSON 与 JS 字面量 key；一次 input 可含多次调用）
    cmds = re.findall(rf"command\s*:\s*{_JS_STR}", raw)
    if cmds:
        wds = re.findall(rf"workdir\s*:\s*{_JS_STR}", raw)
        text = "\n\n".join(_unescape_js(c) for c in cmds)
        return text, (_unescape_js(wds[0]) if wds else None)
    # 3. cmds 元组数组：["标签", "命令", "workdir"]
    tuples = re.findall(
        rf"\[\s*{_JS_STR}\s*,\s*{_JS_STR}\s*,\s*{_JS_STR}\s*\]", raw
    )
    if tuples:
        lines = [f"[{_unescape_js(lb)}] {_unescape_js(c)}" for lb, c, _ in tuples]
        return "\n\n".join(lines), _unescape_js(tuples[0][2])
    # 4. apply_patch：const patch = "..."
    m = re.search(rf"const\s+patch\s*=\s*{_JS_STR}", raw)
    if m:
        return _unescape_js(m.group(1)), None
    # 5. wait 等 function_call：arguments 为 JSON 字符串
    if kind == "wait":
        try:
            return "wait " + json.dumps(json.loads(raw), ensure_ascii=False), None
        except json.JSONDecodeError:
            pass
    return raw[:500], None


def _join_output(output_field) -> str:
    """*_call_output 的 output 是 [{type:'input_text', text}, ...]，拼接 text。"""
    if not isinstance(output_field, list):
        return str(output_field or "")
    return "".join(
        item.get("text", "") for item in output_field if isinstance(item, dict)
    )


def _judge_ok(output_text: str) -> bool | None:
    head = output_text.lstrip()[:40].lower()
    if head.startswith("script completed"):
        return True
    if head.startswith("script failed") or head.startswith("script error"):
        return False
    return None


# Codex 以 user/developer 角色注入的系统上下文，均以固定标记开头（非真实用户输入）。
# 版本适配层从 response_item/message 提取消息时，按此前缀白名单过滤。
_INJECTED_MESSAGE_PREFIXES = (
    "<environment_context>",
    "<user_instructions>",
    "<skills_instructions>",
    "<permissions instructions>",
    "<multi_agent_mode>",
    "<turn_aborted>",
    "# AGENTS.md instructions",
)


def _is_injected_user_text(text: str) -> bool:
    head = text.lstrip()[:80]
    return any(head.startswith(p) for p in _INJECTED_MESSAGE_PREFIXES)


def parse_session(path: Path) -> Session:
    """逐行解析会话 JSONL，还原按 turn 组织的对话流。"""
    lines: list[tuple[int, dict]] = []
    with path.open("r", encoding="utf-8") as f:
        for i, raw in enumerate(f):
            raw = raw.strip()
            if not raw:
                continue
            try:
                lines.append((i, json.loads(raw)))
            except json.JSONDecodeError:
                continue  # 跳过损坏行，不让整体解析失败

    session: Session | None = None
    turns_by_id: dict[str, Turn] = {}
    turn_order: list[str] = []
    pending_calls: dict[str, ToolCall] = {}  # call_id -> ToolCall（等待 output）
    current_turn_id: str | None = None
    compaction_count = 0
    # —— 版本适配：Codex ≥0.15x 不再写 event_msg 的 user_message/agent_message，
    #    消息文本只在 response_item/message。先收集备用，解析完做会话级判断。 ——
    ri_messages: list[tuple[int, str, str, str]] = []  # (line_no, turn_id, role, text)
    event_msg_message_count = 0
    completed_turn_ids: set[str] = set()
    aborted_turn_ids: set[str] = set()

    def get_turn(turn_id: str) -> Turn:
        if turn_id not in turns_by_id:
            turns_by_id[turn_id] = Turn(turn_id=turn_id)
            turn_order.append(turn_id)
        return turns_by_id[turn_id]

    for line_no, rec in lines:
        rtype = rec.get("type")
        payload = rec.get("payload") or {}

        if rtype == "session_meta":
            session = Session(
                path=path,
                session_id=payload.get("session_id") or payload.get("id") or "",
                created_utc=payload.get("timestamp", ""),
                cwd=payload.get("cwd", ""),
                cli_version=payload.get("cli_version", ""),
            )

        elif rtype == "turn_context":
            # model 以最后一次出现的为准（用户可能中途切换）
            if payload.get("model") and session is not None:
                session.model = payload["model"]

        elif rtype == "event_msg":
            etype = payload.get("type")
            if etype == "task_started":
                current_turn_id = payload.get("turn_id", "")
                if current_turn_id:
                    t = get_turn(current_turn_id)
                    t.started_at = payload.get("started_at")
            elif etype == "task_complete":
                tid = payload.get("turn_id") or current_turn_id
                if tid and tid in turns_by_id:
                    turns_by_id[tid].duration_ms = payload.get("duration_ms")
                if tid:
                    completed_turn_ids.add(tid)
            elif etype == "turn_aborted":
                tid = payload.get("turn_id") or current_turn_id
                if tid:
                    aborted_turn_ids.add(tid)
            elif etype == "user_message" and current_turn_id:
                event_msg_message_count += 1
                get_turn(current_turn_id).items.append(
                    Message(line_no, "user", payload.get("message", ""))
                )
            elif etype == "agent_message" and current_turn_id:
                event_msg_message_count += 1
                phase = payload.get("phase", "commentary")
                role = "agent-final" if phase == "final_answer" else "agent-commentary"
                get_turn(current_turn_id).items.append(
                    Message(line_no, role, payload.get("message", ""))
                )
            elif etype == "patch_apply_end":
                changes = payload.get("changes") or {}
                files = [
                    (fp, (detail or {}).get("type", "?"))
                    for fp, detail in changes.items()
                ]
                details = {
                    fp: {k: v for k, v in (detail or {}).items() if k in ("type", "content", "unified_diff", "move_path")}
                    for fp, detail in changes.items()
                }
                tid = payload.get("turn_id") or current_turn_id
                if tid:
                    get_turn(tid).items.append(
                        Patch(line_no, files, payload.get("stdout", ""), details)
                    )
            elif etype == "token_count":
                # 只保留最后一个，作为会话累计用量
                info = (payload.get("info") or {}).get("total_token_usage")
                if info and session is not None:
                    session.tokens_total = info

        elif rtype == "response_item":
            ptype = payload.get("type")
            tid = (payload.get("internal_chat_message_metadata_passthrough") or {}).get(
                "turn_id"
            )
            if ptype == "message":
                # 版本适配：新格式消息来源，先收集，是否回放见 parse_session 末尾
                text = "".join(
                    c.get("text", "")
                    for c in payload.get("content", [])
                    if isinstance(c, dict)
                )
                ri_messages.append(
                    (line_no, tid or current_turn_id or "", payload.get("role", ""), text)
                )
            if ptype in ("custom_tool_call", "function_call"):
                call = ToolCall(
                    line_no=line_no,
                    kind=payload.get("name", "?"),
                    call_id=payload.get("call_id", ""),
                    raw_input=payload.get("input") or payload.get("arguments") or "",
                )
                call.command, call.workdir = describe_tool_input(
                    call.kind, call.raw_input
                )
                pending_calls[call.call_id] = call
                if tid:
                    get_turn(tid).items.append(call)
            elif ptype in ("custom_tool_call_output", "function_call_output"):
                cid = payload.get("call_id", "")
                if cid in pending_calls:
                    call = pending_calls[cid]
                    call.output = _join_output(payload.get("output"))
                    call.ok = _judge_ok(call.output)

        elif rtype == "compacted":
            compaction_count += 1
            if session is not None:
                session.compactions.append(
                    {
                        "index": compaction_count,
                        "message": payload.get("message", ""),
                        "line_no": line_no,
                    }
                )
            if current_turn_id:
                get_turn(current_turn_id).items.append(
                    CompactionNotice(line_no, compaction_count)
                )

    if session is None:
        sys.exit(f"[错误] {path} 中缺少 session_meta 首行，不是合法的 Codex 会话文件。")

    # —— 版本适配回放：仅当整个会话没有任何 event_msg 消息（Codex ≥0.15x 格式）
    #    才从 response_item/message 提取，老格式会话两种来源并存，避免双写。 ——
    if event_msg_message_count == 0 and ri_messages:
        for line_no, tid, role, text in ri_messages:
            if not tid or not text:
                continue
            if role == "developer" or _is_injected_user_text(text):
                continue  # 系统注入（skills/permissions/environment/AGENTS.md）不迁移
            if role == "user":
                get_turn(tid).items.append(Message(line_no, "user", text))
            elif role == "assistant":
                get_turn(tid).items.append(Message(line_no, "agent-commentary", text))
        # 正常完成的 turn：最后一条 assistant 提升为最终回答；
        # 被中断（turn_aborted）的保持 commentary，让渲染层走"无最终回答"路径。
        for tid in turn_order:
            if tid in completed_turn_ids and tid not in aborted_turn_ids:
                agent_msgs = [
                    it
                    for it in turns_by_id[tid].items
                    if isinstance(it, Message) and it.role == "agent-commentary"
                ]
                if agent_msgs:
                    agent_msgs[-1].role = "agent-final"

    # turn 排序（按首个 item 的行号 / started_at）并整理 items 顺序
    def turn_sort_key(t: Turn):
        first_line = min((it.line_no for it in t.items), default=1 << 30)
        return (t.started_at or 0, first_line)

    session.turns = sorted(
        (turns_by_id[tid] for tid in turn_order), key=turn_sort_key
    )
    for t in session.turns:
        t.items.sort(key=lambda it: it.line_no)
    return session


# ======================================================================
# 3. 渲染
# ======================================================================


def _truncate(text: str, limit: int) -> str:
    if limit <= 0 or len(text) <= limit:
        return text
    return text[:limit] + f"\n...[截断，原文共 {len(text)} 字符]"


def _fmt_ts(unix_sec: int | None) -> str:
    if not unix_sec:
        return "时间未知"
    return datetime.fromtimestamp(unix_sec).strftime("%Y-%m-%d %H:%M:%S")


def _collect_file_changes(session: Session) -> dict[str, list[str]]:
    """聚合全部 patch：文件路径 -> 操作类型列表。"""
    agg: dict[str, list[str]] = {}
    for t in session.turns:
        for it in t.items:
            if isinstance(it, Patch):
                for fp, ctype in it.files:
                    agg.setdefault(fp, []).append(ctype)
    return agg


_CN_NUM = ["", "一", "二", "三", "四", "五", "六", "七", "八"]


def _cn_sec(n: int) -> str:
    return _CN_NUM[n] if 0 < n < len(_CN_NUM) else str(n)


def render_markdown(session: Session, level: str, max_output_chars: int) -> str:
    show_commentary = level in ("normal", "full")
    show_tools = level in ("normal", "full")
    show_patch_detail = level == "full"
    out_limit = 0 if level == "full" else max_output_chars

    w: list[str] = []
    a = w.append
    sec = 0  # 章节计数器

    # ---- 头部：交接说明 ----
    a("# Codex → Claude Code 会话迁移交接文档\n")
    a("> 本文档由 `codex_transfer.py` 从 Codex CLI 会话记录自动生成。")
    a("> 接手 Agent 请完整阅读，理解任务背景与已完成的进展后，从文末「当前状态与继续指示」处接续工作。\n")

    # ---- 会话元信息 ----
    sec += 1
    a(f"## {_cn_sec(sec)}、会话元信息\n")
    a("| 项 | 值 |")
    a("|---|---|")
    a(f"| Session ID | `{session.session_id}` |")
    a(f"| 源文件 | `{session.path}` |")
    a(f"| 创建时间 (UTC) | {session.created_utc} |")
    a(f"| 工作目录 | `{session.cwd}` |")
    a(f"| 模型 | {session.model or '未知'} |")
    a(f"| Codex CLI 版本 | {session.cli_version} |")
    a(f"| 对话轮数 | {len(session.turns)} |")
    a(f"| 上下文压缩次数 | {len(session.compactions)} |")
    if session.tokens_total:
        tk = session.tokens_total
        a(
            f"| 累计 token | 输入 {tk.get('input_tokens', 0):,}"
            f"（缓存 {tk.get('cached_input_tokens', 0):,}）"
            f" / 输出 {tk.get('output_tokens', 0):,} |"
        )
    a("")

    # ---- 压缩摘要（如有）----
    if session.compactions:
        sec += 1
        a(f"## {_cn_sec(sec)}、任务概况（上下文压缩摘要）\n")
        a("Codex 在会话中途触发过上下文压缩，以下摘要由模型在压缩点生成，"
          "概括了压缩点之前的全部工作：\n")
        for c in session.compactions:
            a(f"### 压缩 #{c['index']} 的摘要\n")
            a(c["message"].strip() or "（空摘要）")
            a("")

    # ---- 对话记录 ----
    sec += 1
    a(f"## {_cn_sec(sec)}、对话记录\n")
    a("图例：👤 用户 / 🤖 Codex 过程叙述 / 🔧 工具调用 / 📄 文件修改 / ✅ Codex 最终回答\n")

    for idx, turn in enumerate(session.turns, 1):
        dur = f"，耗时 {turn.duration_ms / 1000:.0f}s" if turn.duration_ms else ""
        a(f"### Turn {idx} · {_fmt_ts(turn.started_at)}{dur}\n")
        for it in turn.items:
            if isinstance(it, Message):
                if it.role == "user":
                    a("**👤 用户：**\n")
                    a(it.text.strip())
                    a("")
                elif it.role == "agent-final":
                    a("**✅ Codex 最终回答：**\n")
                    a(it.text.strip())
                    a("")
                elif show_commentary:
                    a("**🤖 Codex：** " + it.text.strip() + "\n")
            elif isinstance(it, ToolCall) and show_tools:
                status = {True: "成功", False: "失败", None: "未知"}[it.ok]
                cmd = it.command or ""
                if cmd.startswith("更新计划"):
                    title, fence_limit = "📝 更新计划", 2000
                elif cmd.startswith("*** Begin Patch"):
                    # 补丁全文已由 patch_apply_end 记录，这里只留标识
                    title, fence_limit = "📄 应用补丁", 600
                elif it.kind == "exec":
                    title, fence_limit = "🔧 执行命令", 2000
                elif it.kind == "wait":
                    title, fence_limit = "⏳ 等待", 200
                else:
                    title, fence_limit = f"🔧 工具调用 `{it.kind}`", 500
                a(f"**{title}**（结果：{status}）：\n")
                if cmd:
                    a("~~~text")
                    a(_truncate(cmd, fence_limit))
                    a("~~~")
                if it.output and not cmd.startswith("*** Begin Patch"):
                    a(f"输出：\n\n~~~text\n{_truncate(it.output, out_limit)}\n~~~\n")
            elif isinstance(it, Patch):
                files_str = "、".join(f"`{fp}`（{ct}）" for fp, ct in it.files)
                a(f"**📄 文件修改：** {files_str}\n")
                if show_patch_detail:
                    for fp, detail in it.details.items():
                        body = detail.get("content") or detail.get("unified_diff") or ""
                        if body:
                            a(f"<details><summary>{fp} 的变更内容</summary>\n")
                            a(f"~~~diff\n{body}\n~~~\n</details>\n")
            elif isinstance(it, CompactionNotice):
                a(f"> ⚠️ **上下文压缩 #{it.index}**：Codex 上下文窗口将满，"
                  f"此前历史被摘要替换（摘要见上文「任务概况」）。\n")

    # ---- 文件修改汇总 ----
    changes = _collect_file_changes(session)
    sec += 1
    a(f"## {_cn_sec(sec)}、文件修改汇总\n")
    if changes:
        a("| 文件 | 操作历史 |")
        a("|---|---|")
        for fp, ops in sorted(changes.items()):
            a(f"| `{fp}` | {' → '.join(ops)} |")
        a("\n> 以上文件已被 Codex 修改，接手时请直接读取现状，不要重复修改。")
    else:
        a("（本会话无文件修改记录）")
    a("")

    # ---- 五、当前状态与继续指示 ----
    last_turn = session.turns[-1] if session.turns else None
    last_user = next(
        (it for it in reversed(last_turn.items) if isinstance(it, Message) and it.role == "user"),
        None,
    ) if last_turn else None
    last_agent = next(
        (it for it in reversed(last_turn.items) if isinstance(it, Message) and it.role.startswith("agent")),
        None,
    ) if last_turn else None
    last_final = next(
        (it for it in reversed(last_turn.items) if isinstance(it, Message) and it.role == "agent-final"),
        None,
    ) if last_turn else None

    sec += 1
    a(f"## {_cn_sec(sec)}、当前状态与继续指示\n")
    if last_user:
        a(f"### 最后一个 turn（Turn {len(session.turns)}）的用户需求\n")
        a(last_user.text.strip())
        a("")
    if last_final:
        a("该 turn 已有 Codex 最终回答（见对话记录），任务可能已阶段性完成，请先向用户确认下一步。\n")
    elif last_agent:
        a("⚠️ 该 turn **没有最终回答**（会话在过程中中断），这就是需要继续的现场。"
          "Codex 最后一条消息：\n")
        a(_truncate(last_agent.text.strip(), 1500))
        a("")
    elif last_user:
        a("⚠️ 该 turn **尚无 Codex 回复**——用户发完消息后会话即结束，"
          "请直接响应上述需求。\n")
    a("### 环境与进度\n")
    a(f"- **工作目录**：`{session.cwd}`（请确认你在同一目录下操作）。")
    if changes:
        a(f"- **已修改 {len(changes)} 个文件**：见「文件修改汇总」，读取现状即可。")
    a("")
    a("### 给接手 Agent 的指示\n")
    a("1. 完整阅读上文对话记录，特别是用户各轮需求与 Codex 的最终回答；")
    a("2. 用户的语言习惯、项目约定（如 AGENTS.md 规范）在对话中已有体现，请保持一致；")
    a("3. 不要重做已完成的工作（文件修改、已验证的结论）；")
    if last_user and not last_final:
        a("4. **直接从最后一个 turn 的用户需求继续执行**，把中断的工作做完。")
    else:
        a("4. 向用户简要汇报你对现状的理解，并询问/确认下一步任务。")
    a("")

    return "\n".join(w)


def render_json(session: Session, level: str, max_output_chars: int) -> str:
    out_limit = 0 if level == "full" else max_output_chars
    data = {
        "session_id": session.session_id,
        "source_file": str(session.path),
        "created_utc": session.created_utc,
        "cwd": session.cwd,
        "model": session.model,
        "cli_version": session.cli_version,
        "tokens_total": session.tokens_total,
        "compactions": session.compactions,
        "turns": [],
    }
    for t in session.turns:
        td = {
            "turn_id": t.turn_id,
            "started_at": t.started_at,
            "duration_ms": t.duration_ms,
            "items": [],
        }
        for it in t.items:
            if isinstance(it, Message):
                td["items"].append({"kind": it.role, "text": it.text})
            elif isinstance(it, ToolCall):
                if level == "summary":
                    continue
                td["items"].append(
                    {
                        "kind": "tool_call",
                        "tool": it.kind,
                        "command": it.command,
                        "workdir": it.workdir,
                        "ok": it.ok,
                        "output": _truncate(it.output, out_limit),
                    }
                )
            elif isinstance(it, Patch):
                item = {"kind": "patch", "files": it.files}
                if level == "full":
                    item["details"] = it.details
                td["items"].append(item)
            elif isinstance(it, CompactionNotice):
                td["items"].append({"kind": "compaction", "index": it.index})
        data["turns"].append(td)
    return json.dumps(data, ensure_ascii=False, indent=2)


# ======================================================================
# main
# ======================================================================


def main() -> None:
    # Windows 控制台中文输出防御
    try:
        sys.stdout.reconfigure(encoding="utf-8", errors="replace")
        sys.stderr.reconfigure(encoding="utf-8", errors="replace")
    except Exception:
        pass

    ap = argparse.ArgumentParser(
        description="Codex CLI 会话迁移工具：定位会话 JSONL 并渲染成交接文档",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__,
    )
    ap.add_argument(
        "target",
        nargs="?",
        default="latest",
        help="'codex resume <uuid>' 整串 / uuid / uuid 前缀 / JSONL 路径 / latest（默认）",
    )
    ap.add_argument("-o", "--output", help="输出文件路径（默认 output/handoff-<id8>-<时间戳>.<ext>）")
    ap.add_argument(
        "--level",
        choices=["summary", "normal", "full"],
        default="normal",
        help="详略级别：summary=用户+最终回答；normal=+过程与工具摘要（默认）；full=不截断",
    )
    ap.add_argument("--format", choices=["md", "json"], default="md", help="输出格式")
    ap.add_argument(
        "--max-output-chars",
        type=int,
        default=800,
        help="单个工具输出的截断长度（默认 800，full 级别忽略）",
    )
    ap.add_argument("--list", metavar="N", type=int, nargs="?", const=15, help="列出最近 N 个会话后退出")
    ap.add_argument("--stdout", action="store_true", help="不写文件，直接打印到标准输出")
    args = ap.parse_args()

    if args.list is not None:
        list_sessions(args.list)
        return

    path = resolve_session_path(args.target)
    print(f"[定位] {path}", file=sys.stderr)

    session = parse_session(path)
    print(
        f"[解析] session={session.session_id}  turns={len(session.turns)}"
        f"  压缩={len(session.compactions)}  cwd={session.cwd}",
        file=sys.stderr,
    )

    if args.format == "json":
        content = render_json(session, args.level, args.max_output_chars)
        ext = "json"
    else:
        content = render_markdown(session, args.level, args.max_output_chars)
        ext = "md"

    if args.stdout:
        print(content)
        return

    if args.output:
        out_path = Path(args.output)
    else:
        out_dir = Path(__file__).resolve().parent / "output"
        out_dir.mkdir(exist_ok=True)
        ts = datetime.now().strftime("%Y%m%d-%H%M%S")
        out_path = out_dir / f"handoff-{session.session_id[:8]}-{ts}.{ext}"
    out_path.parent.mkdir(parents=True, exist_ok=True)
    out_path.write_text(content, encoding="utf-8")
    size_kb = out_path.stat().st_size / 1024
    print(f"[完成] 交接文档已写入 {out_path}（{size_kb:.1f} KB）", file=sys.stderr)


if __name__ == "__main__":
    main()