Something went wrong. Try again.
paimon The best travel companion ever!
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391import jsonimport osimport tempfileimport unittestfrom pathlib import Pathfrom 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)])