verify_session.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
DSH 会话文件验证器

检查 ~/.dsh/sessions 下（或指定根目录）的所有会话文件是否符合
dsh-session-persistence-jsonl 的读取要求：

  1. zstd 多帧布局：第一帧解压后必须恰好一行 header（以 \n 结尾）
  2. 事件行完整（帧不撕裂 JSONL 记录、无尾随残缺）
  3. 每个 turn/end 事件携带合法 reason（migrateLegacyTurnEndEvent 要求）
  4. header 字段完整（type/version/id/createdAt/cwd/agentPreset）
  5. 记录 session/title 事件（存在则打印标题，供核对与 Claude 原标题一致）

退出码：0 = 全部通过；1 = 存在问题；2 = 用法错误。

用法:
  python3 verify_session.py [ROOT]
  ROOT 默认 ~/.dsh/sessions

配合 convert_claude_to_dsh.py 使用：
  导入后先跑本脚本确认格式合规，再通过 API 挂载工作区。
"""
import json
import os
import subprocess
import sys

ZSTD_MAGIC = b"\x28\xb5\x2f\xfd"  # little-endian 0xFD2FB528


def scan_frames(data: bytes):
    """按 DSH 的 scanZstdFrames 逻辑定位完整帧，返回 (frames, torn_start)。"""
    frames = []
    offset = 0
    while offset < len(data):
        start = offset
        if len(data) - offset < 4:
            return frames, start
        if data[offset:offset + 4] != ZSTD_MAGIC:
            raise ValueError("invalid frame magic at byte %d" % offset)
        offset += 4
        if offset == len(data):
            return frames, start
        descriptor = data[offset]
        offset += 1
        if (descriptor & 24) != 0:
            raise ValueError("reserved frame-header bit at byte %d" % (offset - 1))
        content_size_flag = descriptor >> 6
        single_segment = (descriptor & 32) != 0
        checksum = (descriptor & 4) != 0
        dictionary_flag = descriptor & 3
        dictionary_bytes = 4 if dictionary_flag == 3 else dictionary_flag
        content_size_bytes = (1 if single_segment else 0) if content_size_flag == 0 else 1 << content_size_flag
        remaining_header = (0 if single_segment else 1) + dictionary_bytes + content_size_bytes
        if len(data) - offset < remaining_header:
            return frames, start
        offset += remaining_header
        while True:
            if len(data) - offset < 3:
                return frames, start
            block_header = data[offset] | (data[offset + 1] << 8) | (data[offset + 2] << 16)
            offset += 3
            last_block = (block_header & 1) != 0
            block_type = (block_header >> 1) & 3
            block_size = block_header >> 3
            if block_type == 3:
                raise ValueError("reserved block type at byte %d" % (offset - 3))
            payload_bytes = 1 if block_type == 1 else block_size
            if len(data) - offset < payload_bytes:
                return frames, start
            offset += payload_bytes
            if last_block:
                break
        if checksum:
            if len(data) - offset < 4:
                return frames, start
            offset += 4
        frames.append((start, offset))
    return frames, None


def zstd_decompress_frame(data: bytes) -> bytes:
    r = subprocess.run(["zstd", "-q", "-d", "-c"], input=data,
                       capture_output=True, timeout=60)
    if r.returncode != 0:
        raise ValueError("zstd decode failed: %s" % r.stderr.decode()[:120])
    return r.stdout


def check_session(log_path):
    problems = []
    with open(log_path, "rb") as f:
        data = f.read()

    # 1) 帧布局
    frames, torn = scan_frames(data)
    if torn is not None:
        problems.append("torn/incomplete final frame at byte %d" % torn)
    if len(frames) < 2:
        problems.append("expected >=2 frames (header + events), got %d" % len(frames))

    # 2) 第一帧 = 恰好一行 header
    header_text = zstd_decompress_frame(data[frames[0][0]:frames[0][1]]).decode("utf-8")
    nl = header_text.find("\n")
    if not (len(header_text) > 0 and nl == len(header_text) - 1):
        problems.append("first frame is not exactly one header line")
    try:
        header = json.loads(header_text.strip())
    except Exception as e:
        problems.append("header not valid JSON: %s" % e)
        return None, [], problems

    # 3) 事件内容
    events = []
    for start, end in frames[1:]:
        plain = zstd_decompress_frame(data[start:end]).decode("utf-8")
        events.extend(plain.split("\n"))
    events = [e for e in events if e.strip() != ""]

    turn_ends = 0
    for line in events:
        try:
            ev = json.loads(line)
        except Exception as e:
            problems.append("event line not JSON: %s" % e)
            continue
        if ev.get("type") == "turn/end":
            turn_ends += 1
            reason = (ev.get("data") or {}).get("reason")
            if not isinstance(reason, dict) or not isinstance(reason.get("kind"), str):
                problems.append("turn/end at seq %s missing reason.kind" % ev.get("seq"))

    # 4) header 字段
    for key in ("type", "version", "id", "createdAt", "cwd", "agentPreset"):
        if key not in header:
            problems.append("header missing %r" % key)

    return header, events, problems
def main():
    root = sys.argv[1] if len(sys.argv) > 1 else os.path.expanduser("~/.dsh/sessions")
    if not os.path.isdir(root):
        print("目录不存在: %s" % root, file=sys.stderr)
        return 2

    total = bad = 0
    for proj in sorted(os.listdir(root)):
        proj_dir = os.path.join(root, proj)
        if not os.path.isdir(proj_dir):
            continue
        for sess in sorted(os.listdir(proj_dir)):
            sess_dir = os.path.join(proj_dir, sess)
            log = os.path.join(sess_dir, "session.jsonl.zstd")
            if not os.path.isfile(log):
                log = os.path.join(sess_dir, "session.jsonl")
            if not os.path.isfile(log):
                continue
            total += 1
            header, events, problems = check_session(log)
            title = None
            for ev in events:
                if isinstance(ev, str):
                    try:
                        parsed = json.loads(ev)
                    except Exception:
                        continue
                    if parsed.get("type") == "session/title":
                        title = (parsed.get("data") or {}).get("title")
            status = "OK" if not problems else "BAD"
            if status == "BAD":
                bad += 1
            print("[%s] %s%s cwd=%s%s%s" % (
                status, sess,
                " title=%s" % json.dumps(title, ensure_ascii=False) if title else "",
                (header or {}).get("cwd", "?"),
                "" if not problems else "  problems: " + "; ".join(problems),
                "" if not problems else "  frames=%d events=%d" % (len(scan_frames(open(log, "rb").read())[0]), len(events)),
            ))
    print("\n总计 %d 个会话, 问题 %d" % (total, bad))
    return 1 if bad else 0


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