diff --git a/README.md b/README.md index 36d55f8..f755b30 100644 --- a/README.md +++ b/README.md @@ -193,9 +193,9 @@ bundle.zip/ - `--mode overwrite`：备份后覆盖已有会话 **推荐冲突策略（`--on-conflict`）**： -- `--on-conflict abort`（默认）：如果目标机器已有该会话，则报错退出 +- `--on-conflict abort`（默认）：如果目标机器已有该会话，则报错退出（JSONL、SQLite、index 任一项存在都算冲突；读取失败或身份不明会阻止导入） - `--on-conflict overwrite`：备份后覆盖已有会话 -- `--on-conflict import-as-new`：即使目标机器已有该会话，导入为新会话 +- `--on-conflict import-as-new`：即使目标机器已有该会话，导入为新会话。每次自动生成新 ID，重复执行会创建独立克隆；不按内容相似度去重 **导入时新建 session ID**： - `--new-session-id auto`：自动生成新 session ID @@ -359,3 +359,5 @@ python3 codex_workdir_migrate.py import-bundle \ - [import_safety.md](docs/import_safety.md) - 导入安全指南 - [known_limits.md](docs/known_limits.md) - 已知限制 - [migration_design.md](docs/migration_design.md) - 迁移设计文档 + +不确定导入结果时，保存 bundle、目标 ID、映射、冲突模式、备份和 `imported_files`，先检查目标再重试。需要重启后沿用同一克隆身份时，首次执行前指定 `--new-session-id `；现有 ID 是冲突证据，不是成功或内容等价凭据。详见 [导入安全指南](docs/import_safety.md#目标检查失败与不确定重试)。 diff --git a/codex_workdir_migrate.py b/codex_workdir_migrate.py index 830b875..ed80972 100755 --- a/codex_workdir_migrate.py +++ b/codex_workdir_migrate.py @@ -15,10 +15,12 @@ import shutil import socket import sqlite3 +import stat import subprocess import sys import uuid import zipfile +from contextlib import closing from dataclasses import dataclass, field, asdict from datetime import datetime from pathlib import Path @@ -102,6 +104,10 @@ def to_dict(self): return asdict(self) +class TargetInspectionError(RuntimeError): + """Target ownership or stored state could not be established safely.""" + + class CodexSessionMigrator: @staticmethod def _valid_session_id(session_id): @@ -117,41 +123,111 @@ def __init__(self, codex_home=None): self.state_db_path = os.path.join(self.codex_home, "state_5.sqlite") self.session_index_path = os.path.join(self.codex_home, "session_index.jsonl") + @staticmethod + def _target_path_present(path): + try: + os.stat(path) + return True + except FileNotFoundError as e: + if os.path.lexists(path): + raise TargetInspectionError(f"Cannot inspect target {path}: dangling symlink") from e + return False + except OSError as e: + raise TargetInspectionError(f"Cannot inspect target {path}: {e}") from e + def find_session_file(self, session_id): - """Find JSONL file(s) for a given session ID.""" + """Select JSONL only by exact session_meta identity, never filename text.""" candidates = [] + + def walk_error(error): + raise TargetInspectionError(f"Cannot inspect session directory: {error}") from error + for base_dir in [self.sessions_dir, self.archived_sessions_dir]: - if not os.path.exists(base_dir): + if not self._target_path_present(base_dir): continue - for root, _, files in os.walk(base_dir): - for file in files: - if file.endswith(".jsonl") and session_id in file: - candidates.append(os.path.join(root, file)) - - if not candidates: - for base_dir in [self.sessions_dir, self.archived_sessions_dir]: - if not os.path.exists(base_dir): - continue - for root, _, files in os.walk(base_dir): - for file in files: - if not file.endswith(".jsonl"): - continue - full_path = os.path.join(root, file) - try: - with open(full_path, "r", encoding="utf-8") as f: - for line in f: - try: - data = json.loads(line) - if data.get("payload", {}).get("id") == session_id: - candidates.append(full_path) - break - except json.JSONDecodeError: - continue - except Exception: - continue - + for root, dirs, files in os.walk(base_dir, onerror=walk_error): + dirs.sort() + # os.walk does not descend directory links, including links + # to sessions with irregular filenames. Skipping is unknown + # ownership, not authoritative absence. Dangling links can + # appear in files instead of dirs and must also block. + for entry in dirs + files: + full_path = os.path.join(root, entry) + try: + linked = stat.S_ISLNK(os.lstat(full_path).st_mode) + except OSError as e: + raise TargetInspectionError(f"Cannot inspect session entry {full_path}: {e}") from e + if linked and (entry in dirs or not entry.endswith(".jsonl")): + raise TargetInspectionError(f"Unsupported session directory link: {full_path}") + for file in sorted(files): + if not file.endswith(".jsonl"): + continue + full_path = os.path.join(root, file) + identity = None + try: + with open(full_path, "r", encoding="utf-8") as f: + for line_num, line in enumerate(f, 1): + if not line.strip(): + continue + try: + data = json.loads(line) + except json.JSONDecodeError as e: + raise TargetInspectionError( + f"Malformed session JSONL {full_path} line {line_num}: {e}" + ) from e + if not isinstance(data, dict): + raise TargetInspectionError( + f"Invalid session record {full_path} line {line_num}" + ) + if data.get("type") == "session_meta": + payload = data.get("payload") + record_id = payload.get("id") if isinstance(payload, dict) else None + if not isinstance(record_id, str) or not record_id: + raise TargetInspectionError( + f"Missing session_meta identity {full_path} line {line_num}" + ) + if identity is not None and identity != record_id: + raise TargetInspectionError(f"Ambiguous session identity in {full_path}") + identity = record_id + except (OSError, UnicodeError) as e: + raise TargetInspectionError(f"Cannot read session JSONL {full_path}: {e}") from e + if identity is None: + raise TargetInspectionError(f"Missing session_meta identity in {full_path}") + if identity == session_id: + candidates.append(full_path) return candidates + def _read_target_index(self): + """Validate the entire index, including unrelated rows and interrupted tails.""" + records = {} + lines = [] + if not self._target_path_present(self.session_index_path): + return records, lines + try: + with open(self.session_index_path, "r", encoding="utf-8", newline="") as f: + for line_num, line in enumerate(f, 1): + lines.append(line) + if not line.strip(): + continue + try: + record = json.loads(line) + except json.JSONDecodeError as e: + raise TargetInspectionError( + f"Malformed target index {self.session_index_path} line {line_num}: {e}" + ) from e + if not isinstance(record, dict) or not isinstance(record.get("id"), str) or not record["id"]: + raise TargetInspectionError( + f"Invalid target index record {self.session_index_path} line {line_num}: expected object with nonempty string id" + ) + if record["id"] in records: + raise TargetInspectionError( + f"Ambiguous target index ID {record['id']} at line {line_num}" + ) + records[record["id"]] = record + except (OSError, UnicodeError) as e: + raise TargetInspectionError(f"Cannot read target index {self.session_index_path}: {e}") from e + return records, lines + def _rewrite_path_value( self, value: Optional[str], old_cwd: str, new_cwd: str, rewrite_prefix_paths: bool ) -> Tuple[Optional[str], bool]: @@ -401,17 +477,9 @@ def inspect_session(self, session_id): result["jsonl_files"].append(file_info) try: - if os.path.exists(self.session_index_path): - with open(self.session_index_path, "r", encoding="utf-8") as f: - for line in f: - try: - data = json.loads(line) - if data.get("id") == session_id: - result["session_index"] = data - break - except json.JSONDecodeError: - continue - except Exception as e: + records, _ = self._read_target_index() + result["session_index"] = records.get(session_id, {}) + except TargetInspectionError as e: result["session_index"]["error"] = str(e) try: @@ -997,47 +1065,32 @@ def export_bundle(self, session_id, output_path, allow_sensitive_content=False): return result def check_import_target(self, session_id): - """Check if session already exists on target machine.""" + """Return known target state; inspection failures never imply absence.""" target_info = { "session_exists": False, - "jsonl_files": [], + "jsonl_files": self.find_session_file(session_id), "sqlite_record": None, "session_index_record": None, } - - target_jsonl_files = self.find_session_file(session_id) - if target_jsonl_files: - target_info["session_exists"] = True - target_info["jsonl_files"] = target_jsonl_files - - try: - if os.path.exists(self.state_db_path): - db = self._connect_state_db(readonly=True) - cursor = db.cursor() - cursor.execute("SELECT * FROM threads WHERE id = ?", (session_id,)) - row = cursor.fetchone() - if row: - cursor.execute("PRAGMA table_info(threads)") - col_names = [c[1] for c in cursor.fetchall()] - target_info["sqlite_record"] = dict(zip(col_names, row)) - db.close() - except Exception: - pass - - try: - if os.path.exists(self.session_index_path): - with open(self.session_index_path, "r", encoding="utf-8") as f: - for line in f: - try: - data = json.loads(line.strip()) - if data.get("id") == session_id: - target_info["session_index_record"] = data - break - except json.JSONDecodeError: - continue - except Exception: - pass - + if self._target_path_present(self.state_db_path): + try: + with closing(self._connect_state_db(readonly=True)) as db: + cursor = db.execute("SELECT * FROM threads WHERE id = ?", (session_id,)) + row = cursor.fetchone() + if row: + target_info["sqlite_record"] = dict(zip( + [column[0] for column in cursor.description], row + )) + except (sqlite3.Error, OSError) as e: + raise TargetInspectionError(f"Cannot read target SQLite {self.state_db_path}: {e}") from e + records, lines = self._read_target_index() + target_info["session_index_record"] = records.get(session_id) + target_info["index_records"] = records + target_info["index_lines"] = lines + target_info["session_exists"] = bool( + target_info["jsonl_files"] or target_info["sqlite_record"] is not None + or target_info["session_index_record"] is not None + ) return target_info def import_plan(self, bundle_path, cwd_mappings, policy=None): @@ -1103,10 +1156,10 @@ def import_plan(self, bundle_path, cwd_mappings, policy=None): target_info = self.check_import_target(source_session_id) - has_jsonl = target_info["session_exists"] + has_jsonl = bool(target_info["jsonl_files"]) has_sqlite = target_info["sqlite_record"] is not None has_index = target_info["session_index_record"] is not None - session_exists = has_jsonl or has_sqlite + session_exists = has_jsonl or has_sqlite or has_index result["session_exists_on_target"] = session_exists if not has_jsonl and not has_sqlite and not has_index: @@ -1179,25 +1232,22 @@ def import_plan(self, bundle_path, cwd_mappings, policy=None): ) else: try: - db = self._connect_state_db(readonly=True) - cursor = db.cursor() - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='threads'") - if not cursor.fetchone(): - result["errors"].append( - "Bundle contains SQLite data but target has no 'threads' table. Import will fail. Initialize Codex on target machine first." - ) - elif cwd_map: - planned_sqlite_updates.append({ - "action": "update", - "table": "threads", - "fields": ["cwd", "sandbox_policy"], - "cwd_mapping": cwd_map, - }) - db.close() + with closing(self._connect_state_db(readonly=True)) as db: + cursor = db.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='threads'") + if not cursor.fetchone(): + result["errors"].append( + "Bundle contains SQLite data but target has no 'threads' table. Import will fail. Initialize Codex on target machine first." + ) + elif cwd_map: + planned_sqlite_updates.append({ + "action": "update", + "table": "threads", + "fields": ["cwd", "sandbox_policy"], + "cwd_mapping": cwd_map, + }) except Exception as e: - result["warnings"].append( - f"Could not check target SQLite: {e}. SQLite import may fail." - ) + raise TargetInspectionError(f"Cannot check target SQLite schema: {e}") from e elif cwd_map: planned_sqlite_updates.append({ "action": "update", @@ -1242,6 +1292,10 @@ def import_plan(self, bundle_path, cwd_mappings, policy=None): if not result["errors"]: result["success"] = True + except TargetInspectionError as e: + result["target_status"] = {"status": "unknown", "message": str(e)} + result["session_exists_on_target"] = None + result["errors"].append(f"Target inspection failed: {e}") except Exception as e: result["errors"].append(f"Failed to read bundle: {e}") @@ -1322,10 +1376,10 @@ def import_bundle( target_info = self.check_import_target(source_session_id) - has_jsonl = target_info["session_exists"] + has_jsonl = bool(target_info["jsonl_files"]) has_sqlite = target_info["sqlite_record"] is not None has_index = target_info["session_index_record"] is not None - session_exists = has_jsonl or has_sqlite + session_exists = has_jsonl or has_sqlite or has_index effective_on_conflict = on_conflict if mode == "overwrite" and on_conflict == "abort": @@ -1371,18 +1425,20 @@ def import_bundle( return result jsonl_targets.append((member, target_path)) + destination_info = target_info if target_session_id != source_session_id: target_info_new = self.check_import_target(target_session_id) - has_jsonl_new = target_info_new["session_exists"] - has_sqlite_new = target_info_new["sqlite_record"] is not None - has_index_new = target_info_new["session_index_record"] is not None - target_session_exists = has_jsonl_new or has_sqlite_new or has_index_new - - if target_session_exists: - result["errors"].append( - f"Target session id '{target_session_id}' already exists on target. Use a different --new-session-id or --on-conflict overwrite." - ) - return result + destination_info = target_info_new + # A clone needs an unoccupied effective ID even when an + # explicit choice equals the source ID. Reuse its known state. + if destination_info["session_exists"] and ( + target_session_id != source_session_id + or effective_on_conflict == "import-as-new" + ): + result["errors"].append( + f"Target session id '{target_session_id}' already exists on target. Use a different --new-session-id or --on-conflict overwrite." + ) + return result if session_exists and effective_on_conflict == "abort": result["errors"].append( @@ -1400,6 +1456,13 @@ def import_bundle( ) return result + for _, target_path in jsonl_targets: + if self._target_path_present(target_path) and ( + effective_on_conflict != "overwrite" + or target_path not in destination_info["jsonl_files"] + ): + raise TargetInspectionError(f"JSONL destination belongs to another session or overwrite is not authorized: {target_path}") + sqlite_member = "sqlite/state_db_matching_rows.json" has_sqlite_in_bundle = sqlite_member in zf.namelist() @@ -1416,47 +1479,40 @@ def import_bundle( return result try: - db = self._connect_state_db(readonly=True) - cursor = db.cursor() - cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='threads'") - if not cursor.fetchone(): - result["errors"].append( - f"Bundle contains SQLite data but target has no 'threads' table. " - f"Import will fail. Initialize Codex on target machine first." - ) - db.close() - return result - - cursor.execute("PRAGMA table_info(threads)") - not_null_columns = set() - for row in cursor.fetchall(): - col_name = row[1] - not_null = row[3] - dflt_value = row[4] - if not_null and dflt_value is None: - not_null_columns.add(col_name) - - if bundle_rows and not_null_columns: - missing_not_null = [] - bundle_row = bundle_rows[0] - for col in not_null_columns: - if col not in bundle_row or bundle_row.get(col) is None: - missing_not_null.append(col) - - if missing_not_null: + with closing(self._connect_state_db(readonly=True)) as db: + cursor = db.cursor() + cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name='threads'") + if not cursor.fetchone(): result["errors"].append( - f"Bundle SQLite row missing NOT NULL columns without defaults: {missing_not_null}. " - f"Import will fail. Ensure bundle contains all required columns." + f"Bundle contains SQLite data but target has no 'threads' table. " + f"Import will fail. Initialize Codex on target machine first." ) - db.close() return result - db.close() + cursor.execute("PRAGMA table_info(threads)") + not_null_columns = set() + for row in cursor.fetchall(): + col_name = row[1] + not_null = row[3] + dflt_value = row[4] + if not_null and dflt_value is None: + not_null_columns.add(col_name) + + if bundle_rows and not_null_columns: + missing_not_null = [] + bundle_row = bundle_rows[0] + for col in not_null_columns: + if col not in bundle_row or bundle_row.get(col) is None: + missing_not_null.append(col) + + if missing_not_null: + result["errors"].append( + f"Bundle SQLite row missing NOT NULL columns without defaults: {missing_not_null}. " + f"Import will fail. Ensure bundle contains all required columns." + ) + return result except Exception as e: - result["errors"].append( - f"Failed to check target SQLite: {e}. Import will fail." - ) - return result + raise TargetInspectionError(f"Cannot check target SQLite schema: {e}") from e files_to_backup = [] if os.path.exists(self.state_db_path): @@ -1690,17 +1746,8 @@ def import_bundle( index_content = zf.read(index_member).decode("utf-8") index_path = self.session_index_path - existing_records = {} - - if os.path.exists(index_path): - with open(index_path, "r", encoding="utf-8") as f: - for line in f: - try: - record = json.loads(line.strip()) - if record.get("id"): - existing_records[record["id"]] = line - except json.JSONDecodeError: - continue + existing_records = target_info["index_records"] + existing_index_lines = target_info["index_lines"] session_index_record = None for line in index_content.splitlines(): @@ -1743,20 +1790,25 @@ def import_bundle( source_session_id, target_session_id ) - if mode == "skip" and target_session_id in existing_records: + if mode == "skip" and effective_on_conflict != "overwrite" and target_session_id in existing_records: result["imported_files"].append({ "type": "session_index", "action": "skipped", "reason": "session exists", }) else: - existing_records[target_session_id] = json.dumps(session_index_record, ensure_ascii=False) + "\n" + new_index_line = json.dumps(session_index_record, ensure_ascii=False) + "\n" if not dry_run: temp_path = index_path + ".tmp" - with open(temp_path, "w", encoding="utf-8") as f: - for rec_line in existing_records.values(): + with open(temp_path, "w", encoding="utf-8", newline="") as f: + for rec_line in existing_index_lines: + if rec_line.strip() and json.loads(rec_line)["id"] == target_session_id: + continue f.write(rec_line) + if existing_index_lines and not existing_index_lines[-1].endswith(("\n", "\r")): + f.write("\n") + f.write(new_index_line) os.replace(temp_path, index_path) result["imported_files"].append({ @@ -1768,6 +1820,8 @@ def import_bundle( result["success"] = True + except TargetInspectionError as e: + result["errors"].append(f"Target inspection failed: {e}") except Exception as e: result["errors"].append(f"Import failed: {e}") diff --git a/docs/import_safety.md b/docs/import_safety.md index 073b413..622d661 100644 --- a/docs/import_safety.md +++ b/docs/import_safety.md @@ -91,7 +91,7 @@ ### 场景 3: 目标机器已有该会话（残缺） -- 仅 JSONL 或 SQLite 之一存在 +- JSONL、SQLite 或 index 任意一项存在，都视为会话冲突；残缺状态不是不存在 - 使用 `--mode overwrite` 替换完整记录 - **警告**: 这可能丢失目标机器上的某些更新 @@ -121,3 +121,13 @@ Bundle 的 SHA256SUMS.txt 和 manifest checksums 仅供人工审计；导入当 - ❌ 不要使用 merge 模式 - ❌ 不要跳过 import-plan 直接 import - ❌ 不要在 Codex 运行时执行导入 + +## 目标检查失败与不确定重试 + +默认 `--on-conflict abort` 会拒绝任何同 ID 的 JSONL、SQLite 或 index 记录。只有成功检查且三者都不存在，才视为目标不存在。数据库锁定、损坏、权限不足、缺少 `threads.id`，或 index 非空行不是带非空字符串 ID 的对象、JSON 损坏或 ID 重复时，`import-plan` 报告 `unknown`，导入在备份、删除和写入之前失败。不会静默修复或丢弃残缺 index；保留原始字节，先人工恢复再重新检查。空行和有效的末行没有换行符可以读取，导入保留无关记录及空行。 + +JSONL 文件名不证明会话归属。工具扫描 `sessions/` 和 `archived_sessions/`，只按 `session_meta.payload.id` 精确匹配，支持不规则文件名及安全的非 UUID ID。无法读取、JSON 损坏、缺少或冲突的 metadata ID 会阻止检查；其他消息的 `payload.id` 不作为会话身份。目的文件属于其他 ID 时，即使选择 overwrite 也拒绝写入。明确选择 `--mode overwrite` 或 `--on-conflict overwrite` 才允许替换同 ID 的数据，包括 index。 + +自动 `--on-conflict import-as-new` 每次生成新的 UUID；重复执行是两次有意克隆，不按源内容相似度去重。需要可重启的克隆时，在首次执行前选定 `--new-session-id `，保存原始 bundle、返回/指定的目标 ID、路径映射、冲突模式、备份和 `imported_files`。同 ID 已存在只证明冲突，不证明内容等价或导入完整。 + +若响应丢失或导入中断，先用已知目标 ID 检查目标再决定下一步，不要直接再次自动克隆。`success=false` 也可能已经写入 JSONL；工具仍没有跨文件事务或自动回滚保证。部分导入会阻止默认重试；覆盖必须是检查后的明确选择。导入期间目标 home 必须停止其他写入，本次检查不是并发锁或在线快照。 diff --git a/tests/test_import_review_guards.py b/tests/test_import_review_guards.py new file mode 100644 index 0000000..2229497 --- /dev/null +++ b/tests/test_import_review_guards.py @@ -0,0 +1,162 @@ +"""Regressions for independently observed destination and enumeration guards.""" +import json +import sqlite3 +import tempfile +import unittest +import zipfile +from contextlib import closing +from datetime import datetime +from pathlib import Path +from unittest.mock import patch + +from codex_workdir_migrate import CodexSessionMigrator + +SID = "synthetic-review-source" + + +class ImportReviewGuardTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix="migrator-review-synthetic-") + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.home = self.root / "home" + (self.home / "sessions").mkdir(parents=True) + self.m = CodexSessionMigrator(str(self.home)) + self.bundle = self.root / "synthetic.zip" + self.backups = self.root / "backups" + self.metadata = json.dumps({"type": "session_meta", "payload": {"id": SID}}) + "\n" + + def make_bundle(self, filename): + with zipfile.ZipFile(self.bundle, "w") as z: + z.writestr("MANIFEST.json", json.dumps({"session_id": SID})) + z.writestr("sessions/raw_jsonl/" + filename, self.metadata) + + def snapshot(self): + # Includes linked source bytes and the link itself without relying on + # rglob following directory symlinks. + files = {str(p.relative_to(self.root)): p.read_bytes() + for p in self.root.rglob("*") if p.is_file()} + links = {str(p.relative_to(self.root)): str(p.readlink()) + for p in self.root.rglob("*") if p.is_symlink()} + return files, links + + def assert_import_blocked(self, before, **options): + for dry_run in [True, False]: + with self.subTest(dry_run=dry_run), patch("codex_workdir_migrate.shutil.copy2") as copy, patch("codex_workdir_migrate.os.unlink") as unlink, patch("codex_workdir_migrate.os.replace") as replace: + result = self.m.import_bundle(str(self.bundle), [], str(self.backups), dry_run=dry_run, **options) + self.assertFalse(result["success"], result) + self.assertTrue(result["errors"], result) + self.assertEqual([], result["imported_files"]) + self.assertIsNone(result["backup"]) + copy.assert_not_called() + unlink.assert_not_called() + replace.assert_not_called() + self.assertEqual(before, self.snapshot()) + self.assertFalse(self.backups.exists()) + + def test_irregular_source_collision_blocks_auto_and_explicit_clone(self): + self.make_bundle("irregular.jsonl") + source = self.home / "sessions" / datetime.now().strftime("%Y/%m/%d") / "irregular.jsonl" + source.parent.mkdir(parents=True) + source.write_text(self.metadata) + before = self.snapshot() + for no_backup in [False, True]: + for options in [{"on_conflict": "import-as-new"}, + {"on_conflict": "import-as-new", "new_session_id": "explicit-clone"}, + {"on_conflict": "import-as-new", "mode": "overwrite"}, + {"on_conflict": "overwrite", "new_session_id": "explicit-clone"}]: + with self.subTest(no_backup=no_backup, options=options): + self.assert_import_blocked(before, no_backup=no_backup, **options) + self.assertEqual(self.metadata.encode(), source.read_bytes()) + + def test_irregular_same_target_explicit_overwrite_remains_supported(self): + self.make_bundle("irregular.jsonl") + source = self.home / "sessions" / datetime.now().strftime("%Y/%m/%d") / "irregular.jsonl" + source.parent.mkdir(parents=True) + source.write_text(self.metadata) + for options in [{"mode": "overwrite"}, {"on_conflict": "overwrite"}]: + with self.subTest(options=options): + result = self.m.import_bundle(str(self.bundle), [], None, dry_run=False, no_backup=True, **options) + self.assertTrue(result["success"], result) + self.assertEqual(SID, json.loads(source.read_text())["payload"]["id"]) + + def test_nested_active_and_archived_directory_links_block_as_unknown(self): + self.make_bundle(SID + ".jsonl") + for tree in ["sessions", "archived_sessions"]: + for dangling in [False, True]: + with self.subTest(tree=tree, dangling=dangling): + parent = self.home / tree / "nested" + parent.mkdir(parents=True, exist_ok=True) + held = self.home / f"held-{tree}" + if not dangling: + held.mkdir(exist_ok=True) + (held / "irregular.jsonl").write_text(self.metadata) + target = self.home / "absent-directory" if dangling else held + link = parent / "linked-directory" + link.symlink_to(target, target_is_directory=True) + before = self.snapshot() + plan = self.m.import_plan(str(self.bundle), []) + self.assertFalse(plan["success"], plan) + self.assertEqual("unknown", plan["target_status"]["status"]) + self.assertIsNone(plan["session_exists_on_target"]) + self.assertIn("directory link", " ".join(plan["errors"])) + for options in [{}, {"mode": "overwrite"}, {"on_conflict": "import-as-new"}]: + self.assert_import_blocked(before, **options) + link.unlink() + + def test_directory_link_with_jsonl_suffix_is_still_blocked(self): + self.make_bundle(SID + ".jsonl") + held = self.home / "held" + held.mkdir() + (held / "existing.jsonl").write_text(self.metadata) + (self.home / "sessions" / "nested.jsonl").symlink_to(held, target_is_directory=True) + self.assert_import_blocked(self.snapshot()) + + def occupied_source_bundle(self): + self.make_bundle("different-path-" + SID + ".jsonl") + archive = self.home / "archived_sessions" + archive.mkdir() + (archive / "irregular.jsonl").write_text(self.metadata) + (self.home / "session_index.jsonl").write_text(json.dumps({"id": SID, "title": "existing"}) + "\n") + db_path = self.home / "state_5.sqlite" + with closing(sqlite3.connect(db_path)) as db: + db.execute("CREATE TABLE threads (id TEXT PRIMARY KEY, cwd TEXT)") + db.execute("INSERT INTO threads VALUES (?, ?)", (SID, "/existing")) + db.commit() + with zipfile.ZipFile(self.bundle, "a") as z: + z.writestr("sqlite/state_db_matching_rows.json", json.dumps({"matching_rows": [{"id": SID, "cwd": "/imported"}]})) + z.writestr("index/session_index_records.jsonl", json.dumps({"id": SID, "title": "imported"}) + "\n") + return db_path + + def test_occupied_equal_id_clone_blocks_before_any_mutation(self): + db_path = self.occupied_source_bundle() + before = self.snapshot() + for no_backup in [False, True]: + for mode in ["skip", "overwrite"]: + with self.subTest(no_backup=no_backup, mode=mode): + self.assert_import_blocked(before, new_session_id=SID, + on_conflict="import-as-new", mode=mode, + no_backup=no_backup) + with closing(sqlite3.connect(db_path)) as db: + self.assertEqual("/existing", db.execute("SELECT cwd FROM threads WHERE id = ?", (SID,)).fetchone()[0]) + + def test_distinct_clone_and_equal_id_authorized_overwrite_remain_supported(self): + db_path = self.occupied_source_bundle() + source = self.home / "archived_sessions" / "irregular.jsonl" + original = source.read_bytes() + clone = self.m.import_bundle(str(self.bundle), [], None, dry_run=False, + no_backup=True, new_session_id="distinct-clone", + on_conflict="import-as-new") + self.assertTrue(clone["success"], clone) + self.assertEqual(original, source.read_bytes()) + with closing(sqlite3.connect(db_path)) as db: + self.assertEqual("/existing", db.execute("SELECT cwd FROM threads WHERE id = ?", (SID,)).fetchone()[0]) + self.assertEqual("/imported", db.execute("SELECT cwd FROM threads WHERE id = ?", ("distinct-clone",)).fetchone()[0]) + overwritten = self.m.import_bundle(str(self.bundle), [], None, dry_run=False, + no_backup=True, new_session_id=SID, + on_conflict="overwrite") + self.assertTrue(overwritten["success"], overwritten) + with closing(sqlite3.connect(db_path)) as db: + self.assertEqual("/imported", db.execute("SELECT cwd FROM threads WHERE id = ?", (SID,)).fetchone()[0]) + index = [json.loads(line) for line in (self.home / "session_index.jsonl").read_text().splitlines()] + self.assertEqual("imported", next(row for row in index if row["id"] == SID)["title"]) diff --git a/tests/test_migrator.py b/tests/test_migrator.py index 26b7328..b2c1c52 100644 --- a/tests/test_migrator.py +++ b/tests/test_migrator.py @@ -14,7 +14,7 @@ def _write_jsonl(path: Path): records = [ - {"type": "session_meta", "payload": {"cwd": OLD}}, + {"type": "session_meta", "payload": {"id": SESSION_ID, "cwd": OLD}}, {"type": "turn_context", "payload": {"cwd": OLD}}, { "type": "response_item", diff --git a/tests/test_target_inspection.py b/tests/test_target_inspection.py new file mode 100644 index 0000000..13c845e --- /dev/null +++ b/tests/test_target_inspection.py @@ -0,0 +1,333 @@ +"""Import safety contracts using disposable synthetic homes and injected failures.""" +import builtins +import json +import os +import sqlite3 +import tempfile +import unittest +import zipfile +from datetime import datetime +from pathlib import Path +from unittest.mock import patch + +from codex_workdir_migrate import CodexSessionMigrator + +SID = "synthetic-workflow-01" + + +class TargetInspectionTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory(prefix="migrator-target-synthetic-") + self.addCleanup(self.tmp.cleanup) + self.root = Path(self.tmp.name) + self.home = self.root / "home" + (self.home / "sessions").mkdir(parents=True) + self.m = CodexSessionMigrator(str(self.home)) + self.bundle = self.root / "synthetic.zip" + self.make_bundle() + + def make_bundle(self, sqlite=False, cwd="/synthetic-source"): + with zipfile.ZipFile(self.bundle, "w") as z: + z.writestr("MANIFEST.json", json.dumps({"session_id": SID})) + z.writestr(f"sessions/raw_jsonl/rollout-{SID}.jsonl", json.dumps({ + "type": "session_meta", "payload": {"id": SID, "cwd": cwd}}) + "\n") + z.writestr("index/session_index_records.jsonl", json.dumps({"id": SID, "title": cwd}) + "\n") + if sqlite: + z.writestr("sqlite/state_db_matching_rows.json", json.dumps({ + "matching_rows": [{"id": SID, "cwd": cwd}]})) + + def db(self, existing=False, schema="id TEXT PRIMARY KEY, cwd TEXT"): + with sqlite3.connect(self.home / "state_5.sqlite") as db: + db.execute(f"CREATE TABLE threads ({schema})") + if existing: + db.execute("INSERT INTO threads VALUES (?, ?)", (SID, "/existing")) + + def jsonl(self, sid=SID, name=None, archived=False): + p = self.home / ("archived_sessions" if archived else "sessions") / (name or f"rollout-{sid}.jsonl") + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(json.dumps({"type": "session_meta", "payload": {"id": sid}}) + "\n") + return p + + def snapshot(self): + return {str(p.relative_to(self.home)): p.read_bytes() + for p in self.home.rglob("*") if p.is_file()} + + def run_import(self, **kwargs): + return self.m.import_bundle(str(self.bundle), [], str(self.root / "backups"), + dry_run=False, **kwargs) + + def assert_blocked(self, before, result, error=None): + self.assertFalse(result["success"], result) + self.assertEqual(before, self.snapshot()) + self.assertIsNone(result["backup"]) + self.assertEqual([], result["imported_files"]) + self.assertFalse((self.root / "backups").exists()) + if error: + self.assertIn(error, " ".join(result["errors"])) + + def assert_inspection_blocks(self, error): + before = self.snapshot() + plan = self.m.import_plan(str(self.bundle), []) + self.assertFalse(plan["success"], plan) + self.assertEqual("unknown", plan["target_status"]["status"]) + self.assertIsNone(plan["session_exists_on_target"]) + for options in [{}, {"mode": "overwrite"}, {"on_conflict": "import-as-new"}]: + self.assert_blocked(before, self.run_import(**options), error) + + def test_M1b_first_read_failure_never_reaches_later_writes(self): + self.db(existing=True) + self.make_bundle(sqlite=True) + before = self.snapshot() + connect = self.m._connect_state_db + calls = [] + def fail_first(readonly=False): + calls.append(readonly) + if len(calls) == 1: + raise sqlite3.OperationalError("synthetic transient read failure") + return connect(readonly=readonly) + with patch.object(self.m, "_connect_state_db", side_effect=fail_first): + self.assert_blocked(before, self.run_import(), "transient read failure") + self.assertEqual([True], calls) + with sqlite3.connect(self.home / "state_5.sqlite") as db: + self.assertEqual("/existing", db.execute("SELECT cwd FROM threads").fetchone()[0]) + + def test_M1_corrupt_db_is_unknown_even_for_jsonl_bundle(self): + (self.home / "state_5.sqlite").write_bytes(b"synthetic non-SQLite bytes") + self.assert_inspection_blocks("target SQLite") + + def test_schema_mismatch_closes_read_handles(self): + self.db(schema="wrong_id TEXT") + connections = [] + connect = self.m._connect_state_db + def track(readonly=False): + conn = connect(readonly=readonly) + connections.append(conn) + return conn + with patch.object(self.m, "_connect_state_db", side_effect=track): + self.assert_inspection_blocks("no such column: id") + for conn in connections: + with self.assertRaises(sqlite3.ProgrammingError): + conn.execute("SELECT 1") + + def test_locked_db_blocks_before_backup(self): + self.db() + lock = sqlite3.connect(self.home / "state_5.sqlite") + self.addCleanup(lock.close) + lock.execute("BEGIN EXCLUSIVE") + def fast(readonly=False): + return sqlite3.connect(f"file:{self.home / 'state_5.sqlite'}?mode=ro", uri=True, timeout=0.01) + with patch.object(self.m, "_connect_state_db", side_effect=fast): + self.assert_inspection_blocks("locked") + lock.rollback() + + def test_unreadable_db(self): + self.db() + with patch.object(self.m, "_connect_state_db", side_effect=PermissionError("synthetic unreadable DB")): + self.assert_inspection_blocks("unreadable DB") + + def test_stat_read_failure_is_not_absence(self): + real_stat = os.stat + def denied(path, *args, **kwargs): + if str(path) == self.m.state_db_path: + raise PermissionError("synthetic stat denied") + return real_stat(path, *args, **kwargs) + with patch("codex_workdir_migrate.os.stat", side_effect=denied): + self.assert_inspection_blocks("stat denied") + + def test_dangling_db_is_not_absence(self): + (self.home / "state_5.sqlite").symlink_to(self.root / "missing-db") + self.assert_inspection_blocks("dangling symlink") + + def test_M2_every_partial_authority_and_complete_target_conflict(self): + for state in ["index", "db", "jsonl", "complete"]: + with self.subTest(state=state), tempfile.TemporaryDirectory() as td: + self.home = Path(td) + self.m = CodexSessionMigrator(td) + if state in ["db", "complete"]: + self.db(existing=True) + if state in ["jsonl", "complete"]: + self.jsonl() + if state in ["index", "complete"]: + (self.home / "session_index.jsonl").write_text(json.dumps({"id": SID, "title": "existing"}) + "\n") + plan = self.m.import_plan(str(self.bundle), []) + self.assertTrue(plan["success"], plan) + self.assertTrue(plan["session_exists_on_target"]) + self.assertEqual("complete" if state == "complete" else "incomplete", plan["target_status"]["status"]) + before = self.snapshot() + self.assert_blocked(before, self.run_import(), "already exists") + + def test_index_only_explicit_overwrite_and_clone(self): + idx = self.home / "session_index.jsonl" + idx.write_text(json.dumps({"id": SID, "title": "existing"}) + "\n") + result = self.run_import(on_conflict="import-as-new") + self.assertTrue(result["success"], result) + rows = [json.loads(line) for line in idx.read_text().splitlines()] + self.assertEqual("existing", rows[0]["title"]) + self.assertEqual(result["target_session_id"], rows[1]["id"]) + result = self.run_import(on_conflict="overwrite") + self.assertTrue(result["success"], result) + rows = [json.loads(line) for line in idx.read_text().splitlines()] + self.assertEqual("/synthetic-source", next(row for row in rows if row["id"] == SID)["title"]) + + def test_M3_malformed_middle_tail_shape_and_duplicate_index_preserved(self): + valid = json.dumps({"id": "unrelated", "title": "preserve"}) + "\n" + cases = [(valid + '{"id":"partial"', "line 2"), + (json.dumps({"id": SID}) + '\n' + '{"id":"partial"', "line 2"), + (valid + 'broken\n' + json.dumps({"id": SID}) + "\n", "line 2"), + ('[]\n', "expected object"), ('{"title":"missing id"}\n', "expected object"), + ('{"id":123}\n', "expected object"), + (valid + valid, "Ambiguous target index")] + for contents, error in cases: + with self.subTest(contents=contents): + (self.home / "session_index.jsonl").write_text(contents) + self.assert_inspection_blocks(error) + + def test_unreadable_index_blocks(self): + idx = self.home / "session_index.jsonl" + idx.write_text(json.dumps({"id": SID}) + "\n") + real_open = builtins.open + def denied(path, *args, **kwargs): + if str(path) == str(idx): + raise PermissionError("synthetic index unreadable") + return real_open(path, *args, **kwargs) + with patch("builtins.open", side_effect=denied): + self.assert_inspection_blocks("index unreadable") + + def test_valid_index_blank_lines_and_unterminated_valid_row_preserved(self): + idx = self.home / "session_index.jsonl" + original = '\r\n {"id":"unrelated","title":"keep spacing"}\r\n\n{"id":"last"}' + idx.write_bytes(original.encode()) + result = self.run_import() + self.assertTrue(result["success"], result) + self.assertTrue(idx.read_bytes().startswith(original.encode() + b"\n")) + self.assertEqual(SID, json.loads(idx.read_text().splitlines()[-1])["id"]) + + def test_M5_overlapping_id_never_selects_or_deletes_unrelated(self): + other = self.jsonl(SID + "-other") + original = other.read_bytes() + self.assertEqual([], self.m.find_session_file(SID)) + result = self.run_import(mode="overwrite") + self.assertTrue(result["success"], result) + self.assertEqual(original, other.read_bytes()) + + def test_short_id_and_irregular_archived_filenames_use_metadata(self): + for sid in ["l", "jsonl", SID]: + with self.subTest(sid=sid): + match = self.jsonl(sid, name=f"irregular-{sid}-name.jsonl", archived=True) + other = self.jsonl(sid + "-other") + self.assertEqual([str(match)], self.m.find_session_file(sid)) + self.assertNotIn(str(other), self.m.find_session_file(sid)) + + def test_unrelated_payload_id_is_not_session_identity(self): + other = self.jsonl(SID + "-other") + with other.open("a") as f: + f.write(json.dumps({"type": "response_item", "payload": {"id": SID}}) + "\n") + self.assertEqual([], self.m.find_session_file(SID)) + + def test_malformed_missing_and_ambiguous_jsonl_block(self): + p = self.home / "sessions" / f"rollout-{SID}.jsonl" + meta = json.dumps({"type": "session_meta", "payload": {"id": SID}}) + "\n" + for content in ['broken\n', '{}\n', '[]\n', meta + '{"type":"partial"', + meta + json.dumps({"type": "session_meta", "payload": {"id": "other"}}) + "\n"]: + with self.subTest(content=content): + p.write_text(content) + self.assert_inspection_blocks("Target inspection failed") + + def test_unreadable_jsonl_blocks(self): + p = self.jsonl() + real_open = builtins.open + def denied(path, *args, **kwargs): + if str(path) == str(p): + raise PermissionError("synthetic JSONL unreadable") + return real_open(path, *args, **kwargs) + with patch("builtins.open", side_effect=denied): + self.assert_inspection_blocks("JSONL unreadable") + + def test_directory_enumeration_failure_blocks(self): + def failed_walk(path, onerror=None): + onerror(PermissionError("synthetic directory unreadable")) + return iter([]) + with patch("codex_workdir_migrate.os.walk", side_effect=failed_walk): + self.assert_inspection_blocks("directory unreadable") + + def test_matching_filename_with_wrong_metadata_never_overwritten(self): + p = self.home / "sessions" / datetime.now().strftime("%Y/%m/%d") / f"rollout-{SID}.jsonl" + p.parent.mkdir(parents=True) + p.write_text(json.dumps({"type": "session_meta", "payload": {"id": "unrelated"}}) + "\n") + self.assert_blocked(self.snapshot(), self.run_import(mode="overwrite"), "another session") + + def test_M4_automatic_repeats_clone_and_explicit_retry_conflicts(self): + first = self.run_import(on_conflict="import-as-new", no_backup=True) + second = self.run_import(on_conflict="import-as-new", no_backup=True) + self.assertTrue(first["success"], first) + self.assertTrue(second["success"], second) + self.assertNotEqual(first["target_session_id"], second["target_session_id"]) + before = self.snapshot() + self.assert_blocked(before, self.run_import(new_session_id=first["target_session_id"]), "already exists") + + def test_unchanged_and_changed_source_repeat_never_overwrite_by_default(self): + self.db() + self.make_bundle(sqlite=True) + first = self.run_import(no_backup=True) + self.assertTrue(first["success"], first) + before = self.snapshot() + self.assert_blocked(before, self.run_import(), "already exists") + self.make_bundle(sqlite=True, cwd="/changed-remote-equivalent-source") + self.assert_blocked(before, self.run_import(), "already exists") + overwritten = self.run_import(mode="overwrite") + self.assertTrue(overwritten["success"], overwritten) + with sqlite3.connect(self.home / "state_5.sqlite") as db: + self.assertEqual("/changed-remote-equivalent-source", db.execute("SELECT cwd FROM threads").fetchone()[0]) + + def test_interrupted_import_partial_jsonl_blocks_retry(self): + self.db() + self.make_bundle(sqlite=True) + connect = self.m._connect_state_db + def interrupt(readonly=False): + if not readonly: + raise sqlite3.OperationalError("synthetic interruption after JSONL write") + return connect(readonly=readonly) + with patch.object(self.m, "_connect_state_db", side_effect=interrupt): + first = self.run_import(no_backup=True) + self.assertFalse(first["success"]) + self.assertTrue(first["imported_files"]) + self.assertEqual(1, len(list((self.home / "sessions").rglob("*.jsonl")))) + before = self.snapshot() + self.assert_blocked(before, self.run_import(), "already exists") + self.assertEqual("incomplete", self.m.import_plan(str(self.bundle), [])["target_status"]["status"]) + + def test_later_schema_read_failure_closes_handles_and_blocks_plan_and_import(self): + self.db() + self.make_bundle(sqlite=True) + class BrokenCursor(sqlite3.Cursor): + def execute(self, *args, **kwargs): + raise sqlite3.OperationalError("synthetic later schema read failure") + class BrokenConnection(sqlite3.Connection): + def cursor(self): + return super().cursor(factory=BrokenCursor) + before = self.snapshot() + for action in [lambda: self.m.import_plan(str(self.bundle), []), self.run_import]: + connections = [] + def connect(readonly=False): + factory = BrokenConnection if connections else sqlite3.Connection + conn = sqlite3.connect(self.home / "state_5.sqlite", factory=factory) + connections.append(conn) + return conn + with self.subTest(action=action), patch.object(self.m, "_connect_state_db", side_effect=connect): + result = action() + self.assertFalse(result["success"], result) + self.assertIn("later schema read failure", " ".join(result["errors"])) + self.assertEqual(before, self.snapshot()) + self.assertFalse((self.root / "backups").exists()) + if "target_status" in result: + self.assertEqual("unknown", result["target_status"]["status"]) + else: + self.assertEqual([], result["imported_files"]) + for conn in connections: + with self.assertRaises(sqlite3.ProgrammingError): + conn.execute("SELECT 1") + + def test_inspect_reports_malformed_index_instead_of_hiding_tail(self): + (self.home / "session_index.jsonl").write_text(json.dumps({"id": SID}) + '\n{"id":"partial"') + inspected = self.m.inspect_session(SID) + self.assertIn("line 2", inspected["session_index"]["error"])