| 1 | from __future__ import annotations |
| 2 | |
| 3 | import importlib.util |
| 4 | import io |
| 5 | import json |
| 6 | import sqlite3 |
| 7 | import sys |
| 8 | import tempfile |
| 9 | import unittest |
| 10 | import zipfile |
| 11 | from pathlib import Path |
| 12 | |
| 13 | |
| 14 | ROOT = Path(__file__).resolve().parents[1] |
| 15 | SPEC = importlib.util.spec_from_file_location("_migrate_agents_core", ROOT / "helpers" / "migration.py") |
| 16 | assert SPEC and SPEC.loader |
| 17 | migration = importlib.util.module_from_spec(SPEC) |
| 18 | sys.modules[SPEC.name] = migration |
| 19 | SPEC.loader.exec_module(migration) |
| 20 | |
| 21 | |
| 22 | def jsonl(*rows) -> bytes: |
| 23 | return ("\n".join(json.dumps(row) for row in rows) + "\n").encode() |
| 24 | |
| 25 | |
| 26 | class MigrationTests(unittest.TestCase): |
| 27 | def test_codex_uses_public_events_tools_and_excludes_reasoning(self): |
| 28 | upload = migration.Upload( |
| 29 | "rollout-test.jsonl", |
| 30 | jsonl( |
| 31 | {"type": "session_meta", "payload": {"id": "codex-1", "cwd": "/work"}}, |
| 32 | {"type": "event_msg", "timestamp": "2026-01-01T00:00:00Z", "payload": {"type": "user_message", "message": "Fix it"}}, |
| 33 | {"type": "event_msg", "timestamp": "2026-01-01T00:00:01Z", "payload": {"type": "agent_message", "phase": "commentary", "message": "Checking"}}, |
| 34 | {"type": "response_item", "timestamp": "2026-01-01T00:00:02Z", "payload": {"type": "function_call", "call_id": "c1", "name": "shell", "arguments": "{\"api_key\":\"secret-value\"}"}}, |
| 35 | {"type": "response_item", "timestamp": "2026-01-01T00:00:03Z", "payload": {"type": "function_call_output", "call_id": "c1", "output": "done"}}, |
| 36 | {"type": "event_msg", "timestamp": "2026-01-01T00:00:04Z", "payload": {"type": "agent_message", "phase": "final_answer", "message": "Fixed"}}, |
| 37 | {"type": "response_item", "payload": {"type": "reasoning", "text": "hidden"}}, |
| 38 | ), |
| 39 | ) |
| 40 | bundle = migration.parse_bundle("codex", [upload]) |
| 41 | self.assertEqual(bundle.summary()["chats"], 1) |
| 42 | self.assertEqual([event.kind for event in bundle.conversations[0].events], ["user", "tool", "assistant"]) |
| 43 | self.assertEqual(bundle.conversations[0].events[1].tool_result, "done") |
| 44 | self.assertEqual(bundle.conversations[0].events[1].tool_args["api_key"], "[REDACTED]") |
| 45 | self.assertNotIn('"text": "hidden"', json.dumps(migration.build_a0_chat(bundle.conversations[0], "codex"))) |
| 46 | |
| 47 | def test_claude_ignores_thinking_and_sidechains(self): |
| 48 | upload = migration.Upload( |
| 49 | "project/session.jsonl", |
| 50 | jsonl( |
| 51 | {"type": "user", "sessionId": "claude-1", "timestamp": "2026-01-01T00:00:00Z", "message": {"role": "user", "content": [{"type": "text", "text": "Hello"}]}}, |
| 52 | {"type": "assistant", "sessionId": "claude-1", "timestamp": "2026-01-01T00:00:01Z", "message": {"role": "assistant", "content": [{"type": "thinking", "thinking": "private"}, {"type": "text", "text": "Hi"}]}}, |
| 53 | {"type": "assistant", "isSidechain": True, "sessionId": "claude-1", "message": {"role": "assistant", "content": [{"type": "text", "text": "branch"}]}}, |
| 54 | ), |
| 55 | ) |
| 56 | bundle = migration.parse_bundle("claude", [upload]) |
| 57 | text = json.dumps(migration.build_a0_chat(bundle.conversations[0], "claude")) |
| 58 | self.assertIn("Hello", text) |
| 59 | self.assertIn("Hi", text) |
| 60 | self.assertNotIn("private", text) |
| 61 | self.assertNotIn("branch", text) |
| 62 | |
| 63 | def test_opencode_export_keeps_text_and_completed_tool(self): |
| 64 | data = { |
| 65 | "info": {"id": "ses_1", "title": "Ship it", "directory": "/repo"}, |
| 66 | "messages": [ |
| 67 | {"info": {"role": "user", "time": {"created": 1000}}, "parts": [{"type": "text", "text": "Build"}]}, |
| 68 | {"info": {"role": "assistant", "time": {"created": 2000}}, "parts": [{"type": "text", "text": "Done"}, {"type": "tool", "tool": "bash", "state": {"status": "completed", "input": {"command": "true"}, "output": "ok"}}]}, |
| 69 | ], |
| 70 | } |
| 71 | bundle = migration.parse_bundle("opencode", [migration.Upload("session.json", json.dumps(data).encode())]) |
| 72 | self.assertEqual(bundle.conversations[0].title, "Ship it") |
| 73 | self.assertEqual([event.kind for event in bundle.conversations[0].events], ["user", "assistant", "tool"]) |
| 74 | |
| 75 | def test_hermes_jsonl_and_openclaw_jsonl(self): |
| 76 | hermes = migration.parse_bundle( |
| 77 | "hermes", |
| 78 | [migration.Upload("backup.jsonl", jsonl({"id": "h1", "title": "Hermes", "messages": [{"role": "user", "content": "One"}, {"role": "assistant", "content": "Two"}]}))], |
| 79 | ) |
| 80 | claw = migration.parse_bundle( |
| 81 | "openclaw", |
| 82 | [migration.Upload("trace.jsonl", jsonl({"type": "session", "id": "o1"}, {"type": "message", "message": {"role": "user", "content": "One"}}, {"type": "message", "message": {"role": "assistant", "content": "Two"}}))], |
| 83 | ) |
| 84 | self.assertEqual(hermes.summary()["chats"], 1) |
| 85 | self.assertEqual(claw.summary()["chats"], 1) |
| 86 | |
| 87 | def test_hermes_sqlite(self): |
| 88 | with tempfile.NamedTemporaryFile(suffix=".db") as handle: |
| 89 | db = sqlite3.connect(handle.name) |
| 90 | db.executescript( |
| 91 | "CREATE TABLE sessions (id TEXT, source TEXT, model TEXT, title TEXT, cwd TEXT, started_at REAL);" |
| 92 | "CREATE TABLE messages (id INTEGER, session_id TEXT, role TEXT, content TEXT, timestamp REAL, tool_calls TEXT, tool_name TEXT, tool_call_id TEXT);" |
| 93 | "INSERT INTO sessions VALUES ('h1','cli','model','SQLite chat','/repo',1);" |
| 94 | "INSERT INTO messages VALUES (1,'h1','user','Hello',1,NULL,NULL,NULL);" |
| 95 | "INSERT INTO messages VALUES (2,'h1','assistant','Hi',2,NULL,NULL,NULL);" |
| 96 | ) |
| 97 | db.commit() |
| 98 | data = Path(handle.name).read_bytes() |
| 99 | db.close() |
| 100 | bundle = migration.parse_bundle("hermes", [migration.Upload("state.db", data)]) |
| 101 | self.assertEqual(bundle.conversations[0].title, "SQLite chat") |
| 102 | |
| 103 | def test_openclaw_sqlite(self): |
| 104 | with tempfile.NamedTemporaryFile(suffix=".sqlite") as handle: |
| 105 | db = sqlite3.connect(handle.name) |
| 106 | db.executescript( |
| 107 | "CREATE TABLE session_windows (session_id TEXT, session_key TEXT, created_at INTEGER, display_name TEXT, channel TEXT, model TEXT);" |
| 108 | "CREATE TABLE transcript_events (session_id TEXT, seq INTEGER, event_json TEXT, created_at INTEGER);" |
| 109 | "INSERT INTO session_windows VALUES ('o1','agent:main:main',1,'Claw chat','web','model');" |
| 110 | ) |
| 111 | event = json.dumps({"type": "message", "message": {"role": "user", "content": "Hello"}}) |
| 112 | db.execute("INSERT INTO transcript_events VALUES ('o1',1,?,1)", (event,)) |
| 113 | db.commit() |
| 114 | data = Path(handle.name).read_bytes() |
| 115 | db.close() |
| 116 | bundle = migration.parse_bundle("openclaw", [migration.Upload("openclaw-agent.sqlite", data)]) |
| 117 | self.assertEqual(bundle.conversations[0].title, "Claw chat") |
| 118 | |
| 119 | def test_discovers_knowledge_and_complete_skill_folder(self): |
| 120 | bundle = migration.parse_bundle( |
| 121 | "claude", |
| 122 | [ |
| 123 | migration.Upload("project/memory/MEMORY.md", b"Remember api_key=secret"), |
| 124 | migration.Upload("project/CLAUDE.md", b"Remember password=secret"), |
| 125 | migration.Upload("project/skills/release/SKILL.md", b"---\nname: release\n---"), |
| 126 | migration.Upload("project/skills/release/scripts/run.py", b"API_KEY=secret"), |
| 127 | migration.Upload("project/skills/release/.env", b"API_KEY=never"), |
| 128 | ], |
| 129 | ) |
| 130 | self.assertEqual(bundle.summary()["memories"], 1) |
| 131 | self.assertEqual(bundle.summary()["instructions"], 1) |
| 132 | self.assertEqual(bundle.summary()["knowledge"], 2) |
| 133 | self.assertEqual(bundle.summary()["skills"], 1) |
| 134 | self.assertIn(b"[REDACTED]", bundle.memories[0].data) |
| 135 | self.assertIn(b"[REDACTED]", bundle.instructions[0].data) |
| 136 | self.assertEqual(len(next(iter(bundle.skills.values()))), 2) |
| 137 | self.assertIn(b"[REDACTED]", bundle.skills["release"][1].data) |
| 138 | self.assertEqual(bundle.summary()["excluded"], 1) |
| 139 | |
| 140 | def test_discovers_projects_from_retained_workspace_context(self): |
| 141 | data = { |
| 142 | "info": {"id": "ses_1", "title": "Ship it", "directory": "/work/acme"}, |
| 143 | "messages": [ |
| 144 | {"info": {"role": "user"}, "parts": [{"type": "text", "text": "Build"}]}, |
| 145 | ], |
| 146 | } |
| 147 | bundle = migration.parse_bundle( |
| 148 | "opencode", |
| 149 | [migration.Upload("session.json", json.dumps(data).encode())], |
| 150 | ) |
| 151 | self.assertEqual(bundle.summary()["projects"], 1) |
| 152 | self.assertEqual(bundle.projects[0].title, "acme") |
| 153 | self.assertEqual(bundle.projects[0].path, "/work/acme") |
| 154 | self.assertEqual(bundle.projects[0].conversation_ids, ["ses_1"]) |
| 155 | |
| 156 | def test_rejects_archive_traversal(self): |
| 157 | stream = io.BytesIO() |
| 158 | with zipfile.ZipFile(stream, "w") as archive: |
| 159 | archive.writestr("../escape.json", "{}") |
| 160 | with self.assertRaisesRegex(ValueError, "Unsafe archive path"): |
| 161 | migration.parse_bundle("opencode", [migration.Upload("bad.zip", stream.getvalue())]) |
| 162 | |
| 163 | def test_a0_chat_log_and_history_sequences_are_valid(self): |
| 164 | conversation = migration.Conversation( |
| 165 | "id", |
| 166 | "Title", |
| 167 | [migration.Event("user", "Hi", 1), migration.Event("assistant", "Hello", 2)], |
| 168 | ) |
| 169 | chat = migration.build_a0_chat(conversation, "test") |
| 170 | history = json.loads(chat["agents"][0]["history"]) |
| 171 | messages = history["current"]["messages"] |
| 172 | self.assertEqual(history["counter"], len(messages)) |
| 173 | self.assertEqual([item["sequence"] for item in messages], list(range(1, len(messages) + 1))) |
| 174 | self.assertEqual([item["no"] for item in chat["log"]["logs"]], list(range(len(chat["log"]["logs"])))) |
| 175 | self.assertEqual([item["type"] for item in chat["log"]["logs"]], ["user", "agent", "response"]) |
| 176 | |
| 177 | |
| 178 | if __name__ == "__main__": |
| 179 | unittest.main() |