From 1d5b60346df848cfaaa033a3dc01feaf22776a6c Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 22 Aug 2026 00:32:04 +0200 Subject: [PATCH] feat(identifier): add reversible file migration Assistant: codex Assistant-Model: gpt-5.6-sol Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d --- src/repo_manager/cli.py | 31 +++ src/repo_manager/identifiers.py | 235 +++++++++++++++++- tests/test_identifiers.py | 65 +++++ ...gistrar-consolidation-deterministic-ids.md | 25 +- 4 files changed, 354 insertions(+), 2 deletions(-) diff --git a/src/repo_manager/cli.py b/src/repo_manager/cli.py index f5e6f98..3334fb3 100644 --- a/src/repo_manager/cli.py +++ b/src/repo_manager/cli.py @@ -300,6 +300,21 @@ def main(argv: list[str] | None = None) -> int: ) p_id_verify.add_argument("--plan", required=True) p_id_verify.add_argument("--repo", default=None, help="Verify one repository atomic unit") + p_id_files = identifier_sub.add_parser( + "migration-files", + help="Validate or execute one repository's sealed UUID file rewrite", + ) + p_id_files.add_argument("--plan", required=True) + p_id_files.add_argument("--repo", required=True) + p_id_files.add_argument("--confirm-plan-sha256", required=True) + p_id_files.add_argument( + "--direction", choices=["forward", "reverse"], default="forward" + ) + p_id_files.add_argument( + "--execute", + action="store_true", + help="Write files; without this flag only validate and report", + ) p_sbom = sub.add_parser("sbom", help="Derive SBOM snapshots and licence reports from repository files") sbom_sub = p_sbom.add_subparsers(dest="sbom_command") @@ -623,6 +638,7 @@ def main(argv: list[str] | None = None) -> int: from repo_manager.identifiers import ( derive_work_record_uuid, load_fleet_namespace, + migrate_repository_identifier_files, plan_identifier_migration, scan_live_identifier_collisions, verify_identifier_migration_plan, @@ -653,6 +669,21 @@ def main(argv: list[str] | None = None) -> int: except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) return 1 + elif args.identifier_command == "migration-files": + try: + plan = json.loads(Path(args.plan).read_text(encoding="utf-8")) + if not isinstance(plan, dict): + raise TypeError("migration plan must be a JSON object") + result = migrate_repository_identifier_files( + plan, + repo_slug=args.repo, + confirm_plan_sha256=args.confirm_plan_sha256, + direction=args.direction, + execute=args.execute, + ) + except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc: + print(json.dumps({"ok": False, "error": str(exc)}, indent=2)) + return 1 else: try: result = plan_identifier_migration(Path(args.root), namespace) diff --git a/src/repo_manager/identifiers.py b/src/repo_manager/identifiers.py index c9cd720..3ed82f9 100644 --- a/src/repo_manager/identifiers.py +++ b/src/repo_manager/identifiers.py @@ -4,7 +4,9 @@ from __future__ import annotations import hashlib import json +import os import re +import tempfile import uuid from collections import defaultdict from pathlib import Path @@ -14,7 +16,12 @@ import yaml from repo_manager.cache import source_fingerprint from repo_manager.gitops import head_sha -from repo_manager.parse.workplan import iter_workplan_files, parse_workplan_file +from repo_manager.parse.workplan import ( + _TASK_BLOCK_RE, + _parse_yaml_block, + iter_workplan_files, + parse_workplan_file, +) from repo_manager.prefix_registry import iter_repo_roots from repo_manager.time import utc_now_text @@ -287,3 +294,229 @@ def verify_identifier_migration_plan( "apply_authorized": False, "remaining_gate": "transactional central-projection migration and rollback", } + + +def _verified_plan_seal(plan: dict[str, Any]) -> str: + if plan.get("schema") != "repo-manager.identifier-migration-plan.v1": + raise ValueError("unsupported identifier migration plan schema") + expected = plan.get("plan_sha256") + if not isinstance(expected, str) or not re.fullmatch(r"[0-9a-f]{64}", expected): + raise ValueError("migration plan has no valid SHA-256 seal") + unsealed = {key: value for key, value in plan.items() if key != "plan_sha256"} + canonical = json.dumps(unsealed, sort_keys=True, separators=(",", ":")).encode("utf-8") + if hashlib.sha256(canonical).hexdigest() != expected: + raise ValueError("migration plan SHA-256 mismatch") + if plan.get("namespace") != load_fleet_namespace(): + raise ValueError("migration plan namespace does not match the fleet declaration") + return expected + + +def _yaml_scalar(value: str | None) -> str | None: + return value.strip().strip('"').strip("'") if value is not None else None + + +def _patch_yaml_field( + raw: str, + *, + key: str, + expected: str | None, + target: str | None, + scope: str, +) -> str: + pattern = re.compile(rf"^(?P{re.escape(key)}:\s*)(?P.*)$", re.MULTILINE) + match = pattern.search(raw) + current = _yaml_scalar(match.group("value")) if match else None + if current != expected: + raise ValueError(f"{scope}: expected {key}={expected!r}, found {current!r}") + if target is None: + if match is None: + return raw + start, end = match.span() + if end < len(raw) and raw[end : end + 1] == "\n": + end += 1 + return raw[:start] + raw[end:] + replacement = f'{key}: "{target}"' + if match is not None: + return raw[: match.start()] + replacement + raw[match.end() :] + separator = "" if not raw or raw.endswith("\n") else "\n" + return f"{raw}{separator}{replacement}\n" + + +def _patch_workplan_identifier( + text: str, + *, + record_id: str, + expected: str | None, + target: str | None, +) -> str: + if not text.startswith("---"): + raise ValueError(f"{record_id}: missing YAML frontmatter") + parts = text.split("---", 2) + if len(parts) != 3: + raise ValueError(f"{record_id}: malformed YAML frontmatter") + metadata = _parse_yaml_block(parts[1].strip()) + if metadata.get("id") != record_id: + raise ValueError(f"{record_id}: workplan id does not match source file") + patched = _patch_yaml_field( + parts[1].lstrip("\n"), + key="state_hub_workstream_id", + expected=expected, + target=target, + scope=record_id, + ) + return f"---\n{patched.rstrip()}\n---{parts[2]}" + + +def _patch_task_identifier( + text: str, + *, + record_id: str, + expected: str | None, + target: str | None, +) -> str: + matched = 0 + + def replace(match: re.Match[str]) -> str: + nonlocal matched + raw = match.group(1) + metadata = _parse_yaml_block(raw.strip()) + if metadata.get("id") != record_id: + return match.group(0) + matched += 1 + patched = _patch_yaml_field( + raw, + key="state_hub_task_id", + expected=expected, + target=target, + scope=record_id, + ) + return f"```task\n{patched.rstrip()}\n```" + + patched_text = _TASK_BLOCK_RE.sub(replace, text) + if matched != 1: + raise ValueError(f"{record_id}: expected exactly one task block, found {matched}") + return patched_text + + +def _atomic_write(path: Path, content: str) -> None: + mode = path.stat().st_mode + descriptor, temporary_name = tempfile.mkstemp( + prefix=f".{path.name}.identifier-migration-", + dir=path.parent, + ) + try: + with os.fdopen(descriptor, "w", encoding="utf-8") as temporary: + temporary.write(content) + temporary.flush() + os.fsync(temporary.fileno()) + os.chmod(temporary_name, mode) + os.replace(temporary_name, path) + finally: + if os.path.exists(temporary_name): + os.unlink(temporary_name) + + +def migrate_repository_identifier_files( + plan: dict[str, Any], + *, + repo_slug: str, + confirm_plan_sha256: str, + direction: str = "forward", + execute: bool = False, +) -> dict[str, Any]: + """Validate and optionally rewrite one repository's UUID fields atomically. + + This is the file half of the governed cutover. The caller coordinates it + with State Hub's database transaction and invokes ``direction='reverse'`` + if the file or subsequent consistency phase fails. + """ + plan_sha256 = _verified_plan_seal(plan) + if confirm_plan_sha256 != plan_sha256: + raise ValueError("--confirm-plan-sha256 does not match the sealed plan") + if direction not in {"forward", "reverse"}: + raise ValueError("direction must be 'forward' or 'reverse'") + + repositories = [item for item in plan.get("repositories", []) if item.get("repo") == repo_slug] + if len(repositories) != 1: + raise ValueError(f"repository {repo_slug!r} must occur exactly once in the plan") + repository = repositories[0] + if repository.get("eligible") is not True or repository.get("atomic_unit") is not True: + raise ValueError(f"repository {repo_slug!r} is not eligible and atomic") + + if direction == "forward": + verification = verify_identifier_migration_plan(plan, repo_slug=repo_slug) + if not verification["ok"]: + reasons = "; ".join(error["reason"] for error in verification["errors"]) + raise ValueError(f"repository source preconditions failed: {reasons}") + + repo_root = Path(str(repository.get("path") or "")).resolve() + originals: dict[Path, str] = {} + patched: dict[Path, str] = {} + replacements = 0 + assignments = 0 + for mapping in repository.get("mappings", []): + action = mapping.get("action") + if action == "unchanged": + continue + if action not in {"replace", "assign"}: + raise ValueError(f"unsupported mapping action {action!r}") + if mapping.get("repo") != repo_slug: + raise ValueError("mapping repository does not match the atomic unit") + path = (repo_root / str(mapping.get("path") or "")).resolve() + if repo_root not in path.parents or not path.is_file(): + raise ValueError(f"mapping source is missing or escapes repository: {path}") + if path not in originals: + originals[path] = path.read_text(encoding="utf-8") + patched[path] = originals[path] + + old_id = mapping.get("current_uuid") + new_id = mapping.get("derived_uuid") + expected = old_id if direction == "forward" else new_id + target = new_id if direction == "forward" else old_id + record_id = str(mapping.get("record_id") or "") + if mapping.get("kind") == "workplan": + patched[path] = _patch_workplan_identifier( + patched[path], record_id=record_id, expected=expected, target=target + ) + elif mapping.get("kind") == "task": + patched[path] = _patch_task_identifier( + patched[path], record_id=record_id, expected=expected, target=target + ) + else: + raise ValueError(f"unsupported record kind {mapping.get('kind')!r}") + replacements += action == "replace" + assignments += action == "assign" + + touched = [str(path.relative_to(repo_root)) for path in sorted(patched)] + if execute: + written: list[Path] = [] + try: + for path, content in patched.items(): + if content != originals[path]: + _atomic_write(path, content) + written.append(path) + except Exception: + restore_errors: list[str] = [] + for path in reversed(written): + try: + _atomic_write(path, originals[path]) + except Exception as exc: # noqa: BLE001 + restore_errors.append(f"{path}: {exc}") + if restore_errors: + raise RuntimeError( + "file migration failed and rollback was incomplete: " + "; ".join(restore_errors) + ) + raise + + return { + "schema": "repo-manager.identifier-file-migration.v1", + "ok": True, + "executed": execute, + "direction": direction, + "repo": repo_slug, + "plan_sha256": plan_sha256, + "files_touched": touched, + "replacements": replacements, + "assignments": assignments, + "database_coordinated": False, + } diff --git a/tests/test_identifiers.py b/tests/test_identifiers.py index 6f23657..fc2b751 100644 --- a/tests/test_identifiers.py +++ b/tests/test_identifiers.py @@ -8,6 +8,7 @@ import pytest from repo_manager.identifiers import ( derive_work_record_uuid, load_fleet_namespace, + migrate_repository_identifier_files, plan_identifier_migration, scan_live_identifier_collisions, verify_identifier_migration_plan, @@ -138,3 +139,67 @@ def test_migration_verification_detects_tampering_and_source_drift(tmp_path: Pat error["reason"] == "plan SHA-256 mismatch" for error in verify_identifier_migration_plan(plan)["errors"] ) + + +def test_identifier_file_migration_dry_run_forward_and_reverse(tmp_path: Path) -> None: + repo = tmp_path / "one" + path = repo / "workplans" / "one.md" + _workplan(path, "ONE-WP-0001", "active") + old_workplan = "11111111-1111-4111-8111-111111111111" + old_task = "22222222-2222-4222-8222-222222222222" + original = path.read_text(encoding="utf-8") + original = original.replace( + "status: active\n---", + f'status: active\nstate_hub_workstream_id: "{old_workplan}"\n---', + ).replace( + "status: todo\n```", + f'status: todo\nstate_hub_task_id: "{old_task}"\n```', + ) + path.write_text(original, encoding="utf-8") + plan = plan_identifier_migration(tmp_path, "helixforge") + seal = plan["plan_sha256"] + + report = migrate_repository_identifier_files( + plan, + repo_slug="one", + confirm_plan_sha256=seal, + ) + assert report["executed"] is False + assert report["replacements"] == 2 + assert path.read_text(encoding="utf-8") == original + + report = migrate_repository_identifier_files( + plan, + repo_slug="one", + confirm_plan_sha256=seal, + execute=True, + ) + assert report["executed"] is True + migrated = path.read_text(encoding="utf-8") + assert str(derive_work_record_uuid("helixforge", "ONE-WP-0001")) in migrated + assert str(derive_work_record_uuid("helixforge", "ONE-WP-0001-T01")) in migrated + assert old_workplan not in migrated + assert old_task not in migrated + + report = migrate_repository_identifier_files( + plan, + repo_slug="one", + confirm_plan_sha256=seal, + direction="reverse", + execute=True, + ) + assert report["direction"] == "reverse" + assert path.read_text(encoding="utf-8") == original + + +def test_identifier_file_migration_requires_exact_plan_confirmation(tmp_path: Path) -> None: + repo = tmp_path / "one" + _workplan(repo / "workplans" / "one.md", "ONE-WP-0001", "active") + plan = plan_identifier_migration(tmp_path, "helixforge") + + with pytest.raises(ValueError, match="confirm-plan-sha256"): + migrate_repository_identifier_files( + plan, + repo_slug="one", + confirm_plan_sha256="0" * 64, + ) diff --git a/workplans/RMGR-WP-0005-registrar-consolidation-deterministic-ids.md b/workplans/RMGR-WP-0005-registrar-consolidation-deterministic-ids.md index f5f3127..f0fd8cc 100644 --- a/workplans/RMGR-WP-0005-registrar-consolidation-deterministic-ids.md +++ b/workplans/RMGR-WP-0005-registrar-consolidation-deterministic-ids.md @@ -8,7 +8,7 @@ status: active owner: codex topic_slug: infotech created: "2026-08-17" -updated: "2026-08-21" +updated: "2026-08-22" parent_project: prj-state-hub-retirement parent_workplan: SHR-WP-0001 related: @@ -380,6 +380,29 @@ file-first or direct-PK rewrite would split the projection. Evidence and the required transactional/alias/rollback contract are recorded in `docs/evidence/RMGR-WP-0005-helixforge-migration-readiness-2026-08-21.md`. +**Central projection gate passed in isolation (2026-08-22):** State Hub commit +`cb1b028` adds Alembic revision `b8d4f0a2c6e1`, durable old→new alias +provenance, and a repository-atomic forward/reverse executor. All 20 foreign +keys into workplans/tasks now retain their delete policy while cascading primary +key updates. An isolated PostgreSQL upgrade/downgrade rehearsal verified 20/20 +constraints in both directions and alias-table creation/removal; service tests +prove forward/reverse cascades and all-or-nothing failure, and the full State +Hub suite passes (`622 passed`). No live database or fleet file was migrated. + +**Repository file gate implemented (2026-08-22):** `rmgr identifier +migration-files` validates the plan seal and repository source before touching +bytes, requires the exact `--confirm-plan-sha256` plus an explicit `--execute`, +prepares all mapped files before replacing any, restores original bytes on a +write failure, and supports reverse. Reverse restores replaced IDs and removes +fields introduced by `action: assign`. Focused forward/dry-run/reverse tests and +the full Repo Manager suite pass (`70 passed`); Ruff is clean. + +The guard correctly rejected the 2026-08-21 fleet plan after this workplan +changed, proving source drift cannot slip into apply. Regenerate and reseal that +plan after this implementation commit. The remaining gate is a single-repository +pilot that couples database apply, authoritative file rewrite, consistency +verification, and database rollback if the file phase fails. + ## Retire the interim rule ```task