tegwick
cb60a00ee1
Assistant: codex Assistant-Model: gpt-6-astra Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
1175 lines
46 KiB
Python
1175 lines
46 KiB
Python
"""Deterministic work-record identifiers and live-collision preflight."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import tempfile
|
|
import uuid
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
import httpx
|
|
import yaml
|
|
|
|
from repo_manager.cache import source_fingerprint
|
|
from repo_manager.gitops import head_sha
|
|
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.record_identity import classify_record_id, scan_record_identities
|
|
from repo_manager.time import utc_now_text
|
|
|
|
DERIVATION_NAMESPACE_UUID = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009")
|
|
DERIVATION_VERSION = "repo-manager.work-record-uuid.v1"
|
|
FLEET_NAMESPACE_SCHEMA = "repo-manager.fleet-namespace.v1"
|
|
DEFAULT_NAMESPACE_CONTRACT = Path(__file__).resolve().parents[2] / "config" / "fleet-namespace.yaml"
|
|
LIVE_WORKPLAN_STATUSES = frozenset({"proposed", "ready", "active", "blocked", "backlog"})
|
|
_NAMESPACE_RE = re.compile(r"^[a-z0-9][a-z0-9.-]{0,62}$")
|
|
|
|
|
|
def load_fleet_namespace(path: Path = DEFAULT_NAMESPACE_CONTRACT) -> str:
|
|
data = yaml.safe_load(path.read_text(encoding="utf-8")) or {}
|
|
if not isinstance(data, dict) or data.get("schema") != FLEET_NAMESPACE_SCHEMA:
|
|
raise ValueError(f"unsupported fleet namespace contract: {path}")
|
|
namespace = str(data.get("namespace") or "").strip()
|
|
if not _NAMESPACE_RE.fullmatch(namespace):
|
|
raise ValueError(f"invalid fleet namespace in contract: {path}")
|
|
return namespace
|
|
|
|
|
|
def derivation_name(namespace: str, identifier: str) -> str:
|
|
"""Return the exact UTF-8 UUIDv5 name input defined by contract v1."""
|
|
namespace = namespace.strip()
|
|
identifier = identifier.strip()
|
|
if not _NAMESPACE_RE.fullmatch(namespace):
|
|
raise ValueError("namespace must be lowercase DNS-label style")
|
|
canonical_kind = next(
|
|
(
|
|
kind
|
|
for kind in ("workplan", "task")
|
|
if classify_record_id(kind, identifier) == "canonical"
|
|
),
|
|
None,
|
|
)
|
|
if canonical_kind is None:
|
|
raise ValueError("identifier must be a canonical workplan or task id")
|
|
return f"{namespace}\n{identifier}"
|
|
|
|
|
|
def derive_work_record_uuid(namespace: str, identifier: str) -> uuid.UUID:
|
|
return uuid.uuid5(DERIVATION_NAMESPACE_UUID, derivation_name(namespace, identifier))
|
|
|
|
|
|
def _migration_mapping(
|
|
*,
|
|
namespace: str,
|
|
repo: Path,
|
|
path: str,
|
|
kind: str,
|
|
identifier: str,
|
|
current_uuid: str | None,
|
|
) -> dict[str, str | None]:
|
|
derived_uuid = str(derive_work_record_uuid(namespace, identifier))
|
|
if not current_uuid:
|
|
action = "assign"
|
|
elif current_uuid == derived_uuid:
|
|
action = "unchanged"
|
|
else:
|
|
action = "replace"
|
|
return {
|
|
"repo": repo.name,
|
|
"path": path,
|
|
"kind": kind,
|
|
"record_id": identifier,
|
|
"current_uuid": current_uuid,
|
|
"derived_uuid": derived_uuid,
|
|
"action": action,
|
|
}
|
|
|
|
|
|
def scan_live_identifier_collisions(root: Path) -> dict[str, Any]:
|
|
"""Report live workplan/task identifiers that cannot safely be derived."""
|
|
by_id: dict[str, list[dict[str, str]]] = defaultdict(list)
|
|
repos = iter_repo_roots(root)
|
|
for repo in repos:
|
|
for path in iter_workplan_files(repo):
|
|
parsed = parse_workplan_file(path, repo_root=repo)
|
|
if parsed.status not in LIVE_WORKPLAN_STATUSES:
|
|
continue
|
|
if parsed.id:
|
|
by_id[parsed.id].append(
|
|
{"repo": repo.name, "path": parsed.path, "kind": "workplan", "status": parsed.status}
|
|
)
|
|
for task in parsed.tasks:
|
|
if task.id and task.status not in {"done", "cancel"}:
|
|
by_id[task.id].append(
|
|
{
|
|
"repo": repo.name,
|
|
"path": parsed.path,
|
|
"kind": "task",
|
|
"status": task.status or "unknown",
|
|
}
|
|
)
|
|
collisions = {identifier: entries for identifier, entries in sorted(by_id.items()) if len(entries) > 1}
|
|
return {
|
|
"ok": not collisions,
|
|
"root": str(root.resolve()),
|
|
"repos_scanned": [repo.name for repo in repos],
|
|
"live_identifiers": len(by_id),
|
|
"collisions": collisions,
|
|
"safe_to_derive": not collisions,
|
|
}
|
|
|
|
|
|
def plan_identifier_migration(root: Path, namespace: str) -> dict[str, Any]:
|
|
"""Plan a live-record UUID migration without changing files or databases.
|
|
|
|
The report is the durable old-to-derived mapping required by RMGR-WP-0005.
|
|
Eligibility is deliberately computed per repository: a repository with one
|
|
colliding or malformed live identifier is skipped as a whole.
|
|
"""
|
|
# Validate even for an empty fleet.
|
|
derivation_name(namespace, "RMGR-WP-0000")
|
|
root = root.resolve()
|
|
collision_report = scan_live_identifier_collisions(root)
|
|
collision_ids = set(collision_report["collisions"])
|
|
repositories: list[dict[str, Any]] = []
|
|
totals = {"repositories": 0, "eligible": 0, "skipped": 0, "records": 0, "replace": 0, "assign": 0, "unchanged": 0}
|
|
|
|
for repo in iter_repo_roots(root):
|
|
fingerprint, _source_files = source_fingerprint(repo)
|
|
mappings: list[dict[str, str | None]] = []
|
|
blockers: list[dict[str, str]] = []
|
|
for path in iter_workplan_files(repo):
|
|
parsed = parse_workplan_file(path, repo_root=repo)
|
|
if parsed.status not in LIVE_WORKPLAN_STATUSES:
|
|
continue
|
|
if not parsed.id:
|
|
blockers.append({"path": parsed.path, "reason": "live workplan has no canonical id"})
|
|
continue
|
|
if parsed.id in collision_ids:
|
|
blockers.append({"path": parsed.path, "record_id": parsed.id, "reason": "live identifier collision"})
|
|
try:
|
|
mappings.append(
|
|
_migration_mapping(
|
|
namespace=namespace,
|
|
repo=repo,
|
|
path=parsed.path,
|
|
kind="workplan",
|
|
identifier=parsed.id,
|
|
current_uuid=parsed.state_hub_workstream_id,
|
|
)
|
|
)
|
|
except ValueError as exc:
|
|
blockers.append({"path": parsed.path, "record_id": parsed.id, "reason": str(exc)})
|
|
|
|
for task in parsed.tasks:
|
|
if task.status in {"done", "cancel"}:
|
|
continue
|
|
if not task.id:
|
|
blockers.append({"path": parsed.path, "reason": "unfinished task has no canonical id"})
|
|
continue
|
|
if task.id in collision_ids:
|
|
blockers.append({"path": parsed.path, "record_id": task.id, "reason": "live identifier collision"})
|
|
try:
|
|
mappings.append(
|
|
_migration_mapping(
|
|
namespace=namespace,
|
|
repo=repo,
|
|
path=parsed.path,
|
|
kind="task",
|
|
identifier=task.id,
|
|
current_uuid=task.state_hub_task_id,
|
|
)
|
|
)
|
|
except ValueError as exc:
|
|
blockers.append({"path": parsed.path, "record_id": task.id, "reason": str(exc)})
|
|
|
|
if not mappings and not blockers:
|
|
continue
|
|
eligible = not blockers
|
|
repositories.append(
|
|
{
|
|
"repo": repo.name,
|
|
"path": str(repo),
|
|
"planned_head_sha": head_sha(repo),
|
|
"source_fingerprint": fingerprint,
|
|
"eligible": eligible,
|
|
"atomic_unit": True,
|
|
"blockers": blockers,
|
|
"mappings": mappings,
|
|
}
|
|
)
|
|
totals["repositories"] += 1
|
|
totals["eligible" if eligible else "skipped"] += 1
|
|
totals["records"] += len(mappings)
|
|
for mapping in mappings:
|
|
totals[str(mapping["action"])] += 1
|
|
|
|
report = {
|
|
"schema": "repo-manager.identifier-migration-plan.v1",
|
|
"ok": totals["skipped"] == 0,
|
|
"ready_to_apply": totals["skipped"] == 0,
|
|
"root": str(root),
|
|
"namespace": namespace.strip(),
|
|
"derivation_version": DERIVATION_VERSION,
|
|
"namespace_uuid": str(DERIVATION_NAMESPACE_UUID),
|
|
"scope": "live workplans and unfinished tasks",
|
|
"apply_policy": "all-or-nothing per repository",
|
|
"generated_at": utc_now_text(),
|
|
"totals": totals,
|
|
"repositories": repositories,
|
|
}
|
|
canonical = json.dumps(report, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
report["plan_sha256"] = hashlib.sha256(canonical).hexdigest()
|
|
return report
|
|
|
|
|
|
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":
|
|
errors.append({"scope": "plan", "reason": "unsupported schema"})
|
|
expected_seal = plan.get("plan_sha256")
|
|
unsealed = {key: value for key, value in plan.items() if key != "plan_sha256"}
|
|
canonical = json.dumps(unsealed, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
actual_seal = hashlib.sha256(canonical).hexdigest()
|
|
if expected_seal != actual_seal:
|
|
errors.append({"scope": "plan", "reason": "plan SHA-256 mismatch"})
|
|
try:
|
|
declared_namespace = load_fleet_namespace()
|
|
except (OSError, ValueError) as exc:
|
|
errors.append({"scope": "namespace", "reason": str(exc)})
|
|
declared_namespace = None
|
|
if plan.get("namespace") != declared_namespace:
|
|
errors.append(
|
|
{
|
|
"scope": "namespace",
|
|
"reason": f"plan={plan.get('namespace')!r}, declared={declared_namespace!r}",
|
|
}
|
|
)
|
|
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
|
|
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
|
|
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,
|
|
"remaining_gate": "transactional central-projection migration and rollback",
|
|
}
|
|
|
|
|
|
def _git_cutover_preflight(repo: Path, *, expected_head_sha: str | None) -> dict[str, Any]:
|
|
errors: list[str] = []
|
|
|
|
def git(*args: str) -> subprocess.CompletedProcess[str]:
|
|
return subprocess.run(
|
|
["git", *args],
|
|
cwd=repo,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
|
|
current_head = head_sha(repo)
|
|
if current_head is None:
|
|
errors.append("repository has no Git HEAD")
|
|
elif current_head != expected_head_sha:
|
|
errors.append("Git HEAD changed after migration planning")
|
|
|
|
status = git("status", "--porcelain")
|
|
if status.returncode != 0:
|
|
errors.append(status.stderr.strip() or "could not inspect Git worktree")
|
|
elif status.stdout.strip():
|
|
errors.append("worktree is not clean")
|
|
|
|
upstream = git("rev-parse", "--abbrev-ref", "--symbolic-full-name", "@{u}")
|
|
upstream_name = upstream.stdout.strip() if upstream.returncode == 0 else None
|
|
behind: int | None = None
|
|
ahead: int | None = None
|
|
if upstream_name is None:
|
|
errors.append("current branch has no upstream")
|
|
else:
|
|
counts = git("rev-list", "--left-right", "--count", "@{u}...HEAD")
|
|
try:
|
|
behind, ahead = (int(value) for value in counts.stdout.split())
|
|
except (TypeError, ValueError):
|
|
errors.append(counts.stderr.strip() or "could not compare HEAD with upstream")
|
|
else:
|
|
if behind or ahead:
|
|
errors.append("branch does not exactly match its upstream")
|
|
|
|
remote = git("remote", "get-url", "origin")
|
|
origin = remote.stdout.strip() if remote.returncode == 0 else None
|
|
if origin is None:
|
|
errors.append("origin remote is missing")
|
|
elif any(
|
|
marker in origin
|
|
for marker in ("gitea-remote", "gitea.coulomb.social", "92.205.130.254")
|
|
):
|
|
errors.append("origin targets the retired Gitea lineage")
|
|
|
|
return {
|
|
"ok": not errors,
|
|
"head_sha": current_head,
|
|
"expected_head_sha": expected_head_sha,
|
|
"upstream": upstream_name,
|
|
"behind": behind,
|
|
"ahead": ahead,
|
|
"origin": origin,
|
|
"errors": errors,
|
|
}
|
|
|
|
|
|
def _projection_migration_preflight(
|
|
mappings: list[dict[str, Any]], api_bases: list[str]
|
|
) -> dict[str, Any]:
|
|
"""Classify every replacement without assuming which side currently exists.
|
|
|
|
A projection may legitimately be mixed when forge reconciliation created
|
|
some deterministic targets before the sealed migration ran. Legacy-source
|
|
and derived-target are both convergent states. An exact task duplicate is
|
|
also convergent when the legacy row has no conflicting canonical identity
|
|
and both API representations have identical business fields; State Hub
|
|
rechecks that equivalence and proves the legacy row is unreferenced inside
|
|
the apply transaction. Other both-present and all neither-present states
|
|
remain hard refusals.
|
|
"""
|
|
duplicate_fields = (
|
|
"workplan_id",
|
|
"title",
|
|
"description",
|
|
"status",
|
|
"priority",
|
|
"assignee",
|
|
"due_date",
|
|
"blocking_reason",
|
|
"needs_human",
|
|
"intervention_note",
|
|
"parent_task_id",
|
|
)
|
|
|
|
def mergeable_task_duplicate(
|
|
mapping: dict[str, Any], current: httpx.Response, derived: httpx.Response
|
|
) -> bool:
|
|
if mapping.get("kind") != "task":
|
|
return False
|
|
try:
|
|
legacy_payload = current.json()
|
|
derived_payload = derived.json()
|
|
except ValueError:
|
|
return False
|
|
record_id = mapping.get("record_id")
|
|
return (
|
|
isinstance(legacy_payload, dict)
|
|
and isinstance(derived_payload, dict)
|
|
and legacy_payload.get("id") == mapping.get("current_uuid")
|
|
and derived_payload.get("id") == mapping.get("derived_uuid")
|
|
and legacy_payload.get("record_id") in {None, record_id}
|
|
and derived_payload.get("record_id") == record_id
|
|
and all(
|
|
legacy_payload.get(field) == derived_payload.get(field)
|
|
for field in duplicate_fields
|
|
)
|
|
)
|
|
|
|
projections: list[dict[str, Any]] = []
|
|
errors: list[dict[str, str]] = [
|
|
{
|
|
"scope": str(mapping.get("record_id") or "unknown"),
|
|
"reason": "projection-aware approval does not yet support UUID assignments",
|
|
}
|
|
for mapping in mappings
|
|
if mapping.get("action") == "assign"
|
|
]
|
|
for raw_base in api_bases:
|
|
api_base = raw_base.strip().rstrip("/")
|
|
try:
|
|
url = httpx.URL(api_base)
|
|
if (
|
|
url.scheme not in {"http", "https"}
|
|
or not url.host
|
|
or url.username
|
|
or url.password
|
|
or url.query
|
|
or url.fragment
|
|
):
|
|
raise ValueError
|
|
except (TypeError, ValueError):
|
|
errors.append(
|
|
{"scope": raw_base, "reason": "projection API base must be a plain HTTP(S) origin"}
|
|
)
|
|
continue
|
|
|
|
checks: list[dict[str, Any]] = []
|
|
try:
|
|
with httpx.Client(timeout=10.0, follow_redirects=False) as client:
|
|
for mapping in mappings:
|
|
if mapping.get("action") != "replace":
|
|
continue
|
|
route = "workplans" if mapping.get("kind") == "workplan" else "tasks"
|
|
current = client.get(f"{api_base}/{route}/{mapping.get('current_uuid')}")
|
|
derived = client.get(f"{api_base}/{route}/{mapping.get('derived_uuid')}")
|
|
status_pair = (current.status_code, derived.status_code)
|
|
state = {
|
|
(200, 404): "legacy_source",
|
|
(404, 200): "derived_target",
|
|
(200, 200): "both_present",
|
|
(404, 404): "neither_present",
|
|
}.get(status_pair, "unexpected_status")
|
|
if state == "both_present" and mergeable_task_duplicate(
|
|
mapping, current, derived
|
|
):
|
|
state = "duplicate_target"
|
|
check_ok = state in {
|
|
"legacy_source",
|
|
"derived_target",
|
|
"duplicate_target",
|
|
}
|
|
check = {
|
|
"record_id": mapping.get("record_id"),
|
|
"kind": mapping.get("kind"),
|
|
"current_status": current.status_code,
|
|
"derived_status": derived.status_code,
|
|
"state": state,
|
|
"ok": check_ok,
|
|
}
|
|
checks.append(check)
|
|
if not check_ok:
|
|
errors.append(
|
|
{
|
|
"scope": f"{api_base}:{mapping.get('record_id')}",
|
|
"reason": (
|
|
"projection is ambiguous: both legacy and derived UUIDs exist"
|
|
if state == "both_present"
|
|
else "projection is incomplete: neither legacy nor derived UUID exists"
|
|
if state == "neither_present"
|
|
else "projection returned an unexpected HTTP status"
|
|
),
|
|
}
|
|
)
|
|
except httpx.HTTPError:
|
|
errors.append({"scope": api_base, "reason": "projection API is unavailable"})
|
|
state_counts = {
|
|
state: sum(check["state"] == state for check in checks)
|
|
for state in (
|
|
"legacy_source",
|
|
"derived_target",
|
|
"duplicate_target",
|
|
"both_present",
|
|
"neither_present",
|
|
"unexpected_status",
|
|
)
|
|
}
|
|
convergent_states = {check["state"] for check in checks if check["ok"]}
|
|
mode = (
|
|
"legacy_migration"
|
|
if convergent_states == {"legacy_source"}
|
|
else "file_convergence"
|
|
if convergent_states == {"derived_target"}
|
|
else "duplicate_convergence"
|
|
if convergent_states == {"duplicate_target"}
|
|
else "mixed_convergence"
|
|
if convergent_states
|
|
else "blocked"
|
|
)
|
|
projections.append(
|
|
{
|
|
"api_base": api_base,
|
|
"ok": all(check["ok"] for check in checks),
|
|
"mode": mode,
|
|
"state_counts": state_counts,
|
|
"replacement_checks": checks,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"required": True,
|
|
"ok": bool(projections) and not errors,
|
|
"projections": projections,
|
|
"errors": errors,
|
|
}
|
|
|
|
|
|
def plan_identifier_migration_batch(
|
|
plan: dict[str, Any],
|
|
*,
|
|
repo_slugs: list[str],
|
|
projection_api_bases: list[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Pin a reviewed repository batch after source and Git preflight.
|
|
|
|
The result is an approval package, not apply authority. Live database and
|
|
file mutation still require an explicit decision citing ``batch_sha256``.
|
|
"""
|
|
plan_sha256 = _verified_plan_seal(plan)
|
|
requested = [slug.strip() for slug in repo_slugs if slug.strip()]
|
|
errors: list[dict[str, str]] = []
|
|
if not requested:
|
|
errors.append({"scope": "batch", "reason": "at least one repository is required"})
|
|
if len(set(requested)) != len(requested):
|
|
errors.append({"scope": "batch", "reason": "repository list contains duplicates"})
|
|
|
|
repositories_by_slug: dict[str, list[dict[str, Any]]] = defaultdict(list)
|
|
for repository in plan.get("repositories", []):
|
|
repositories_by_slug[str(repository.get("repo") or "")].append(repository)
|
|
|
|
projection_api_bases = list(projection_api_bases or [])
|
|
batch_repositories: list[dict[str, Any]] = []
|
|
totals = {"repositories": 0, "records": 0, "replace": 0, "assign": 0, "unchanged": 0}
|
|
for slug in requested:
|
|
matches = repositories_by_slug.get(slug, [])
|
|
if len(matches) != 1:
|
|
errors.append(
|
|
{
|
|
"scope": slug,
|
|
"reason": "repository must occur exactly once in the source plan",
|
|
}
|
|
)
|
|
continue
|
|
repository = matches[0]
|
|
source_verification = verify_identifier_migration_plan(plan, repo_slug=slug)
|
|
repo = Path(str(repository.get("path") or "")).resolve()
|
|
git_preflight = _git_cutover_preflight(
|
|
repo,
|
|
expected_head_sha=repository.get("planned_head_sha"),
|
|
)
|
|
mappings = list(repository.get("mappings") or [])
|
|
action_counts = {
|
|
action: sum(mapping.get("action") == action for mapping in mappings)
|
|
for action in ("replace", "assign", "unchanged")
|
|
}
|
|
actionable = action_counts["replace"] + action_counts["assign"]
|
|
if actionable == 0:
|
|
errors.append({"scope": slug, "reason": "repository has no migration action"})
|
|
for error in source_verification["errors"]:
|
|
errors.append({"scope": slug, "reason": error["reason"]})
|
|
for reason in git_preflight["errors"]:
|
|
errors.append({"scope": slug, "reason": reason})
|
|
projection_preflight = None
|
|
if projection_api_bases:
|
|
projection_preflight = _projection_migration_preflight(
|
|
mappings, projection_api_bases
|
|
)
|
|
errors.extend(projection_preflight["errors"])
|
|
|
|
batch_repositories.append(
|
|
{
|
|
"repo": slug,
|
|
"path": str(repo),
|
|
"planned_head_sha": repository.get("planned_head_sha"),
|
|
"source_fingerprint": repository.get("source_fingerprint"),
|
|
"source_verified": source_verification["ok"],
|
|
"git_preflight": git_preflight,
|
|
"projection_preflight": projection_preflight,
|
|
"mapping_counts": {"records": len(mappings), **action_counts},
|
|
"ready": (
|
|
source_verification["ok"]
|
|
and git_preflight["ok"]
|
|
and actionable > 0
|
|
and (projection_preflight is None or projection_preflight["ok"])
|
|
),
|
|
}
|
|
)
|
|
totals["repositories"] += 1
|
|
totals["records"] += len(mappings)
|
|
for action, count in action_counts.items():
|
|
totals[action] += count
|
|
|
|
result = {
|
|
"schema": "repo-manager.identifier-migration-batch.v1",
|
|
"ok": not errors,
|
|
"ready_for_approval": not errors,
|
|
"apply_authorized": False,
|
|
"approval_required": True,
|
|
"projection_api_bases": [base.strip().rstrip("/") for base in projection_api_bases],
|
|
"namespace": plan.get("namespace"),
|
|
"source_plan_sha256": plan_sha256,
|
|
"batch_policy": "repository-atomic, sequential, stop on first failure",
|
|
"rollback_order": "reverse files if written, then reverse central projection",
|
|
"generated_at": utc_now_text(),
|
|
"totals": totals,
|
|
"repositories": batch_repositories,
|
|
"errors": errors,
|
|
}
|
|
canonical = json.dumps(result, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
result["batch_sha256"] = hashlib.sha256(canonical).hexdigest()
|
|
return result
|
|
|
|
|
|
def verify_identifier_migration_batch(
|
|
batch: dict[str, Any], *, plan: dict[str, Any]
|
|
) -> dict[str, Any]:
|
|
"""Verify a saved batch seal and repeat every selected source/Git check."""
|
|
errors: list[dict[str, str]] = []
|
|
expected_batch_sha256 = batch.get("batch_sha256")
|
|
unsealed = {key: value for key, value in batch.items() if key != "batch_sha256"}
|
|
canonical = json.dumps(unsealed, sort_keys=True, separators=(",", ":")).encode("utf-8")
|
|
actual_batch_sha256 = hashlib.sha256(canonical).hexdigest()
|
|
if batch.get("schema") != "repo-manager.identifier-migration-batch.v1":
|
|
errors.append({"scope": "batch", "reason": "unsupported batch schema"})
|
|
if expected_batch_sha256 != actual_batch_sha256:
|
|
errors.append({"scope": "batch", "reason": "batch SHA-256 mismatch"})
|
|
|
|
try:
|
|
plan_sha256 = _verified_plan_seal(plan)
|
|
except (OSError, ValueError) as exc:
|
|
errors.append({"scope": "plan", "reason": str(exc)})
|
|
plan_sha256 = None
|
|
if batch.get("source_plan_sha256") != plan_sha256:
|
|
errors.append({"scope": "plan", "reason": "batch source plan SHA-256 mismatch"})
|
|
if batch.get("apply_authorized") is not False or batch.get("approval_required") is not True:
|
|
errors.append(
|
|
{
|
|
"scope": "batch",
|
|
"reason": "batch must remain unauthorized and approval-required",
|
|
}
|
|
)
|
|
|
|
saved_repositories = list(batch.get("repositories") or [])
|
|
repo_slugs = [str(item.get("repo") or "") for item in saved_repositories]
|
|
if not repo_slugs or any(not slug for slug in repo_slugs):
|
|
errors.append({"scope": "batch", "reason": "batch has no valid repository scope"})
|
|
fresh = None
|
|
else:
|
|
try:
|
|
fresh = plan_identifier_migration_batch(
|
|
plan,
|
|
repo_slugs=repo_slugs,
|
|
projection_api_bases=list(batch.get("projection_api_bases") or []),
|
|
)
|
|
except (OSError, ValueError) as exc:
|
|
errors.append({"scope": "batch", "reason": str(exc)})
|
|
fresh = None
|
|
|
|
if fresh is not None:
|
|
errors.extend(fresh["errors"])
|
|
fresh_by_slug = {item["repo"]: item for item in fresh["repositories"]}
|
|
for saved in saved_repositories:
|
|
slug = str(saved.get("repo") or "")
|
|
current = fresh_by_slug.get(slug)
|
|
if current is None:
|
|
errors.append({"scope": slug, "reason": "repository disappeared from batch"})
|
|
continue
|
|
for field in ("planned_head_sha", "source_fingerprint", "mapping_counts"):
|
|
if saved.get(field) != current.get(field):
|
|
errors.append({"scope": slug, "reason": f"saved {field} differs from plan"})
|
|
|
|
return {
|
|
"schema": "repo-manager.identifier-migration-batch-verification.v1",
|
|
"ok": not errors,
|
|
"ready_for_decision": not errors,
|
|
"apply_authorized": False,
|
|
"approval_required": True,
|
|
"batch_sha256": expected_batch_sha256,
|
|
"source_plan_sha256": plan_sha256,
|
|
"repositories": repo_slugs,
|
|
"errors": errors,
|
|
}
|
|
|
|
|
|
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<prefix>{re.escape(key)}:\s*)(?P<value>.*)$", 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 ensure_missing_work_record_identifiers(
|
|
repo_root: Path,
|
|
*,
|
|
namespace: str | None = None,
|
|
execute: bool = True,
|
|
) -> dict[str, Any]:
|
|
"""Assign only missing canonical UUIDs, deterministically, in one repository.
|
|
|
|
Existing UUIDs are deliberately preserved. Replacing a legacy UUID is a
|
|
separately sealed migration; ordinary registration must never perform it.
|
|
Every source file is validated and rendered before any file is changed.
|
|
"""
|
|
repo_root = repo_root.resolve()
|
|
namespace = namespace or load_fleet_namespace()
|
|
identity = scan_record_identities(repo_root)
|
|
# Completed, already-bound suffix tasks are historical evidence, not
|
|
# allocation candidates. Keep conformance reporting strict while allowing
|
|
# ordinary registration to preserve those bindings. Local aliases still
|
|
# require the explicit parent allowlist in the kind registry.
|
|
historical_sources: set[str] = set()
|
|
for path in iter_workplan_files(repo_root):
|
|
parsed = parse_workplan_file(path, repo_root=repo_root)
|
|
if parsed.status not in {"finished", "archived", "completed"} or not parsed.id:
|
|
continue
|
|
for number, task in enumerate(parsed.tasks, start=1):
|
|
if task.status not in {"done", "cancel"} or not task.state_hub_task_id:
|
|
continue
|
|
# RAILIANCE-WP-0004 is a closed historical workplan whose source
|
|
# used the old I01..I07 item labels. These exact, UUID-bound
|
|
# labels are preserved like the older suffix records below; they
|
|
# must not become a general I-suffix grandfather rule.
|
|
legacy_item = (
|
|
parsed.id == "RAILIANCE-WP-0004"
|
|
and re.fullmatch(r"RAILIANCE-WP-0004-I(?:0[1-7])", task.id or "")
|
|
)
|
|
if not (
|
|
re.fullmatch(re.escape(parsed.id) + r"-T[0-9]{2}[a-z]", task.id or "")
|
|
or legacy_item
|
|
):
|
|
continue
|
|
try:
|
|
uuid.UUID(task.state_hub_task_id)
|
|
except ValueError:
|
|
continue
|
|
historical_sources.add(f"{parsed.path}#task-block-{number}")
|
|
invalid = [
|
|
item for item in identity["invalid_identifiers"]
|
|
if item["source"] not in historical_sources
|
|
]
|
|
historical = [
|
|
item for item in identity["invalid_identifiers"]
|
|
if item["source"] in historical_sources
|
|
]
|
|
if invalid or identity["identity_collisions"]:
|
|
raise ValueError(
|
|
"work-record identities are not safe to derive: "
|
|
f"invalid={invalid!r}, "
|
|
f"collisions={identity['identity_collisions']!r}"
|
|
)
|
|
|
|
rendered: dict[Path, str] = {}
|
|
assignments: list[dict[str, str]] = []
|
|
skipped: list[dict[str, str]] = []
|
|
for path in iter_workplan_files(repo_root):
|
|
parsed = parse_workplan_file(path, repo_root=repo_root)
|
|
text = path.read_text(encoding="utf-8")
|
|
changed = False
|
|
if parsed.id and not parsed.state_hub_workstream_id:
|
|
if classify_record_id("workplan", parsed.id) != "canonical":
|
|
skipped.append(
|
|
{"kind": "workplan", "record_id": parsed.id, "path": parsed.path}
|
|
)
|
|
else:
|
|
derived = str(derive_work_record_uuid(namespace, parsed.id))
|
|
text = _patch_workplan_identifier(
|
|
text,
|
|
record_id=parsed.id,
|
|
expected=None,
|
|
target=derived,
|
|
)
|
|
assignments.append(
|
|
{
|
|
"kind": "workplan",
|
|
"record_id": parsed.id,
|
|
"uuid": derived,
|
|
"path": parsed.path,
|
|
}
|
|
)
|
|
changed = True
|
|
for task in parsed.tasks:
|
|
if not task.id or task.state_hub_task_id:
|
|
continue
|
|
if classify_record_id("task", task.id) != "canonical":
|
|
skipped.append({"kind": "task", "record_id": task.id, "path": parsed.path})
|
|
continue
|
|
derived = str(derive_work_record_uuid(namespace, task.id))
|
|
text = _patch_task_identifier(
|
|
text,
|
|
record_id=task.id,
|
|
expected=None,
|
|
target=derived,
|
|
)
|
|
assignments.append(
|
|
{
|
|
"kind": "task",
|
|
"record_id": task.id,
|
|
"uuid": derived,
|
|
"path": parsed.path,
|
|
}
|
|
)
|
|
changed = True
|
|
if changed:
|
|
rendered[path] = text
|
|
|
|
if skipped:
|
|
raise ValueError(
|
|
"missing UUIDs on grandfathered or noncanonical records require "
|
|
f"an explicit identity decision: {skipped!r}"
|
|
)
|
|
if execute:
|
|
for path, content in rendered.items():
|
|
_atomic_write(path, content)
|
|
return {
|
|
"schema": "repo-manager.identifier-assignment.v1",
|
|
"ok": True,
|
|
"executed": execute,
|
|
"repo": repo_root.name,
|
|
"namespace": namespace,
|
|
"assignments": assignments,
|
|
"historical_invalid_identifiers": historical,
|
|
"files_changed": sorted(str(path.relative_to(repo_root)) for path in rendered),
|
|
}
|
|
|
|
|
|
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,
|
|
}
|
|
|
|
|
|
def migrate_repository_projection(
|
|
plan: dict[str, Any],
|
|
*,
|
|
repo_slug: str,
|
|
confirm_plan_sha256: str,
|
|
api_base: str,
|
|
direction: str = "forward",
|
|
expected_instance_label: str | None = "railiance01",
|
|
transport: httpx.BaseTransport | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Apply or reverse the sealed projection half on the authoritative hub."""
|
|
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'")
|
|
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}")
|
|
repositories = [
|
|
item for item in plan.get("repositories", []) if item.get("repo") == repo_slug
|
|
]
|
|
repository = repositories[0]
|
|
git_preflight = _git_cutover_preflight(
|
|
Path(str(repository.get("path") or "")).resolve(),
|
|
expected_head_sha=repository.get("planned_head_sha"),
|
|
)
|
|
if not git_preflight["ok"]:
|
|
raise ValueError(
|
|
"repository Git preconditions failed: " + "; ".join(git_preflight["errors"])
|
|
)
|
|
api_base = api_base.strip().rstrip("/")
|
|
try:
|
|
url = httpx.URL(api_base)
|
|
if (
|
|
url.scheme not in {"http", "https"}
|
|
or not url.host
|
|
or url.username
|
|
or url.password
|
|
or url.query
|
|
or url.fragment
|
|
):
|
|
raise ValueError
|
|
except (TypeError, ValueError) as exc:
|
|
raise ValueError("projection API base must be a plain HTTP(S) origin") from exc
|
|
|
|
headers = {
|
|
"Idempotency-Key": f"rmgr-identifier-migration:{direction}:{repo_slug}:{plan_sha256}",
|
|
"X-StateHub-Source-Agent": "repo-manager",
|
|
}
|
|
try:
|
|
with httpx.Client(
|
|
base_url=api_base,
|
|
timeout=httpx.Timeout(120.0, connect=5.0),
|
|
follow_redirects=False,
|
|
transport=transport,
|
|
) as client:
|
|
health = client.get("/state/health")
|
|
health.raise_for_status()
|
|
identity = health.json()
|
|
if identity.get("instance_role") != "primary" or (
|
|
expected_instance_label is not None
|
|
and identity.get("instance_label") != expected_instance_label
|
|
):
|
|
raise ValueError("identifier migration requires the expected primary State Hub")
|
|
response = client.post(
|
|
f"/identifier-migrations/repositories/{repo_slug}/"
|
|
f"{'apply' if direction == 'forward' else 'reverse'}",
|
|
json={
|
|
"plan": plan,
|
|
"expected_plan_sha256": plan_sha256,
|
|
"primary_confirmed": True,
|
|
},
|
|
headers=headers,
|
|
)
|
|
response.raise_for_status()
|
|
except httpx.HTTPStatusError as exc:
|
|
try:
|
|
detail = exc.response.json()
|
|
except ValueError:
|
|
detail = exc.response.text[:500]
|
|
raise ValueError(f"projection migration rejected: {detail}") from exc
|
|
except httpx.HTTPError as exc:
|
|
raise ValueError(f"projection migration unavailable: {exc}") from exc
|
|
receipt = response.json()
|
|
return {
|
|
"schema": "repo-manager.identifier-projection-migration.v1",
|
|
"ok": True,
|
|
"direction": direction,
|
|
"repo": repo_slug,
|
|
"plan_sha256": plan_sha256,
|
|
"api_base": api_base,
|
|
"instance": identity,
|
|
"source_verification": verification,
|
|
"git_preflight": git_preflight,
|
|
"receipt": receipt,
|
|
}
|