"""Unit tests for the Claude Code session provider.""" import json import pytest from src.blocks import ( BLOCK_TYPE_COLLAPSED, BLOCK_TYPE_TEXT, BLOCK_TYPE_THINKING, BLOCK_TYPE_TOOL_RESULT, BLOCK_TYPE_TOOL_USE, render_blocks_to_markdown, ) from src.loss_report import LossReport from src.providers.claude_code import ClaudeCodeProvider def _write_session(tmp_path, records, project_dir="-home-jesse-myproj", name="abc-123"): proj = tmp_path / project_dir proj.mkdir(parents=True, exist_ok=True) f = proj / f"{name}.jsonl" f.write_text("\n".join(json.dumps(r) for r in records), encoding="utf-8") return f def _records(): """A representative session: title, harness noise, dialogue, tool traffic.""" return [ {"type": "ai-title", "aiTitle": "First title", "sessionId": "abc-123"}, { "type": "user", "cwd": "/home/jesse/myproj", "timestamp": "2026-05-01T10:00:00.000Z", "message": { "role": "user", "content": ( "Caveat: ignore" "/model" "Set model" "Please review my project." ), }, }, { "type": "assistant", "timestamp": "2026-05-01T10:00:05.000Z", "message": { "role": "assistant", "content": [ {"type": "thinking", "thinking": "Let me think about this."}, {"type": "text", "text": "I'll review it now."}, {"type": "tool_use", "id": "t1", "name": "Read", "input": {"file_path": "/x"}}, {"type": "tool_use", "id": "t2", "name": "Read", "input": {"file_path": "/y"}}, {"type": "tool_use", "id": "t3", "name": "Bash", "input": {"command": "ls"}}, ], }, }, { "type": "user", "timestamp": "2026-05-01T10:00:08.000Z", "message": { "role": "user", "content": [ {"type": "tool_result", "tool_use_id": "t1", "content": "file contents " * 50}, ], }, }, { "type": "assistant", "timestamp": "2026-05-01T10:00:20.000Z", "message": { "role": "assistant", "content": [{"type": "text", "text": "Here is my review: all good."}], }, }, # Harness records — skipped {"type": "file-history-snapshot", "snapshot": {"big": "blob"}}, {"type": "permission-mode", "mode": "plan"}, # Subagent and meta records — skipped { "type": "assistant", "isSidechain": True, "message": {"role": "assistant", "content": [{"type": "text", "text": "subagent noise"}]}, }, { "type": "user", "isMeta": True, "message": {"role": "user", "content": "meta noise"}, }, # Later title wins {"type": "ai-title", "aiTitle": "Review my project", "sessionId": "abc-123"}, ] class TestClaudeCodeProvider: def _provider(self, tmp_path, policy="placeholder"): return ClaudeCodeProvider(projects_dir=tmp_path, hidden_content=policy) def test_listing_metadata(self, tmp_path): _write_session(tmp_path, _records()) p = self._provider(tmp_path) convs = p.fetch_all_conversations() assert len(convs) == 1 c = convs[0] assert c["id"] == "abc-123" assert c["title"] == "Review my project" # last ai-title wins assert c["_project_name"] == "myproj" # from cwd basename assert c["created_at"] == "2026-05-01T10:00:00.000Z" assert c["updated_at"] # file mtime def test_empty_files_skipped(self, tmp_path): proj = tmp_path / "-home-x" proj.mkdir() (proj / "empty.jsonl").write_text("") p = self._provider(tmp_path) assert p.fetch_all_conversations() == [] def test_normalize_prose_only_default(self, tmp_path): _write_session(tmp_path, _records()) p = self._provider(tmp_path) p.fetch_all_conversations() raw = p.get_conversation("abc-123") result = p.normalize_conversation(raw) assert result["provider"] == "claude-code" assert result["title"] == "Review my project" assert result["project"] == "myproj" roles = [m["role"] for m in result["messages"]] assert roles == ["user", "assistant", "assistant"] # Harness tags stripped, dialogue kept user_text = result["messages"][0]["blocks"][0]["text"] assert user_text == "Please review my project." assert "Caveat" not in user_text # Tool activity collapsed into one grouped placeholder on the # assistant message that ran the tools (includes the tool_result # bytes from the following user record). first_assistant = result["messages"][1] types = [b["type"] for b in first_assistant["blocks"]] assert types == [BLOCK_TYPE_TEXT, BLOCK_TYPE_COLLAPSED] collapsed = first_assistant["blocks"][1] assert "3 calls" in collapsed["origin"] assert "Read ×2" in collapsed["origin"] assert "Bash ×1" in collapsed["origin"] assert collapsed["size_bytes"] > 500 # Thinking dropped without placeholder; sidechain/meta absent all_blocks = [b for m in result["messages"] for b in m["blocks"]] assert not any(b["type"] == BLOCK_TYPE_THINKING for b in all_blocks) assert not any( "subagent noise" in (b.get("text") or "") or "meta noise" in (b.get("text") or "") for b in all_blocks ) def test_collapsed_counted_in_loss_report(self, tmp_path): _write_session(tmp_path, _records()) p = self._provider(tmp_path) p.fetch_all_conversations() report = LossReport() p.normalize_conversation(p.get_conversation("abc-123"), report) assert report.collapsed["Read"] == 2 assert report.collapsed["Bash"] == 1 assert report.collapsed["tool_result"] == 1 assert report.collapsed["thinking"] == 1 assert report.collapsed_bytes > 500 def test_full_policy_keeps_everything(self, tmp_path): _write_session(tmp_path, _records()) p = self._provider(tmp_path, policy="full") p.fetch_all_conversations() result = p.normalize_conversation(p.get_conversation("abc-123")) all_blocks = [b for m in result["messages"] for b in m["blocks"]] types = {b["type"] for b in all_blocks} assert BLOCK_TYPE_THINKING in types assert BLOCK_TYPE_TOOL_USE in types assert BLOCK_TYPE_TOOL_RESULT in types assert BLOCK_TYPE_COLLAPSED not in types def test_updated_at_consistent_with_listing(self, tmp_path): """Listing and normalized updated_at must match or the cache would consider every session stale on every run (mtime vs last timestamp).""" _write_session(tmp_path, _records()) p = self._provider(tmp_path) listing = p.fetch_all_conversations()[0] normalized = p.normalize_conversation(p.get_conversation("abc-123")) assert normalized["updated_at"] == listing["updated_at"] def test_title_falls_back_to_first_prompt(self, tmp_path): records = [r for r in _records() if r.get("type") != "ai-title"] _write_session(tmp_path, records) p = self._provider(tmp_path) p.fetch_all_conversations() result = p.normalize_conversation(p.get_conversation("abc-123")) assert result["title"] == "Please review my project." def test_unknown_block_type_surfaces(self, tmp_path): records = [ { "type": "assistant", "timestamp": "2026-05-01T10:00:00.000Z", "cwd": "/home/jesse/myproj", "message": { "role": "assistant", "content": [ {"type": "text", "text": "hello"}, {"type": "future_block_xyz", "data": 1}, ], }, }, ] _write_session(tmp_path, records) p = self._provider(tmp_path) p.fetch_all_conversations() report = LossReport() result = p.normalize_conversation(p.get_conversation("abc-123"), report) assert report.unknown_blocks["claude-code.future_block_xyz"] == 1 rendered = render_blocks_to_markdown(result["messages"][0]["blocks"]) assert "Unsupported content" in rendered def test_collapsed_placeholder_renders_grouped_line(self, tmp_path): _write_session(tmp_path, _records()) p = self._provider(tmp_path) p.fetch_all_conversations() result = p.normalize_conversation(p.get_conversation("abc-123")) rendered = render_blocks_to_markdown(result["messages"][1]["blocks"]) assert "> 🔧 **Tool output** — `3 calls: Read ×2, Bash ×1`" in rendered assert "omitted (EXPORTER_HIDDEN_CONTENT=full to keep)" in rendered # --------------------------------------------------------------------------- # Subagent folding (Part B) # --------------------------------------------------------------------------- def _write_subagent( session_file, tool_use_id, records, agent_type="Explore", description="do a thing", name="agent-x", ): """Write a /subagents/.jsonl + .meta.json sidecar.""" subdir = session_file.parent / session_file.stem / "subagents" subdir.mkdir(parents=True, exist_ok=True) (subdir / f"{name}.jsonl").write_text( "\n".join(json.dumps(r) for r in records), encoding="utf-8" ) (subdir / f"{name}.meta.json").write_text( json.dumps({ "agentType": agent_type, "description": description, "toolUseId": tool_use_id, }), encoding="utf-8", ) def _parent_with_task(tool_use_id="toolu_sub1"): return [ {"type": "ai-title", "aiTitle": "Parent session"}, { "type": "user", "cwd": "/home/jesse/myproj", "timestamp": "2026-05-01T10:00:00.000Z", "message": {"role": "user", "content": "delegate the research"}, }, { "type": "assistant", "timestamp": "2026-05-01T10:00:05.000Z", "message": {"role": "assistant", "content": [ {"type": "text", "text": "I'll delegate this."}, {"type": "tool_use", "id": tool_use_id, "name": "Task", "input": {"description": "research"}}, ]}, }, ] def _subagent_records(): return [ { "type": "user", "isSidechain": True, "timestamp": "2026-05-01T10:00:06.000Z", "message": {"role": "user", "content": "Go research X"}, }, { "type": "assistant", "isSidechain": True, "timestamp": "2026-05-01T10:00:07.000Z", "message": {"role": "assistant", "content": [ {"type": "text", "text": "Here is the research result."}, {"type": "tool_use", "id": "st1", "name": "Grep", "input": {"pattern": "x"}}, ]}, }, ] class TestSubagentFold: def _provider(self, tmp_path, policy="placeholder"): return ClaudeCodeProvider(projects_dir=tmp_path, hidden_content=policy) def _normalize(self, tmp_path, policy="placeholder"): f = _write_session(tmp_path, _parent_with_task(), name="parent-1") _write_subagent( f, "toolu_sub1", _subagent_records(), agent_type="Explore", description="research X", ) p = self._provider(tmp_path, policy) p.fetch_all_conversations() return p.normalize_conversation(p.get_conversation("parent-1")) def test_subagent_folded_into_parent(self, tmp_path): conv = self._normalize(tmp_path) blocks = [b for m in conv["messages"] for b in m["blocks"]] subs = [b for b in blocks if b["type"] == "subagent"] assert len(subs) == 1 assert subs[0]["agent_type"] == "Explore" assert subs[0]["description"] == "research X" inner_text = [ bb.get("text") for m in subs[0]["messages"] for bb in m["blocks"] ] assert "Go research X" in inner_text assert "Here is the research result." in inner_text def test_subagent_tools_collapsed(self, tmp_path): conv = self._normalize(tmp_path) sub = next( b for m in conv["messages"] for b in m["blocks"] if b["type"] == "subagent" ) inner_blocks = [bb for m in sub["messages"] for bb in m["blocks"]] assert any(b["type"] == BLOCK_TYPE_COLLAPSED for b in inner_blocks) assert not any(b["type"] == BLOCK_TYPE_TOOL_USE for b in inner_blocks) collapsed = next(b for b in inner_blocks if b["type"] == BLOCK_TYPE_COLLAPSED) assert "Grep" in collapsed["origin"] def test_subagent_not_listed_as_conversation(self, tmp_path): f = _write_session(tmp_path, _parent_with_task(), name="parent-1") _write_subagent(f, "toolu_sub1", _subagent_records()) p = self._provider(tmp_path) convs = p.fetch_all_conversations() assert [c["id"] for c in convs] == ["parent-1"] def test_renders_as_details_block(self, tmp_path): conv = self._normalize(tmp_path) spawning = conv["messages"][1] # the assistant turn with the Task call rendered = render_blocks_to_markdown(spawning["blocks"]) assert "
" in rendered assert "🤖 Subagent: Explore — research X" in rendered assert "
" in rendered assert "Here is the research result." in rendered def test_main_pass_still_skips_stray_sidechain(self, tmp_path): # A sidechain record with no matching subagent file must NOT leak into # the top-level dialogue. records = _parent_with_task() + [ {"type": "assistant", "isSidechain": True, "message": {"role": "assistant", "content": [{"type": "text", "text": "stray sidechain"}]}}, ] f = _write_session(tmp_path, records, name="parent-2") _write_subagent(f, "toolu_sub1", _subagent_records()) p = self._provider(tmp_path) p.fetch_all_conversations() conv = p.normalize_conversation(p.get_conversation("parent-2")) top_text = [ b.get("text") for m in conv["messages"] for b in m["blocks"] ] assert "stray sidechain" not in top_text # --------------------------------------------------------------------------- # Repo tags in title (Part C) # --------------------------------------------------------------------------- class TestRepoTags: @staticmethod def _mkrepo(base, name): """Create a git repo dir (bare .git marker) and return its path.""" (base / name / ".git").mkdir(parents=True, exist_ok=True) return base / name def _title(self, tmp_path, content): (tmp_path / "ws").mkdir(exist_ok=True) records = [ {"type": "ai-title", "aiTitle": "Work session"}, {"type": "user", "cwd": str(tmp_path / "ws"), "timestamp": "2026-05-01T10:00:00.000Z", "message": {"role": "user", "content": "go"}}, {"type": "assistant", "message": {"role": "assistant", "content": content}}, ] _write_session(tmp_path, records, name="ws-1") p = ClaudeCodeProvider(projects_dir=tmp_path) p.fetch_all_conversations() return p.normalize_conversation(p.get_conversation("ws-1"))["title"] @staticmethod def _read(path): return {"type": "tool_use", "id": str(path), "name": "Read", "input": {"file_path": str(path)}} def test_repos_by_git_root_ordered_by_frequency(self, tmp_path): a = self._mkrepo(tmp_path, "repo-a") b = self._mkrepo(tmp_path, "repo-b") content = [ self._read(a / "src/x.py"), self._read(a / "src/y.py"), self._read(b / "z.py"), # a file not inside any git repo → excluded self._read(tmp_path / "CLAUDE.md"), ] assert self._title(tmp_path, content) == "Work session [repo-a, repo-b]" def test_cross_workspace_repo_tagged(self, tmp_path): # Launched in ws/, but touched a repo in a sibling tree — still tagged. other = self._mkrepo(tmp_path / "elsewhere", "faraway-repo") content = [self._read(other / "deep/nested/file.py")] assert self._title(tmp_path, content) == "Work session [faraway-repo]" def test_no_repos_no_bracket(self, tmp_path): content = [self._read(tmp_path / "loose/file.txt")] # no .git anywhere assert self._title(tmp_path, content) == "Work session" def test_cap_three_with_ellipsis(self, tmp_path): content = [ self._read(self._mkrepo(tmp_path, f"repo-{c}") / "f/x.py") for c in "abcd" ] title = self._title(tmp_path, content) assert title.startswith("Work session [repo-a, repo-b, repo-c, …]") assert "repo-d" not in title def test_paths_inside_bash_commands_count(self, tmp_path): x = self._mkrepo(tmp_path, "repo-x") content = [ {"type": "tool_use", "id": "a", "name": "Bash", "input": {"command": f"cat {x / 'main.go'}"}}, ] assert self._title(tmp_path, content) == "Work session [repo-x]" def test_ignore_list_via_env(self, tmp_path, monkeypatch): a = self._mkrepo(tmp_path, "repo-a") b = self._mkrepo(tmp_path, "repo-b") monkeypatch.setenv("CLAUDE_CODE_REPO_TAG_IGNORE", "repo-b, notes") content = [self._read(a / "x.py"), self._read(b / "y.py")] assert self._title(tmp_path, content) == "Work session [repo-a]" # --------------------------------------------------------------------------- # Multiple projects roots (Part D) # --------------------------------------------------------------------------- class TestMultiRoot: def _titled(self, aititle, cwd="/home/jesse/p"): return [ {"type": "ai-title", "aiTitle": aititle}, {"type": "user", "cwd": cwd, "timestamp": "2026-05-01T10:00:00.000Z", "message": {"role": "user", "content": "hi"}}, ] def test_two_roots_merged(self, tmp_path): r1, r2 = tmp_path / "root1", tmp_path / "root2" _write_session(r1, self._titled("From root1"), name="s1") _write_session(r2, self._titled("From root2"), name="s2") p = ClaudeCodeProvider(projects_dir=[r1, r2]) ids = sorted(c["id"] for c in p.fetch_all_conversations()) assert ids == ["s1", "s2"] def test_duplicate_uuid_newer_wins(self, tmp_path): import os import time r1, r2 = tmp_path / "root1", tmp_path / "root2" _write_session(r1, self._titled("Old copy"), name="dup") f2 = _write_session(r2, self._titled("New copy"), name="dup") os.utime(f2, (time.time() + 10, time.time() + 10)) p = ClaudeCodeProvider(projects_dir=[r1, r2]) convs = p.fetch_all_conversations() assert len(convs) == 1 assert convs[0]["title"] == "New copy" assert p._path_map["dup"] == f2 def test_single_path_backward_compatible(self, tmp_path): _write_session(tmp_path, _records()) p = ClaudeCodeProvider(projects_dir=tmp_path) assert len(p.fetch_all_conversations()) == 1 def test_env_pathsep_list(self, tmp_path, monkeypatch): import os from src.providers.claude_code import resolve_roots r1, r2 = tmp_path / "root1", tmp_path / "root2" _write_session(r1, self._titled("From root1"), name="s1") _write_session(r2, self._titled("From root2"), name="s2") monkeypatch.setenv("CLAUDE_CODE_DIR", f"{r1}{os.pathsep}{r2}") assert r1 in resolve_roots() and r2 in resolve_roots() p = ClaudeCodeProvider() assert len(p.fetch_all_conversations()) == 2 def test_config_dir_projects_included(self, tmp_path, monkeypatch): from src.providers.claude_code import resolve_roots monkeypatch.delenv("CLAUDE_CODE_DIR", raising=False) monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(tmp_path / "cfg")) roots = resolve_roots() assert (tmp_path / "cfg" / "projects") in roots