import json import os import tempfile import unittest from pathlib import Path from unittest.mock import patch from pydantic_ai.messages import ( ModelRequest, ModelResponse, TextPart, ToolCallPart, ToolReturnPart, UserPromptPart, ) from paimon.session import ( SESSION_FORMAT_VERSION, Session, SessionBusyError, SessionError, _project_dir, is_shell_message, is_synthetic_user_text, shell_message, shell_text, ) class SessionScanTestCase(unittest.TestCase): """Sessions created via the real API in an isolated PAIMON_DATA_HOME.""" def setUp(self) -> None: tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) env = patch.dict("os.environ", {"PAIMON_DATA_HOME": tmp.name}) env.start() self.addCleanup(env.stop) self.cwd = Path(tmp.name) / "project" self.cwd.mkdir() def _session_with_message(self, content: str, mtime: float) -> Session: session = Session.create(self.cwd) session.append_message(ModelRequest(parts=[UserPromptPart(content=content)])) os.utime(session.path, (mtime, mtime)) return session class ListTest(SessionScanTestCase): def test_newest_first_and_empty_sessions_excluded(self) -> None: old = self._session_with_message("old", mtime=1_000) new = self._session_with_message("new", mtime=2_000) empty = Session.create(self.cwd) listed = Session.list(self.cwd) self.assertEqual([session.id for session in listed], [new.id, old.id]) self.assertNotIn(empty.id, [session.id for session in listed]) def test_invalid_files_are_skipped(self) -> None: session = self._session_with_message("hi", mtime=1_000) (_project_dir(self.cwd) / "garbage.jsonl").write_text("not json\n") (_project_dir(self.cwd) / "missing-id.jsonl").write_text( '{"type": "session"}\n' ) self.assertEqual([s.id for s in Session.list(self.cwd)], [session.id]) def test_no_project_dir_gives_empty_list(self) -> None: self.assertEqual(Session.list(self.cwd / "elsewhere"), []) class PreviewTest(SessionScanTestCase): def test_created_at_from_header(self) -> None: session = Session.create(self.cwd) created = session.created_at() self.assertIsNotNone(created) self.assertIn("T", created) def test_first_user_text_skips_assistant_messages(self) -> None: session = Session.create(self.cwd) session.append_message(ModelResponse(parts=[TextPart(content="hello!")])) session.append_message(ModelRequest(parts=[UserPromptPart(content="fix the bug")])) session.append_message(ModelRequest(parts=[UserPromptPart(content="second")])) self.assertEqual(session.first_user_text(), "fix the bug") def test_first_user_text_none_for_empty_session(self) -> None: self.assertIsNone(Session.create(self.cwd).first_user_text()) def test_first_user_text_skips_a_shell_run(self) -> None: """A "!" command is paimon's message, not the user's, so it can be neither the session's title nor its preview.""" session = Session.create(self.cwd) session.append_message(shell_message("ls", "a.txt")) session.append_message(ModelRequest(parts=[UserPromptPart(content="fix the bug")])) self.assertEqual(session.first_user_text(), "fix the bug") class ShellMessageTest(unittest.TestCase): def test_command_and_output_survive_the_round_trip(self) -> None: message = shell_message("git log --oneline -n 2", "1e34a91 x\n1a3bbc1 y") self.assertTrue(is_shell_message(message)) self.assertEqual(shell_text(message), ("git log --oneline -n 2", "1e34a91 x\n1a3bbc1 y")) def test_a_command_with_no_output_round_trips_too(self) -> None: self.assertEqual(shell_text(shell_message("true", "")), ("true", "")) def test_an_ordinary_user_message_is_not_one(self) -> None: message = ModelRequest(parts=[UserPromptPart(content="$ ls")]) self.assertFalse(is_shell_message(message)) self.assertFalse(is_synthetic_user_text("$ ls")) class EntriesTest(SessionScanTestCase): def test_seq_is_the_physical_line_number_and_corrupt_lines_hold_their_place(self) -> None: session = Session.create(self.cwd) session.append_message(ModelRequest(parts=[UserPromptPart(content="hi")])) with session.path.open("a", encoding="utf-8") as file: file.write("not json\n") session.append_message(ModelResponse(parts=[TextPart(content="hello!")])) entries = session.entries() self.assertEqual([seq for seq, _ in entries], [1, 2, 3, 4]) self.assertEqual(entries[0][1]["type"], "session") self.assertIsNone(entries[2][1], "the corrupt line stays as a placeholder") self.assertEqual(entries[3][1]["type"], "message") def test_unreadable_file_raises_instead_of_yielding_nothing(self) -> None: session = Session.create(self.cwd) session.path.unlink() with self.assertRaises(OSError): session.entries() # One complete session log frozen as text, exactly as a v1 Paimon wrote it. # Never regenerate this from live code: its whole point is to prove that a # dependency upgrade still reads what was written before the upgrade. _V1_FIXTURE = "\n".join([ '{"type": "session", "version": 1, "id": "fix11111-0000-0000-0000-000000000000", ' '"cwd": "/tmp/project", "created_at": "2026-08-18T00:00:00+00:00"}', '{"type": "system_prompt", "timestamp": "2026-08-18T00:00:00+00:00", "content": "snapshot"}', '{"type": "message", "id": "m1", "timestamp": "2026-08-18T00:00:01+00:00", "message": ' '{"parts": [{"content": "list the files", "timestamp": "2026-08-18T00:00:01Z", ' '"part_kind": "user-prompt"}], "kind": "request"}}', '{"type": "message", "id": "m2", "timestamp": "2026-08-18T00:00:02+00:00", "message": ' '{"parts": [{"content": "Listing.", "part_kind": "text"}, {"tool_name": "shell", ' '"args": {"command": "ls"}, "tool_call_id": "c1", "part_kind": "tool-call"}], ' '"model_name": "glm-5.2", "timestamp": "2026-08-18T00:00:02Z", "kind": "response", ' '"provider_name": "zai"}}', '{"type": "message", "id": "m3", "timestamp": "2026-08-18T00:00:03+00:00", "message": ' '{"parts": [{"tool_name": "shell", "content": "a.py", "tool_call_id": "c1", ' '"timestamp": "2026-08-18T00:00:03Z", "outcome": "success", ' '"part_kind": "tool-return"}], "kind": "request"}}', '{"type": "turn_end", "timestamp": "2026-08-18T00:00:04+00:00", "outcome": "success"}', ]) + "\n" class FormatVersionTest(SessionScanTestCase): """SESSION-6: the log format is versioned, a newer format refuses to resume, and a frozen v1 log keeps loading across dependency upgrades.""" def test_new_sessions_declare_the_current_format(self) -> None: session = Session.create(self.cwd) header = json.loads(session.path.read_text().splitlines()[0]) self.assertEqual(header["version"], SESSION_FORMAT_VERSION) self.assertEqual(session.format_version(), SESSION_FORMAT_VERSION) session.require_supported_format() def test_forks_carry_the_current_format(self) -> None: session = Session.create(self.cwd) session.append_message(ModelRequest(parts=[UserPromptPart(content="hi")])) fork = session.fork() header = json.loads(fork.path.read_text().splitlines()[0]) self.assertEqual(header["version"], SESSION_FORMAT_VERSION) def test_pre_versioning_logs_read_as_format_1(self) -> None: path = _project_dir(self.cwd) / "old.jsonl" path.parent.mkdir(parents=True, exist_ok=True) path.write_text('{"type": "session", "id": "old11111", "cwd": "x"}\n') session = Session(path, "old11111", self.cwd) self.assertEqual(session.format_version(), 1) session.require_supported_format() def test_a_newer_format_refuses_to_resume(self) -> None: path = _project_dir(self.cwd) / "future.jsonl" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps({ "type": "session", "version": SESSION_FORMAT_VERSION + 1, "id": "fut11111", "cwd": "x"}) + "\n") session = Session(path, "fut11111", self.cwd) with self.assertRaises(SessionError) as caught: session.require_supported_format() self.assertIn("newer", str(caught.exception)) def test_unparseable_messages_raise_a_session_error(self) -> None: session = Session.create(self.cwd) session.append({"type": "message", "id": "bad", "message": {"kind": "nonsense"}}) with self.assertRaises(SessionError) as caught: session.messages() self.assertIn("cannot parse", str(caught.exception)) def test_the_frozen_v1_fixture_still_loads(self) -> None: path = _project_dir(self.cwd) / "fixture.jsonl" path.parent.mkdir(parents=True, exist_ok=True) path.write_text(_V1_FIXTURE) session = Session(path, "fix11111-0000-0000-0000-000000000000", self.cwd) session.require_supported_format() self.assertEqual(session.system_prompt(), "snapshot") messages = session.messages() self.assertEqual(len(messages), 3) self.assertIsInstance(messages[0], ModelRequest) self.assertIsInstance(messages[1], ModelResponse) self.assertEqual(messages[2].parts[0].content, "a.py") class OrphanToolCallTest(SessionScanTestCase): """SESSION-4: a crash between the tool-call response and its pre-seeded results must not leave a history providers reject.""" def _tool_call_response(self, *ids: str) -> ModelResponse: return ModelResponse(parts=[ ToolCallPart(tool_name="shell", args={"command": "ls"}, tool_call_id=call_id) for call_id in ids ]) def test_a_log_ending_on_a_tool_call_gets_a_synthesized_result(self) -> None: session = Session.create(self.cwd) session.append_message(ModelRequest(parts=[UserPromptPart(content="go")])) session.append_message(self._tool_call_response("c1", "c2")) messages = session.messages() last = messages[-1] self.assertIsInstance(last, ModelRequest) returns = [p for p in last.parts if isinstance(p, ToolReturnPart)] self.assertEqual([r.tool_call_id for r in returns], ["c1", "c2"]) self.assertIn("Interrupted", returns[0].content) # Deterministic: a second load produces the same repair shape. again = session.messages() self.assertEqual(len(again), len(messages)) self.assertEqual([p.tool_call_id for p in again[-1].parts], ["c1", "c2"]) def test_a_user_prompt_after_the_orphan_stays_after_the_repair(self) -> None: session = Session.create(self.cwd) session.append_message(self._tool_call_response("c1")) session.append_message(ModelRequest(parts=[UserPromptPart(content="resumed")])) messages = session.messages() self.assertIsInstance(messages[1], ModelRequest) self.assertEqual(messages[1].parts[0].tool_call_id, "c1") self.assertEqual(messages[2].parts[0].content, "resumed") def test_a_partially_answered_batch_is_completed_in_place(self) -> None: session = Session.create(self.cwd) session.append_message(self._tool_call_response("c1", "c2")) session.append_message(ModelRequest(parts=[ ToolReturnPart(tool_name="shell", content="ok", tool_call_id="c1"), ])) messages = session.messages() self.assertEqual(len(messages), 2, "no extra request is inserted") answered = {p.tool_call_id: p.content for p in messages[1].parts if isinstance(p, ToolReturnPart)} self.assertEqual(answered["c1"], "ok") self.assertIn("Interrupted", answered["c2"]) def test_a_complete_batch_is_untouched(self) -> None: session = Session.create(self.cwd) session.append_message(self._tool_call_response("c1")) session.append_message(ModelRequest(parts=[ ToolReturnPart(tool_name="shell", content="fine", tool_call_id="c1"), ])) messages = session.messages() self.assertEqual(len(messages), 2) self.assertEqual([p.content for p in messages[1].parts], ["fine"]) class ForkTest(SessionScanTestCase): def test_fork_copies_log_under_a_fresh_id(self) -> None: source = Session.create(self.cwd) source.append_system_prompt("be helpful") source.append_message(ModelRequest(parts=[UserPromptPart(content="hi")])) source.append_message(ModelResponse(parts=[TextPart(content="hello!")])) fork = source.fork() self.assertNotEqual(fork.id, source.id) self.assertNotEqual(fork.path, source.path) self.assertEqual(fork.path.parent, source.path.parent) header = Session._read_records(fork.path)[0] self.assertEqual(header["type"], "session") self.assertEqual(header["id"], fork.id) self.assertEqual(fork.system_prompt(), "be helpful") self.assertEqual(fork.messages(), source.messages()) def test_fork_leaves_the_source_untouched(self) -> None: source = Session.create(self.cwd) source.append_message(ModelRequest(parts=[UserPromptPart(content="hi")])) before = source.path.read_text(encoding="utf-8") source.fork() self.assertEqual(source.path.read_text(encoding="utf-8"), before) def test_fork_keeps_line_numbers_aligned(self) -> None: source = Session.create(self.cwd) source.append_message(ModelRequest(parts=[UserPromptPart(content="hi")])) with source.path.open("a", encoding="utf-8") as file: file.write("not json\n") source.append_message(ModelResponse(parts=[TextPart(content="hello!")])) fork = source.fork() source_lines = source.path.read_text(encoding="utf-8").splitlines() fork_lines = fork.path.read_text(encoding="utf-8").splitlines() self.assertEqual(fork_lines[1:], source_lines[1:]) def test_fork_preserves_compaction_checkpoints(self) -> None: source = Session.create(self.cwd) source.append_message(ModelRequest(parts=[UserPromptPart(content="hi")])) kept = ModelResponse(parts=[TextPart(content="hello!")]) source.append_compaction("earlier talk", [kept], tokens_before=100) # The summary message is synthesized on replay with a fresh timestamp, # so compare part contents rather than whole messages. contents = [[part.content for part in message.parts] for message in source.fork().messages()] self.assertEqual(contents, [[part.content for part in message.parts] for message in source.messages()]) self.assertIn("earlier talk", contents[0][0]) class LockTest(SessionScanTestCase): """Lock mechanics live in test_lockfile; this covers the session semantics.""" def test_lock_raises_busy_when_another_process_holds_it(self) -> None: session = Session.create(self.cwd) with patch("paimon.session.lockfile.acquire", return_value=False): with self.assertRaisesRegex(SessionBusyError, "another process"): session.lock() def test_lock_and_unlock_round_trip(self) -> None: session = Session.create(self.cwd) session.lock() session.unlock() session.unlock() # extra unlock is a no-op def test_a_session_already_open_here_is_refused_too(self) -> None: """The lock refcounts per process, so two panes would both be let in — and their histories would interleave into one append-only log.""" session = Session.create(self.cwd) session.lock() self.addCleanup(session.unlock) with self.assertRaisesRegex(SessionBusyError, "already open"): Session(session.path, session.id, self.cwd).lock() class ChildSessionTest(SessionScanTestCase): """Subagent sessions share the project directory but not the listings.""" def test_children_are_hidden_unless_asked_for(self) -> None: mine = self._session_with_message("mine", mtime=1_000) child = Session.create(self.cwd, parent_id=mine.id) child.append_message(ModelRequest(parts=[UserPromptPart(content="theirs")])) self.assertEqual([session.id for session in Session.list(self.cwd)], [mine.id]) listed = Session.list(self.cwd, include_children=True) self.assertEqual(sorted(session.id for session in listed), sorted([mine.id, child.id])) self.assertEqual(next(s for s in listed if s.id == child.id).parent_id, mine.id) def test_a_fork_of_a_child_is_still_a_child(self) -> None: parent = Session.create(self.cwd) child = Session.create(self.cwd, parent_id=parent.id) child.append_message(ModelRequest(parts=[UserPromptPart(content="theirs")])) forked = child.fork() self.assertEqual(forked.parent_id, parent.id) self.assertNotIn(forked.id, [session.id for session in Session.list(self.cwd)])