#!/usr/bin/env python3 """Inspect local Kimi Code CLI sessions and verify captured data quality. Run this to see whether session_capture.py can correctly reconstruct: - user inputs - Kimi's thinking traces - Kimi's final answers - tool calls Examples: python3 tests/inspect_sessions.py python3 tests/inspect_sessions.py --latest python3 tests/inspect_sessions.py --session-dir /Users/admin/.kimi-code/sessions/wd_xxx/session_yyy """ import argparse import os import sys from pathlib import Path sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from session_capture import list_all_sessions, load_session def _format_text(text: str, max_chars: int = 400) -> str: if len(text) <= max_chars: return text return text[:max_chars] + f"\n... ({len(text) - max_chars} more chars)" def inspect_session(session_dir: Path, verbose: bool = False) -> None: session = load_session(session_dir) print(f"\n{'=' * 60}") print(f"Session ID: {session.session_id}") print(f"Work dir: {session.work_dir}") print(f"Created: {session.created_at}") print(f"Turns: {len(session.turns)}") print(f"Wire file: {session.wire_path}") print("=" * 60) if not session.turns: print("No turns parsed.") return for idx, turn in enumerate(session.turns, 1): has_thinking = bool(turn.clean_thinking) has_answer = bool(turn.clean_answer) has_tools = bool(turn.tool_summary) raw_user = turn.user_input clean_user = turn.clean_user_input print( f"\n--- Turn {idx} (id={turn.turn_id}) " f"[thinking={has_thinking}, answer={has_answer}, tools={has_tools}] ---" ) print("USER INPUT (cleaned):") print(_format_text(clean_user)) if clean_user != raw_user and len(raw_user) > len(clean_user): print(f"[raw input was {len(raw_user)} chars, cleaned to {len(clean_user)} chars]") if has_thinking: print("\nKIMI THINKING:") print(_format_text(turn.clean_thinking)) if has_tools: print("\nTOOL SUMMARY:") print(_format_text(turn.tool_summary, max_chars=150)) if has_answer: print("\nKIMI ANSWER:") print(_format_text(turn.clean_answer)) def main(): parser = argparse.ArgumentParser(description="Inspect local Kimi Code CLI sessions") parser.add_argument( "--session-dir", type=Path, default=None, help="Inspect a specific session directory", ) parser.add_argument( "--latest", action="store_true", help="Inspect the latest session by modification time", ) parser.add_argument( "--list", action="store_true", help="List all known sessions and exit", ) parser.add_argument( "--base-dir", type=Path, default=None, help="Override the Kimi Code CLI data directory (default: ~/.kimi-code)", ) parser.add_argument( "--work-dir", type=str, default=None, help="Only inspect sessions whose work dir contains this string", ) parser.add_argument( "--session-id", type=str, default=None, help="Inspect a specific session by its ID", ) args = parser.parse_args() base_dirs = [args.base_dir] if args.base_dir else None if args.session_id: target_id = args.session_id.strip().lower() all_metas = list_all_sessions(base_dirs) matched = [m for m in all_metas if m.session_id.lower() == target_id] if not matched: # Allow partial matching in case the user typed a prefix/suffix. matched = [m for m in all_metas if target_id in m.session_id.lower()] if not matched: print(f"Session '{args.session_id}' not found.") return if len(matched) > 1: print(f"Multiple sessions matched '{args.session_id}':") for m in matched: print(f" {m.session_id} -> {m.session_dir}") return inspect_session(matched[0].session_dir, verbose=True) return if args.list: print("All discovered sessions:") for meta in list_all_sessions(base_dirs, work_dir=args.work_dir): print( f" [{meta.source}] {meta.session_id} -> {meta.session_dir} " f"(work: {meta.work_dir})" ) return if args.session_dir: inspect_session(args.session_dir, verbose=True) return if args.latest: all_metas = list_all_sessions(base_dirs, work_dir=args.work_dir) if not all_metas: print("No sessions found.") return latest = max(all_metas, key=lambda m: m.session_dir.stat().st_mtime) inspect_session(latest.session_dir, verbose=True) return # Default: inspect every session, printing a compact summary. print("Inspecting all sessions...") sessions = list_all_sessions(base_dirs, work_dir=args.work_dir) if not sessions: print("No sessions found.") return total_turns = 0 total_with_thinking = 0 total_with_answer = 0 total_with_tools = 0 total_raw_user_chars = 0 total_clean_user_chars = 0 for meta in sessions: session = load_session(meta.session_dir, source=meta.source) wire_exists = session.wire_path.exists() if session.wire_path else False print( f"{session.session_id}: {len(session.turns)} turns " f"work={session.work_dir or meta.work_dir} " f"wire_exists={wire_exists}" ) total_turns += len(session.turns) for turn in session.turns: total_raw_user_chars += len(turn.user_input) total_clean_user_chars += len(turn.clean_user_input) if turn.clean_thinking: total_with_thinking += 1 if turn.clean_answer: total_with_answer += 1 if turn.tool_summary: total_with_tools += 1 print("\n--- Summary ---") print(f"Sessions loaded: {len(sessions)}") print(f"Total turns: {total_turns}") print(f"With thinking: {total_with_thinking}") print(f"With answer: {total_with_answer}") print(f"With tools: {total_with_tools}") if total_raw_user_chars: ratio = total_clean_user_chars / total_raw_user_chars * 100 print( f"User input chars: {total_raw_user_chars} raw -> " f"{total_clean_user_chars} clean ({ratio:.1f}%)" ) if __name__ == "__main__": main()