coulomb/repo-manager
4
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

fix(identifier): verify migration per repository unit

Browse source 
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a023c0-a0a3-7c03-b395-5a0d2757214d
This commit is contained in:
tegwick 2026-08-22 00:02:48 +02:00
parent 956efbb7ae
commit 887943108e
6 changed files with 49 additions and 19 deletions
Download patch file Download diff file Expand all files Collapse all files

3
src/repo_manager/cli.py
View file

@ -299,6 +299,7 @@ def main(argv: list[str] | None = None) -> int:
help="Verify a sealed migration plan against current repository sources",
)
p_id_verify.add_argument("--plan", required=True)
p_id_verify.add_argument("--repo", default=None, help="Verify one repository atomic unit")
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")
@ -648,7 +649,7 @@ def main(argv: list[str] | None = None) -> int:
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 = verify_identifier_migration_plan(plan)
result = verify_identifier_migration_plan(plan, repo_slug=args.repo)
except (OSError, TypeError, ValueError, json.JSONDecodeError) as exc:
print(json.dumps({"ok": False, "error": str(exc)}, indent=2))
return 1

24
src/repo_manager/identifiers.py
View file

@ -217,7 +217,9 @@ def plan_identifier_migration(root: Path, namespace: str) -> dict[str, Any]:
return report
def verify_identifier_migration_plan(plan: dict[str, Any]) -> dict[str, Any]:
def verify_identifier_migration_plan(
plan: dict[str, Any], *, repo_slug: str | None = None
) -> dict[str, Any]:
"""Verify a saved plan against its seal and current repository sources."""
errors: list[dict[str, str]] = []
if plan.get("schema") != "repo-manager.identifier-migration-plan.v1":
@ -243,29 +245,43 @@ def verify_identifier_migration_plan(plan: dict[str, Any]) -> dict[str, Any]:
if not plan.get("ready_to_apply"):
errors.append({"scope": "plan", "reason": "plan is not ready_to_apply"})
repositories = list(plan.get("repositories") or [])
if repo_slug is not None:
repositories = [item for item in repositories if item.get("repo") == repo_slug]
if not repositories:
errors.append({"scope": repo_slug, "reason": "repository is absent from plan"})
checked_repositories = 0
for repository in plan.get("repositories") or []:
repository_results: list[dict[str, Any]] = []
for repository in repositories:
repo = Path(str(repository.get("path") or ""))
scope = str(repository.get("repo") or repo.name or "repository")
before = len(errors)
if not repository.get("eligible"):
errors.append({"scope": scope, "reason": "repository is ineligible"})
repository_results.append({"repo": scope, "ok": False})
continue
if not repo.is_dir():
errors.append({"scope": scope, "reason": f"repository path is missing: {repo}"})
repository_results.append({"repo": scope, "ok": False})
continue
checked_repositories += 1
if head_sha(repo) != repository.get("planned_head_sha"):
errors.append({"scope": scope, "reason": "Git HEAD changed after planning"})
head_changed = head_sha(repo) != repository.get("planned_head_sha")
current_fingerprint, _source_files = source_fingerprint(repo)
if current_fingerprint != repository.get("source_fingerprint"):
errors.append({"scope": scope, "reason": "authoritative source changed after planning"})
repository_results.append(
{"repo": scope, "ok": len(errors) == before, "head_changed": head_changed}
)
return {
"schema": "repo-manager.identifier-migration-verification.v1",
"ok": not errors,
"namespace": plan.get("namespace"),
"plan_sha256": expected_seal,
"repo_filter": repo_slug,
"checked_repositories": checked_repositories,
"repository_results": repository_results,
"errors": errors,
"source_preconditions_satisfied": not errors,
"apply_authorized": False,