#!/usr/bin/env python3 """session-transcript-converge — hardlink farm logic, tested against a fixture tree so it never touches the real machine's Claude homes.""" import json import os import shutil import sys import tempfile import unittest from importlib.machinery import SourceFileLoader SCRIPT = os.path.join(os.path.dirname(__file__), "..", "bin", "session-transcript-converge") conv = SourceFileLoader("session_transcript_converge", SCRIPT).load_module() def write_transcript(path, session_id): os.makedirs(os.path.dirname(path), exist_ok=True) with open(path, "w") as f: f.write(json.dumps({"type": "user", "sessionId": session_id}) + "\n") class ConvergeTest(unittest.TestCase): def setUp(self): self.tmp = tempfile.mkdtemp(prefix="sess-conv-test-") self.default_projects = os.path.join(self.tmp, "home", ".claude", "projects") os.makedirs(self.default_projects, exist_ok=True) conv.DEFAULT_PROJECTS = os.path.realpath(self.default_projects) def tearDown(self): shutil.rmtree(self.tmp, ignore_errors=True) def test_base_project_name(self): self.assertEqual( conv.base_project_name("-home-user-Projects-overdeck--worktrees-foo"), "-home-user-Projects-overdeck", ) self.assertEqual( conv.base_project_name("-home-user-Projects-overdeck--claude-worktrees-agent-x"), "-home-user-Projects-overdeck", ) self.assertIsNone(conv.base_project_name("-home-user-Projects-overdeck")) def test_worktree_session_links_into_base(self): base = os.path.join(self.default_projects, "-home-user-Projects-overdeck") os.makedirs(base, exist_ok=True) wt = os.path.join(self.default_projects, "-home-user-Projects-overdeck--worktrees-foo") write_transcript(os.path.join(wt, "abc.jsonl"), "abc") result = conv.converge_homes([self.default_projects], dry_run=False) dest = os.path.join(base, "abc.jsonl") self.assertTrue(os.path.exists(dest)) self.assertEqual(os.stat(dest).st_ino, os.stat(os.path.join(wt, "abc.jsonl")).st_ino) self.assertEqual(result["linked"], 1) self.assertEqual(result["errors"], []) def test_agent_subagent_dirs_are_skipped(self): base = os.path.join(self.default_projects, "-home-user-Projects-overdeck") os.makedirs(base, exist_ok=True) agentdir = os.path.join(self.default_projects, "-home-user-Projects-overdeck--claude-worktrees-agent-x") write_transcript(os.path.join(agentdir, "sub.jsonl"), "sub") result = conv.converge_homes([self.default_projects], dry_run=False) self.assertFalse(os.path.exists(os.path.join(base, "sub.jsonl"))) self.assertEqual(result["linked"], 0) self.assertEqual(result["skipped_agent_dirs"], 1) def test_no_base_dir_yet_dry_run_skips_real_run_creates(self): wt = os.path.join(self.default_projects, "-home-user-Projects-newthing--worktrees-foo") write_transcript(os.path.join(wt, "x.jsonl"), "x") dry = conv.converge_homes([self.default_projects], dry_run=True) self.assertEqual(dry["linked"], 0) self.assertEqual(dry["skipped_no_base_dir"], 1) real = conv.converge_homes([self.default_projects], dry_run=False) base = os.path.join(self.default_projects, "-home-user-Projects-newthing") self.assertEqual(real["linked"], 1) self.assertEqual(real["errors"], []) self.assertTrue(os.path.exists(os.path.join(base, "x.jsonl"))) def test_alternate_home_plain_project_dir_adopts_into_default(self): alt = os.path.join(self.tmp, "alt-home-projects") altproj = os.path.join(alt, "-home-user-Projects-multideal") write_transcript(os.path.join(altproj, "recovered.jsonl"), "recovered") defproj = os.path.join(self.default_projects, "-home-user-Projects-multideal") result = conv.converge_homes([self.default_projects, alt], dry_run=False) self.assertEqual(result["errors"], []) self.assertEqual(result["linked"], 1) dest = os.path.join(defproj, "recovered.jsonl") self.assertTrue(os.path.exists(dest)) self.assertTrue(os.stat(dest).st_ino == os.stat(os.path.join(altproj, "recovered.jsonl")).st_ino) def test_default_home_plain_project_dir_is_left_alone(self): base = os.path.join(self.default_projects, "-home-user-Projects-overdeck") write_transcript(os.path.join(base, "already-home.jsonl"), "home") result = conv.converge_homes([self.default_projects], dry_run=False) self.assertEqual(result["linked"], 0) self.assertEqual(result["errors"], []) def test_collision_is_reported_and_never_overwrites(self): base = os.path.join(self.default_projects, "-home-user-Projects-overdeck") os.makedirs(base, exist_ok=True) write_transcript(os.path.join(base, "abc.jsonl"), "abc-original") wt = os.path.join(self.default_projects, "-home-user-Projects-overdeck--worktrees-foo") write_transcript(os.path.join(wt, "abc.jsonl"), "abc-different") result = conv.converge_homes([self.default_projects], dry_run=False) with open(os.path.join(base, "abc.jsonl")) as f: self.assertIn("abc-original", f.read()) self.assertEqual(result["linked"], 0) self.assertEqual(len(result["errors"]), 1) self.assertIn("collision", result["errors"][0]) def test_rerun_is_idempotent(self): base = os.path.join(self.default_projects, "-home-user-Projects-overdeck") os.makedirs(base, exist_ok=True) wt = os.path.join(self.default_projects, "-home-user-Projects-overdeck--worktrees-foo") write_transcript(os.path.join(wt, "abc.jsonl"), "abc") conv.converge_homes([self.default_projects], dry_run=False) second = conv.converge_homes([self.default_projects], dry_run=False) self.assertEqual(second["linked"], 0) self.assertEqual(second["already_linked"], 1) self.assertEqual(second["errors"], []) def test_dry_run_does_not_link(self): base = os.path.join(self.default_projects, "-home-user-Projects-overdeck") os.makedirs(base, exist_ok=True) wt = os.path.join(self.default_projects, "-home-user-Projects-overdeck--worktrees-foo") write_transcript(os.path.join(wt, "abc.jsonl"), "abc") result = conv.converge_homes([self.default_projects], dry_run=True) self.assertFalse(os.path.exists(os.path.join(base, "abc.jsonl"))) self.assertEqual(result["linked"], 1) def test_live_dot_jsonl_companion_is_never_linked(self): base = os.path.join(self.default_projects, "-home-user-Projects-overdeck") os.makedirs(base, exist_ok=True) wt = os.path.join(self.default_projects, "-home-user-Projects-overdeck--worktrees-foo") write_transcript(os.path.join(wt, "abc.jsonl"), "abc") write_transcript(os.path.join(wt, "abc.jsonl.live.jsonl"), "abc-live") conv.converge_homes([self.default_projects], dry_run=False) self.assertTrue(os.path.exists(os.path.join(base, "abc.jsonl"))) self.assertFalse(os.path.exists(os.path.join(base, "abc.jsonl.live.jsonl"))) def test_subagent_subdir_is_never_linked(self): base = os.path.join(self.default_projects, "-home-user-Projects-overdeck") os.makedirs(base, exist_ok=True) wt = os.path.join(self.default_projects, "-home-user-Projects-overdeck--worktrees-foo") write_transcript(os.path.join(wt, "abc.jsonl"), "abc") write_transcript(os.path.join(wt, "subagents", "child.jsonl"), "child") conv.converge_homes([self.default_projects], dry_run=False) self.assertTrue(os.path.exists(os.path.join(base, "abc.jsonl"))) self.assertFalse(os.path.exists(os.path.join(base, "child.jsonl"))) if __name__ == "__main__": sys.exit(0 if unittest.main(exit=False).result.wasSuccessful() else 1)