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

fix: preserve scoped Custodian historical work identities

Browse source 
Assistant: codex
Assistant-Model: gpt-5.6-luna
Assistant-Session: 01a07ff8-19d0-7820-b4d0-1353833cb7fc
This commit is contained in:
tegwick 2026-09-09 12:47:43 +02:00
parent 6854ea10b2
commit 89524d83e4
5 changed files with 657 additions and 10 deletions
Download patch file Download diff file Expand all files Collapse all files

68
src/repo_manager/record_identity.py
View file

@ -24,7 +24,11 @@ def load_kind_registry(path: Path | None = None) -> dict[str, Any]:
def classify_record_id(
kind: str, identifier: str, *, registry_path: Path | None = None
kind: str,
identifier: str,
*,
parent_id: str | None = None,
registry_path: Path | None = None,
) -> str | None:
"""Return canonical/grandfathered when the canon registry accepts an id."""
registry = load_kind_registry(registry_path)
@ -38,35 +42,77 @@ def classify_record_id(
str(legacy.get("pattern") or ""), identifier
):
return "grandfathered"
# Historical task blocks may use a local TNN label. Accept these only
# under an explicitly enumerated parent workplan; the parent scope is
# part of identity and no source id/UUID is rewritten.
if parent_id:
for legacy in entry.get("scoped_legacy_patterns") or []:
if (
legacy.get("grandfathered")
and re.fullmatch(str(legacy.get("pattern") or ""), identifier)
and re.fullmatch(str(legacy.get("parent_pattern") or ""), parent_id)
):
return "grandfathered"
return None
def scan_record_identities(repo_root: Path, *, registry_path: Path | None = None) -> dict[str, Any]:
"""Describe invalid ids, benign duplicate sources, and hard collisions."""
occurrences: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list)
occurrences: dict[tuple[str, str, str | None], list[dict[str, Any]]] = defaultdict(list)
invalid: list[dict[str, str]] = []
def add(kind: str, identifier: str | None, record_uuid: str | None, source: str) -> None:
def add(
kind: str,
identifier: str | None,
record_uuid: str | None,
source: str,
*,
parent_id: str | None = None,
) -> None:
if not identifier:
return
classification = classify_record_id(kind, identifier, registry_path=registry_path)
item = {"kind": kind, "id": identifier, "uuid": record_uuid, "source": source}
occurrences[(kind, identifier)].append(item)
if classification is None:
classification = classify_record_id(
kind, identifier, parent_id=parent_id, registry_path=registry_path
)
item = {
"kind": kind,
"id": identifier,
"uuid": record_uuid,
"source": source,
**({"parent_id": parent_id} if parent_id else {}),
}
# Only the explicitly scoped local aliases use parent scope. Qualified
# canonical IDs remain globally collision checked.
local_alias = (
kind == "task"
and parent_id is not None
and re.fullmatch(r"T[0-9]{2}", identifier) is not None
and classification == "grandfathered"
)
occurrences[(kind, identifier, parent_id if local_alias else None)].append(item)
if classification is None or (local_alias and not record_uuid):
invalid.append({"kind": kind, "id": identifier, "source": source})
for path in iter_workplan_files(repo_root):
workplan = parse_workplan_file(path, repo_root=repo_root)
add("workplan", workplan.id, workplan.state_hub_workstream_id, workplan.path)
for number, task in enumerate(workplan.tasks, start=1):
add("task", task.id, task.state_hub_task_id, f"{workplan.path}#task-block-{number}")
add(
"task",
task.id,
task.state_hub_task_id,
f"{workplan.path}#task-block-{number}",
parent_id=workplan.id,
)
for path in iter_record_files(repo_root):
for number, record in enumerate(parse_record_file(path, repo_root=repo_root), start=1):
add(record.kind, record.id, record.uuid, f"{record.source_path}#record-block-{number}")
duplicates: list[dict[str, Any]] = []
collisions: list[dict[str, Any]] = []
for (kind, identifier), items in sorted(occurrences.items()):
for (kind, identifier, _parent_id), items in sorted(
occurrences.items(), key=lambda pair: (pair[0][0], pair[0][1], pair[0][2] or "")
):
if len(items) < 2:
continue
uuids = {item["uuid"] for item in items}
@ -74,8 +120,10 @@ def scan_record_identities(repo_root: Path, *, registry_path: Path | None = None
"kind": kind,
"id": identifier,
"uuid": items[0]["uuid"] if len(uuids) == 1 else None,
"sources": [item["source"] for item in items],
"sources": sorted(item["source"] for item in items),
}
if items[0].get("parent_id"):
result["parent_id"] = items[0]["parent_id"]
if len(uuids) == 1 and None not in uuids:
duplicates.append(result)
else: