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

feat: generate repository rename adoption plans
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
Details
CI Smoke / container-smoke (push) Successful in 1s
Details
Build and Publish Multi-Context Image / build-and-push (push) Successful in 25s
Details

Browse source 
Assistant: codex
Assistant-Model: gpt-5.6-sol
Assistant-Session: 01a049a4-ee9f-78e1-9d66-2cb0f9bea3e3
This commit is contained in:
tegwick 2026-08-29 13:08:02 +02:00
parent bdf3a1fdf8
commit ef0b6df3b8
7 changed files with 1294 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

2
WORK-RECORDS.md
View file

@ -326,7 +326,7 @@
| task | STATE-WP-0085-T03 | done | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md |
| task | STATE-WP-0085-T04 | done | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md |
| task | STATE-WP-0085-T05 | done | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md |
| task | STATE-WP-0085-T06 | todo | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md |
| task | STATE-WP-0085-T06 | done | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md |
| task | STATE-WP-0085-T07 | todo | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md |
| task | STATE-WP-0085-T08 | todo | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md |
| task | STATE-WP-0085-T09 | wait | — | workplans/STATE-WP-0085-repository-lineage-preserving-rename.md |

2
pyproject.toml
View file

@ -34,6 +34,7 @@ packages = ["api", "mcp_server", "task_flow_engine"]
artifacts = [
"custodian_cli.py",
"repository_rename_cli.py",
"repository_rename_workplan.py",
"statehub_register.py",
"scripts/consistency_check.py",
"scripts/repo_sync.py",
@ -45,6 +46,7 @@ artifacts = [
[tool.hatch.build.targets.wheel.force-include]
"custodian_cli.py" = "custodian_cli.py"
"repository_rename_cli.py" = "repository_rename_cli.py"
"repository_rename_workplan.py" = "repository_rename_workplan.py"
"statehub_register.py" = "statehub_register.py"
"scripts/consistency_check.py" = "scripts/consistency_check.py"
"scripts/ensure_gitignore_claude_rules.py" = "scripts/ensure_gitignore_claude_rules.py"

132
repository_rename_cli.py
View file

@ -16,6 +16,15 @@ import uuid
from pathlib import Path
from typing import Any, Callable
from repository_rename_workplan import (
WorkplanGenerationError,
build_generation_snapshot,
content_checksum,
default_output_path,
render_workplan,
write_workplan_exclusive,
)
CLI_SCHEMA_VERSION = "state-hub.repository-rename-cli.v1"
DEFAULT_API_BASE = os.environ.get("API_BASE", "http://127.0.0.1:8000")
@ -120,7 +129,9 @@ def _api_request(
method: str,
path: str,
body: dict[str, Any] | None = None,
) -> dict[str, Any]:
*,
expected: type | tuple[type, ...] = dict,
) -> Any:
data = json.dumps(body).encode("utf-8") if body is not None else None
headers = dict(_HEADERS)
if data is not None:
@ -165,9 +176,14 @@ def _api_request(
"State Hub returned an invalid JSON response",
code="invalid_state_hub_response",
) from exc
if not isinstance(payload, dict):
if not isinstance(payload, expected):
expected_name = (
", ".join(item.__name__ for item in expected)
if isinstance(expected, tuple)
else expected.__name__
)
raise RenameCLIError(
"State Hub returned a non-object response",
f"State Hub returned JSON with the wrong shape; expected {expected_name}",
code="invalid_state_hub_response",
)
return payload
@ -361,6 +377,9 @@ def _emit(payload: dict[str, Any], *, as_json: bool) -> None:
print(f" blockers: {len(result.get('blockers') or [])}", file=stream)
if result.get("preflight_file"):
print(f" private preflight: {result['preflight_file']}", file=stream)
elif public["command"] == "generate-workplan" and result.get("output"):
print(f" workplan: {result['output']}", file=stream)
print(f" workplan id: {result.get('workplan_id')}", file=stream)
error = public.get("error")
if error:
print(f" error: {error.get('message')}", file=stream)
@ -490,6 +509,88 @@ def cmd_preflight(args: argparse.Namespace) -> None:
_run(args, action)
def cmd_generate_workplan(args: argparse.Namespace) -> None:
def action() -> dict[str, Any]:
if args.output and Path(args.output).expanduser().exists():
raise RenameCLIError(
f"Refusing to overwrite existing workplan {args.output}",
code="output_exists",
)
repo = _resolve_repo(args, args.old_slug)
workplans = _api_request(
args.api_base,
"GET",
f"/workplans/?repo_id={urllib.parse.quote(str(repo['id']), safe='')}",
expected=list,
)
if args.preflight_file:
wrapper = _load_json(args.preflight_file, expected=dict)
if (
wrapper.get("schema_version") != CLI_SCHEMA_VERSION
or wrapper.get("command") != "preflight"
or str(wrapper.get("repo_id")) != str(repo["id"])
):
raise RenameCLIError(
"Preflight file is not for this registered repository",
code="invalid_preflight_file",
)
report = wrapper.get("result") or {}
else:
report = _api_request(
args.api_base,
"POST",
f"/repos/{repo['id']}/rename/preflight",
{"new_slug": args.new_slug, "queued_edge_writes": []},
)
if (
report.get("old_slug") != args.old_slug
or report.get("new_slug") != args.new_slug
):
raise RenameCLIError(
"Preflight snapshot does not address the requested rename",
code="invalid_preflight_file",
)
try:
snapshot = build_generation_snapshot(
repo=repo,
preflight=report,
workplans=workplans,
repo_path_override=args.repo_path,
)
content = render_workplan(snapshot, owner=args.owner)
output = (
Path(args.output).expanduser().resolve()
if args.output
else default_output_path(snapshot)
)
write_workplan_exclusive(output, content)
except WorkplanGenerationError as exc:
raise RenameCLIError(str(exc), code=exc.code) from exc
return _envelope(
"generate-workplan",
state="achieved",
ok=True,
result={
"output": str(output),
"workplan_id": snapshot.workplan_id,
"workplan_prefix": snapshot.prefix,
"content_checksum": content_checksum(content),
"preflight_report_checksum": report.get("report_checksum"),
"safe_to_apply_at_capture": bool(report.get("safe_to_apply")),
"path_available": snapshot.path_available,
"state_hub_uuid_fields_written": [],
},
repo_id=str(repo["id"]),
next_safe_action={
"action": "review-generated-workplan",
"requires_confirmation": False,
"command": None,
},
)
_run(args, action)
def cmd_start(args: argparse.Namespace) -> None:
def action() -> dict[str, Any]:
operation_id = _validated_uuid(args.operation_id)
@ -763,6 +864,31 @@ def configure_repo_commands(subparsers: argparse._SubParsersAction) -> None:
_common(preflight)
preflight.set_defaults(func=cmd_preflight)
generate = rename_sub.add_parser(
"generate-workplan",
help="Generate a repository-native rename adoption workplan",
)
generate.add_argument("old_slug")
generate.add_argument("new_slug")
generate.add_argument(
"--preflight-file",
default=None,
help="Use a private mode-0600 preflight envelope instead of refreshing",
)
generate.add_argument(
"--repo-path",
default=None,
help="Visible target checkout used for workplan convention discovery",
)
generate.add_argument(
"--output",
default=None,
help="Exclusive output path; defaults to the target workplans directory",
)
generate.add_argument("--owner", default="codex")
_common(generate)
generate.set_defaults(func=cmd_generate_workplan)
start = rename_sub.add_parser(
"start",
help="Create an idempotent operation journal from a private preflight file",

724
repository_rename_workplan.py Normal file
View file

@ -0,0 +1,724 @@
"""Deterministic repository-native workplan generation for repository renames."""
from __future__ import annotations
import hashlib
import json
import os
import re
import uuid
from collections import Counter
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Iterable
import yaml
_WORKPLAN_ID = re.compile(
r"^(?P<prefix>[A-Z][A-Z0-9]*(?:-[A-Z][A-Z0-9]*)*-WP)-"
r"(?P<number>\d+)$"
)
_DECLARED_PREFIX = re.compile(
r"workplan\s+prefix[^`\n]*`?(?P<prefix>[A-Z][A-Z0-9-]*-WP)-?`?",
re.IGNORECASE,
)
_FRONTMATTER = re.compile(r"\A---\s*\n(?P<yaml>.*?)\n---(?:\s*\n|\Z)", re.DOTALL)
class WorkplanGenerationError(RuntimeError):
"""Raised when a safe repository-native plan cannot be generated."""
def __init__(self, message: str, *, code: str) -> None:
super().__init__(message)
self.code = code
@dataclass(frozen=True)
class WorkplanGenerationSnapshot:
repo: dict[str, Any]
preflight: dict[str, Any]
workplans: tuple[dict[str, Any], ...]
prefix: str
workplan_id: str
task_width: int
topic_slug: str
repo_path: str | None
path_available: bool
observed_workplan_ids: tuple[str, ...]
archived_workplan_ids: tuple[str, ...]
@property
def old_slug(self) -> str:
return str(self.preflight["old_slug"])
@property
def new_slug(self) -> str:
return str(self.preflight["new_slug"])
def _frontmatter(path: Path) -> dict[str, Any]:
try:
text = path.read_text(encoding="utf-8")
except OSError:
return {}
match = _FRONTMATTER.match(text)
if match is None:
return {}
try:
value = yaml.safe_load(match.group("yaml")) or {}
except yaml.YAMLError:
return {}
return value if isinstance(value, dict) else {}
def _local_metadata(repo_path: Path | None) -> tuple[list[dict[str, Any]], str | None]:
if repo_path is None or not repo_path.is_dir():
return [], None
declared: str | None = None
agents = repo_path / "AGENTS.md"
try:
match = _DECLARED_PREFIX.search(agents.read_text(encoding="utf-8"))
except OSError:
match = None
if match:
declared = match.group("prefix").upper().rstrip("-")
rows: list[dict[str, Any]] = []
workplans = repo_path / "workplans"
if workplans.is_dir():
for path in sorted(workplans.rglob("*.md")):
meta = _frontmatter(path)
record_id = str(meta.get("id") or "").upper()
if not _WORKPLAN_ID.match(record_id):
filename = path.name
if re.match(r"^\d{6}-", filename):
filename = filename[7:]
filename_match = re.match(
r"^(?P<id>[A-Z][A-Z0-9-]*-WP-\d+)", filename
)
record_id = filename_match.group("id") if filename_match else ""
if _WORKPLAN_ID.match(record_id):
rows.append(
{
"record_id": record_id,
"topic_slug": meta.get("topic_slug"),
"status": meta.get("status"),
"archived": "archived" in path.relative_to(workplans).parts,
"state_hub_workstream_id": meta.get(
"state_hub_workstream_id"
),
}
)
return rows, declared
def _record_id(row: dict[str, Any]) -> str | None:
candidates = (
row.get("record_id"),
row.get("slug"),
row.get("backing_filename"),
row.get("backing_relative_path"),
)
for candidate in candidates:
if not candidate:
continue
match = re.search(
r"([A-Z][A-Z0-9]*(?:-[A-Z][A-Z0-9]*)*-WP-\d+)",
str(candidate).upper(),
)
if match and _WORKPLAN_ID.match(match.group(1)):
return match.group(1)
return None
def _select_prefix(
records: Iterable[dict[str, Any]], declared_prefix: str | None
) -> tuple[str, list[str]]:
ids = sorted({record_id for row in records if (record_id := _record_id(row))})
matches = [_WORKPLAN_ID.match(record_id) for record_id in ids]
prefixes = Counter(
match.group("prefix") for match in matches if match is not None
)
if declared_prefix:
if prefixes and declared_prefix not in prefixes:
raise WorkplanGenerationError(
f"Declared workplan prefix {declared_prefix!r} conflicts with "
f"observed prefixes {sorted(prefixes)!r}",
code="workplan_prefix_conflict",
)
return declared_prefix, ids
if not prefixes:
raise WorkplanGenerationError(
"Repository has no established workplan prefix; scaffold or declare one first",
code="workplan_prefix_missing",
)
if len(prefixes) != 1:
raise WorkplanGenerationError(
f"Repository workplan prefix is ambiguous: {sorted(prefixes)!r}",
code="workplan_prefix_ambiguous",
)
return next(iter(prefixes)), ids
def _next_workplan_id(prefix: str, ids: Iterable[str]) -> str:
numbers: list[int] = []
widths: list[int] = []
for record_id in ids:
match = _WORKPLAN_ID.match(record_id)
if match and match.group("prefix") == prefix:
raw = match.group("number")
numbers.append(int(raw))
widths.append(len(raw))
width = max([4, *widths])
return f"{prefix}-{max(numbers, default=0) + 1:0{width}d}"
def _repo_path(repo: dict[str, Any], override: str | None) -> Path | None:
candidates = [override, repo.get("local_path")]
candidates.extend((repo.get("host_paths") or {}).values())
for candidate in candidates:
if candidate:
path = Path(str(candidate)).expanduser()
if path.is_dir():
return path.resolve()
if override:
return Path(override).expanduser().resolve()
return None
def build_generation_snapshot(
*,
repo: dict[str, Any],
preflight: dict[str, Any],
workplans: Iterable[dict[str, Any]],
repo_path_override: str | None = None,
) -> WorkplanGenerationSnapshot:
"""Normalize volatile API data into a deterministic render snapshot."""
try:
repo_id = str(uuid.UUID(str(repo.get("id"))))
except (TypeError, ValueError) as exc:
raise WorkplanGenerationError(
"Registered repository is missing its stable State Hub UUID",
code="repository_identity_missing",
) from exc
if repo_id != str(preflight.get("repo_id")):
raise WorkplanGenerationError(
"Preflight repository UUID does not match the registered repository",
code="preflight_repository_mismatch",
)
if not preflight.get("old_slug") or not preflight.get("new_slug"):
raise WorkplanGenerationError(
"Preflight snapshot is missing old/new repository slugs",
code="invalid_preflight_snapshot",
)
if not re.match(
r"^\d{4}-\d{2}-\d{2}", str(preflight.get("preflighted_at") or "")
):
raise WorkplanGenerationError(
"Preflight snapshot is missing a stable capture date",
code="invalid_preflight_snapshot",
)
path = _repo_path(repo, repo_path_override)
local_rows, declared_prefix = _local_metadata(
path if path and path.is_dir() else None
)
api_rows = [dict(row) for row in workplans]
all_rows = [*api_rows, *local_rows]
prefix, ids = _select_prefix(all_rows, declared_prefix)
workplan_id = _next_workplan_id(prefix, ids)
topic_counts = Counter(
str(row.get("topic_slug"))
for row in local_rows
if row.get("topic_slug")
)
topic_slug = (
topic_counts.most_common(1)[0][0]
if topic_counts
else str(repo.get("domain_slug") or "repository-rename")
)
archived = sorted(
{
record_id
for row in all_rows
if (record_id := _record_id(row))
and (
bool(row.get("archived"))
or str(row.get("status", "")).lower() == "archived"
or bool(row.get("backing_archived"))
)
}
)
return WorkplanGenerationSnapshot(
repo=dict(repo),
preflight=dict(preflight),
workplans=tuple(sorted(api_rows, key=lambda row: str(row.get("slug", "")))),
prefix=prefix,
workplan_id=workplan_id,
task_width=2,
topic_slug=topic_slug,
repo_path=str(path) if path else None,
path_available=bool(path and path.is_dir()),
observed_workplan_ids=tuple(ids),
archived_workplan_ids=tuple(archived),
)
def _json_detail(item: dict[str, Any]) -> str:
safe = {
key: value
for key, value in item.items()
if key not in {"preflight_token", "authorization", "credential", "password"}
}
return json.dumps(safe, sort_keys=True, separators=(",", ":"))
def _risk_register(snapshot: WorkplanGenerationSnapshot) -> str:
report = snapshot.preflight
rows: list[tuple[str, str, str]] = []
for severity, key in (("blocker", "blockers"), ("warning", "warnings")):
for item in report.get(key) or []:
rows.append(
(
severity,
str(item.get("code") or "unnamed"),
_json_detail(item),
)
)
for handoff in (report.get("affected") or {}).get("external_handoffs") or []:
rows.append(
(
"external-handoff",
str(handoff),
"Owning repository must be recorded during inventory",
)
)
forge_identity = (report.get("current") or {}).get("forge_identity") or {}
if forge_identity.get("verification_state") != "verified":
rows.append(
(
"blocker",
"forge-identity-unverified",
"Immutable Forge identity must be verified before mutation",
)
)
if not (report.get("current") or {}).get("forge_available", False):
rows.append(
(
"blocker",
"forge-repository-unreadable",
"Private/unreadable Forge state cannot be inferred from a local clone",
)
)
if not snapshot.path_available:
rows.append(
(
"warning",
"local-path-unavailable",
"Generation used indexed workplans; baseline needs a visible checkout",
)
)
if not rows:
rows.append(("none", "no-preflight-risks", "No blocker or warning reported"))
return "\n".join(
f"| {severity} | `{code}` | `{detail.replace('|', '\\|')}` |"
for severity, code, detail in sorted(rows)
)
def _active_work(snapshot: WorkplanGenerationSnapshot) -> str:
active = snapshot.preflight.get("active_work") or {}
rows: list[str] = []
for workplan in sorted(active.get("workplans") or [], key=lambda row: str(row.get("slug"))):
rows.append(
f"- workplan `{workplan.get('slug')}` / `{workplan.get('id')}` "
f"is `{workplan.get('status')}`"
)
for task in sorted(active.get("tasks") or [], key=lambda row: str(row.get("id"))):
rows.append(
f"- task `{task.get('record_id') or task.get('id')}` / `{task.get('id')}` "
f"is `{task.get('status')}`"
)
return "\n".join(rows) if rows else "- No active work was reported by preflight."
def _task(snapshot: WorkplanGenerationSnapshot, number: int, title: str, status: str) -> str:
task_id = f"{snapshot.workplan_id}-T{number:0{snapshot.task_width}d}"
return (
f"## {number}. {title}\n\n"
"```task\n"
f"id: {task_id}\n"
f"status: {status}\n"
"priority: high\n"
"```\n"
)
def render_workplan(snapshot: WorkplanGenerationSnapshot, *, owner: str) -> str:
"""Render the adoption plan. No current clock or random value is consulted."""
report = snapshot.preflight
repo = snapshot.repo
old_slug = snapshot.old_slug
new_slug = snapshot.new_slug
current = report.get("current") or {}
statehub = current.get("statehub") or {}
forge = current.get("forge") or {}
forge_identity = current.get("forge_identity") or {}
preflight_at = str(report.get("preflighted_at") or "")[:10]
created = preflight_at
source_commit = forge.get("head_commit") or "UNAVAILABLE"
forge_id = (
forge_identity.get("forge_repository_id")
or forge.get("repository_id")
or "UNVERIFIED"
)
aliases = sorted(
{
str(alias)
for alias in [*(repo.get("aliases") or []), old_slug]
if alias
}
)
safe = bool(report.get("safe_to_apply"))
status = "ready" if safe and source_commit != "UNAVAILABLE" else "proposed"
reviewed = (
f'reviewed_at: "{created}"\n'
"reviewed_by: statehub-repository-rename-generator\n"
f'reviewed_against_commit: "{source_commit}"\n'
if source_commit != "UNAVAILABLE"
else ""
)
confirmation = f"rename:{repo['id']}:{old_slug}:{new_slug}"
output = f'''---
id: {snapshot.workplan_id}
type: workplan
title: "Repository identity migration from {old_slug} to {new_slug}"
domain: {json.dumps(str(repo.get("domain_slug") or "unknown"))}
repo: {json.dumps(old_slug)}
status: {status}
owner: {json.dumps(owner)}
topic_slug: {json.dumps(snapshot.topic_slug)}
created: "{created}"
updated: "{created}"
{reviewed}quality_dor: DoR-Ok
quality_dor_at: "{created}"
quality_dor_by: statehub-repository-rename-generator
---
# {snapshot.workplan_id} — {old_slug} to {new_slug}
## Goal and authority boundary
Adopt the new repository coordinate `{new_slug}` while preserving State Hub
repository UUID `{repo['id']}`, Forge repository ID `{forge_id}`, source commit
`{source_commit}`, work-record identities, and telemetry relationships.
This plan changes repository coordinates only. Product/runtime names are a
separate explicit decision. No task in this file may close work owned by
another repository: it records the external handoff ID and waits for evidence
from that owning repository.
Live Forgejo rename, State Hub rebind, rollback, and old-checkout cleanup are
Red-lane actions. They require the exact confirmations below and recorded
human approval; redirects are compatibility evidence, not completion.
## Captured preflight
- schema: `{report.get('schema_version')}`
- report checksum: `{report.get('report_checksum') or 'UNAVAILABLE'}`
- safe to apply at capture: `{str(safe).lower()}`
- State Hub repository UUID: `{repo['id']}`
- Forge repository ID: `{forge_id}`
- default branch: `{forge.get('default_branch') or 'UNAVAILABLE'}`
- source commit: `{source_commit}`
- registered local path: `{statehub.get('local_path') or repo.get('local_path') or 'UNAVAILABLE'}`
- host paths: `{json.dumps(statehub.get('host_paths') or repo.get('host_paths') or {}, sort_keys=True)}`
- protected/current aliases: `{json.dumps(aliases)}`
- existing workplans observed: `{len(snapshot.observed_workplan_ids)}`
- archived workplans observed: `{len(snapshot.archived_workplan_ids)}`
The preflight token is deliberately not stored in this workplan. Generate a
fresh private mode-0600 preflight file immediately before execution.
## Preflight risk register
Every blocker, warning, and external handoff from the captured snapshot is
retained here. T05 must reconcile every row; unknown ownership is a blocker.
| Severity | Code | Captured detail |
| --- | --- | --- |
{_risk_register(snapshot)}
## Active-work coordination snapshot
{_active_work(snapshot)}
## External ownership ledger
| Effect | Owning repository | What this plan may claim |
| --- | --- | --- |
| State Hub identity and aliases | `state-hub` | Record operation phase/evidence only; State Hub owns the mutation. |
| Forge-derived fabric projection | `railiance-fabric` | Record a handoff ID; only that repository closes its source change. |
| Credential-route catalog | `ops-warden` | Record route-review handoff; never request or store a secret here. |
| SBOM projection | `sbom-nexus` | Record re-ingestion evidence; SBOM Nexus owns its projection. |
| CI, package, deployment, policy, and runtime consumers | `<owning-repository>` | T04 must replace this with one exact repository slug and handoff ID per consumer. |
'''
output += _task(snapshot, 1, "Capture immutable cleanliness and identity baseline", "todo")
output += f'''
Owner: `{old_slug}`.
- Require a clean target checkout and every branch/tag/change secured to Forgejo.
- Record `git status`, branch, remote, source commit, default branch, Forge ID,
visibility/readability, protected branches, releases, packages, hooks, deploy
keys, Actions variables, and redirects.
- Reconcile active work above: quiesce it or record an explicit concurrent-work
decision. A moved source commit requires a new preflight.
- If no registered path is visible, stop and register/verify a checkout; do not
infer private Forge state from local Git.
Gate: repository UUID, Forge ID, branch, commit, and clean-state evidence are
recorded and match a fresh State Hub preflight.
'''
output += _task(snapshot, 2, "Prepare repository metadata and work-record frontmatter", "todo")
output += f'''
Owner: `{old_slug}`.
- Prepare repository metadata, README/INTENT/SCOPE/AGENTS references, Forge
description/topics, and canonical clone coordinates for `{new_slug}`.
- Keep the established `{snapshot.prefix}-` workplan/task prefix and all existing
`state_hub_workstream_id` / `state_hub_task_id` values unchanged.
- Do not mass-rewrite historical prose or old-slug provenance. New live
frontmatter may adopt `repo: {new_slug}` only after State Hub rebind.
- Commit preparatory source changes before the final preflight; record the new
intended source commit.
Gate: work-record parsing succeeds and no existing UUID field was removed,
replaced, or invented.
'''
output += _task(snapshot, 3, "Decide product and runtime naming separately", "todo")
output += f'''
Owner: `{old_slug}`.
Record explicit keep/rename decisions for the binary, Go module/import path,
`FLEX_AUTH_*` environment variables, Kubernetes namespace, service/DNS names,
Helm release/chart, container/package coordinates, policy vocabulary, API
names, dashboards, and telemetry service labels. Repository rename approval
does not authorize any of these runtime/product changes.
Gate: every item has a decision record and independently deployable changes
have their own workplan or residual handoff.
'''
output += _task(snapshot, 4, "Inventory consumers and create owned handoffs", "todo")
output += f'''
Owner: `{old_slug}` for inventory and handoff creation only.
- Inventory CI includes/actions, package and image publishers, deployments,
GitOps/Helm/Kubernetes references, credential routes (using `warden route`),
authorization and NetKingdom policy consumers, fabric sources, SBOM scans,
docs, badges, webhooks, mirrors, caches, dashboards, alerts, and local clones.
- For each external effect, replace `<owning-repository>` in the ledger with the
exact repository slug and create a live residual/intake/workplan there.
- Record each external work-record ID here. Do not mark that external work done
from this repository; completion evidence must come from its owner.
- Include every row in the preflight risk register, even when it is only a
warning or currently zero-count projection.
Gate: every discovered external change has one named owning repository and
durable handoff ID; unknown ownership blocks the live rename.
'''
output += _task(snapshot, 5, "Renew State Hub preflight and record approval", "todo")
output += f'''
Owner: `{old_slug}`.
```text
statehub repo rename preflight {old_slug} {new_slug} \\
--operation-id <operation-id> \\
--output <private-preflight.json> --json
```
Verify zero blockers, immutable IDs/commit, active-work disposition, target
availability, every risk-register row, and no queued edge writes. Record a
human decision approving exact operation ID, source commit, target slug, owner,
window, rollback limits, and confirmation `{confirmation}`.
Gate: a current private preflight and explicit human approval exist. Never
commit the private preflight file or its token.
'''
output += _task(snapshot, 6, "Execute the Forgejo repository rename", "wait")
output += f'''
Owner: `{old_slug}`; human approval required.
Start/retry the durable journal, then apply exactly the Forge phase:
```text
statehub repo rename start {old_slug} {new_slug} \\
--operation-id <operation-id> --preflight-file <private-preflight.json> \\
--actor <actor> --confirm '{confirmation}' --json
statehub repo rename apply <operation-id> --phase forge-renamed \\
--confirm '{confirmation}' --json
```
Confirm Forgejo reports the same numeric repository ID and commit under
`{new_slug}`. Stop on identity drift, unreadability, target conflict, or moved
source; do not create a second State Hub repository.
Gate: operation journal is `forge-renamed` with immutable identity evidence.
'''
output += _task(snapshot, 7, "Record the State Hub identity rebind", "wait")
output += f'''
Owner: `{old_slug}` for coordination; `state-hub` owns the mutation.
```text
statehub repo rename apply <operation-id> --phase statehub-rebound \\
--confirm '{confirmation}' --json
statehub repo rename status <operation-id> --json
```
Record State Hub's phase evidence. Verify both `{old_slug}` and `{new_slug}`
resolve to repository UUID `{repo['id']}`, `{new_slug}` is canonical, and the old
slug is a protected alias. Do not manually update the database.
Gate: State Hub journal is `statehub-rebound`; IDs and historical relationships
are unchanged.
'''
output += _task(snapshot, 8, "Establish and register a fresh canonical clone", "wait")
output += f'''
Owner: `{new_slug}` after rebind.
- Preserve the old checkout until verification and rollback decisions finish.
- Clone `{new_slug}` into a new path; verify origin and Forge numeric ID before
trusting redirects.
- Register the fresh path against the existing State Hub UUID, update canonical
source metadata/frontmatter, commit the source-synchronization revision, and
run `statehub fix-consistency` from the new clone.
- Apply `source-synced` with mode-0600 evidence naming clone path, remote, head,
registration result, and consistency result.
Gate: no duplicate repository registration exists and the journal is
`source-synced`.
'''
output += _task(
snapshot,
9,
"Verify identity, history, routes, builds, and deployments",
"wait",
)
output += f'''
Owner: `{new_slug}` for aggregation; each external owner supplies its evidence.
Compare relationship checksums—not only counts—for repository UUID, workplans,
tasks, progress, decisions, token events/totals, SBOM snapshots, services,
capabilities, messages, aliases, bindings, and active dispatch. Verify old/new
routes, Forge ID/commit, clean builds/tests, package/image publication, policy
and credential routes, fabric/SBOM projections, deployments, health, dashboards,
alerts, and telemetry continuity.
```text
statehub repo rename verify <operation-id> --json
statehub repo rename apply <operation-id> --phase consumers-verified \\
--checks-file <private-checks.json> --evidence-file <private-evidence.json> \\
--confirm '{confirmation}' --json
statehub repo rename apply <operation-id> --phase completed \\
--confirm '{confirmation}' --json
```
Gate: every required local check passes and every external handoff has evidence
from its owning repository before completion.
'''
output += _task(snapshot, 10, "Exercise rollback decision points", "wait")
output += f'''
Owner: `{new_slug}` for the decision; State Hub and Forge owners execute their
own reversible phases.
At every phase decide continue, pause safely, or enter rollback preflight.
Rollback is allowed only if the old Forge slug remains available and named
external effects are reversible. It never deletes aliases or history.
```text
statehub repo rename rollback <operation-id> \\
--confirm 'rollback:<operation-id>' --json
# Execute only after reviewing safe_to_rollback and irreversible handoffs:
statehub repo rename rollback <operation-id> \\
--confirm 'rollback:<operation-id>' --execute --json
```
Gate: the forward completion or rolled-back terminal state is explicit; no
generic error is treated as proof of rollback.
'''
output += _task(snapshot, 11, "Soak, hand off residuals, and clean up the old checkout", "wait")
output += f'''
Owner: `{new_slug}`; destructive cleanup requires separate human approval.
- Define and observe a soak window covering deployments, policy decisions,
alerts, telemetry, packages, automation, and old-slug compatibility reads.
- Convert every unresolved item into a live residual (`origin: residual`,
`origin_ref: {snapshot.workplan_id}`) in its owning repository before finishing.
- Retain the protected `{old_slug}` alias. Alias retirement is out of scope.
- Only after terminal verification, soak, residual handoff, and explicit cleanup
approval may the old local checkout be removed. Record what was removed and
whether recovery remains possible from Forgejo.
Gate: no actionable residual exists only in prose, the new clone is canonical,
and old-checkout cleanup evidence is recorded.
'''
return output.rstrip() + "\n"
def default_output_path(snapshot: WorkplanGenerationSnapshot) -> Path:
if not snapshot.path_available or not snapshot.repo_path:
raise WorkplanGenerationError(
"No registered local checkout is visible; pass --output explicitly",
code="repository_path_unavailable",
)
return (
Path(snapshot.repo_path)
/ "workplans"
/ f"{snapshot.workplan_id}-repository-rename-to-{snapshot.new_slug}.md"
)
def write_workplan_exclusive(path: Path, content: str) -> None:
"""Create a workplan exactly once; overwriting is intentionally unsupported."""
path.parent.mkdir(parents=True, exist_ok=True)
try:
descriptor = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o644)
except FileExistsError as exc:
raise WorkplanGenerationError(
f"Refusing to overwrite existing workplan {path}",
code="output_exists",
) from exc
try:
with os.fdopen(descriptor, "w", encoding="utf-8") as handle:
handle.write(content)
except Exception:
path.unlink(missing_ok=True)
raise
def content_checksum(content: str) -> str:
return hashlib.sha256(content.encode("utf-8")).hexdigest()
__all__ = [
"WorkplanGenerationError",
"WorkplanGenerationSnapshot",
"build_generation_snapshot",
"content_checksum",
"default_output_path",
"render_workplan",
"write_workplan_exclusive",
]

83
tests/test_repository_rename_cli.py
View file

@ -388,3 +388,86 @@ def test_human_status_distinguishes_lifecycle_states(
_run(monkeypatch, "repo", "rename", "status", OPERATION_ID)
captured = capsys.readouterr()
assert f"{label}: repository rename status" in captured.out + captured.err
def test_generate_workplan_cli_writes_once_and_reports_contract(
monkeypatch, tmp_path, capsys
):
repo_root = tmp_path / "flex-auth"
repo_root.mkdir()
(repo_root / "AGENTS.md").write_text(
"**Workplan prefix:** `FLEX-WP-`\n"
)
output = repo_root / "workplans" / "rename.md"
report = {
"schema_version": "state-hub.repository-rename-preflight.v1",
"repo_id": REPO_ID,
"old_slug": "flex-auth",
"new_slug": "access-engine",
"safe_to_apply": True,
"blockers": [],
"warnings": [],
"report_checksum": "a" * 64,
"preflight_token": "private-token",
"preflighted_at": "2026-08-29T08:00:00Z",
"current": {
"statehub": {"local_path": str(repo_root), "host_paths": {}},
"forge_identity": {
"verification_state": "verified",
"forge_repository_id": 42,
},
"forge_available": True,
"forge": {
"repository_id": 42,
"default_branch": "main",
"head_commit": "b" * 40,
},
},
"active_work": {"workplans": [], "tasks": []},
"affected": {"external_handoffs": []},
}
repo = {**_repo(), "local_path": str(repo_root), "host_paths": {}}
workplans = [
{
"slug": "flex-wp-0018",
"backing_filename": "FLEX-WP-0018-last.md",
"status": "finished",
}
]
def request(_api_base, method, path, body=None, *, expected=dict):
del body
if path.startswith("/repos/flex-auth"):
return repo
if path.startswith("/workplans/"):
assert expected is list
return workplans
assert method == "POST"
return report
monkeypatch.setattr(rename_cli, "_api_request", request)
args = (
"repo",
"rename",
"generate-workplan",
"flex-auth",
"access-engine",
"--repo-path",
str(repo_root),
"--output",
str(output),
"--json",
)
_run(monkeypatch, *args)
result = json.loads(capsys.readouterr().out)
assert result["state"] == "achieved"
assert result["result"]["workplan_id"] == "FLEX-WP-0019"
assert result["result"]["workplan_prefix"] == "FLEX-WP"
assert result["result"]["state_hub_uuid_fields_written"] == []
assert "private-token" not in output.read_text()
with pytest.raises(SystemExit) as exc:
_run(monkeypatch, *args)
assert exc.value.code == 1
refused = json.loads(capsys.readouterr().out)
assert refused["error"]["code"] == "output_exists"

323
tests/test_repository_rename_workplan.py Normal file
View file

@ -0,0 +1,323 @@
from __future__ import annotations
import hashlib
from pathlib import Path
import pytest
from repository_rename_workplan import (
WorkplanGenerationError,
build_generation_snapshot,
content_checksum,
default_output_path,
render_workplan,
write_workplan_exclusive,
)
from scripts.quality_debt import collect as collect_quality_debt
from scripts.validate_repo_adr import validate
REPO_ID = "fda8ad85-a7d7-4055-8f21-902a533e59df"
def _repo(path: str | None = None, **extra) -> dict:
return {
"id": REPO_ID,
"slug": "flex-auth",
"canonical_slug": "flex-auth",
"domain_slug": "infotech",
"local_path": path,
"host_paths": {},
"aliases": [],
**extra,
}
def _preflight(**extra) -> dict:
report = {
"schema_version": "state-hub.repository-rename-preflight.v1",
"repo_id": REPO_ID,
"old_slug": "flex-auth",
"new_slug": "access-engine",
"safe_to_apply": True,
"blockers": [],
"warnings": [],
"report_checksum": "a" * 64,
"preflight_token": "must-never-appear-in-workplan",
"preflighted_at": "2026-08-29T08:00:00Z",
"current": {
"statehub": {
"repo_id": REPO_ID,
"local_path": None,
"host_paths": {},
},
"forge_identity": {
"verification_state": "verified",
"forge_repository_id": 42,
"forge_owner": "coulomb",
},
"forge_available": True,
"forge": {
"repository_id": 42,
"default_branch": "main",
"head_commit": "b" * 40,
},
},
"active_work": {"workplans": [], "tasks": []},
"affected": {"external_handoffs": []},
}
report.update(extra)
return report
def _workplans() -> list[dict]:
return [
{
"id": "11111111-1111-4111-8111-111111111111",
"slug": "flex-wp-0017",
"status": "active",
"backing_filename": "FLEX-WP-0017-action-contract.md",
},
{
"id": "22222222-2222-4222-8222-222222222222",
"slug": "flex-wp-0018",
"status": "archived",
"backing_filename": "260828-FLEX-WP-0018-inbound-corrections.md",
"backing_archived": True,
},
]
def _repo_root(tmp_path: Path) -> Path:
root = tmp_path / "flex-auth"
root.mkdir()
(root / "AGENTS.md").write_text("**Workplan prefix:** `FLEX-WP-`\n")
return root
def test_flex_auth_snapshot_is_deterministic_and_complete(tmp_path):
root = _repo_root(tmp_path)
report = _preflight(
warnings=[
{
"code": "active_work_present",
"message": "coordinate NetKingdom security-stack work",
"workplan_count": 1,
}
],
affected={
"external_handoffs": [
"fabric-graph-projections",
"interface-change-consumers",
]
},
active_work={
"workplans": [
{
"id": "33333333-3333-4333-8333-333333333333",
"slug": "FLEX-WP-0017",
"status": "active",
}
],
"tasks": [
{
"id": "44444444-4444-4444-8444-444444444444",
"record_id": "FLEX-WP-0017-T05",
"status": "progress",
}
],
},
)
snapshot = build_generation_snapshot(
repo=_repo(str(root), aliases=["flexauth-legacy"]),
preflight=report,
workplans=_workplans(),
)
first = render_workplan(snapshot, owner="codex")
second = render_workplan(snapshot, owner="codex")
assert first == second
assert content_checksum(first) == hashlib.sha256(first.encode()).hexdigest()
assert snapshot.prefix == "FLEX-WP"
assert snapshot.workplan_id == "FLEX-WP-0019"
assert first.count("```task\n") == 11
assert "id: FLEX-WP-0019-T06\nstatus: wait" in first
assert "id: FLEX-WP-0019-T11\nstatus: wait" in first
assert "state_hub_workstream_id:" not in first
assert "state_hub_task_id:" not in first
assert "must-never-appear-in-workplan" not in first
assert "`active_work_present`" in first
assert "`fabric-graph-projections`" in first
assert "`interface-change-consumers`" in first
assert "FLEX_AUTH_*" in first
assert "NetKingdom policy consumers" in first
assert "flexauth-legacy" in first
assert "| Credential-route catalog | `ops-warden` |" in first
assert "only that repository closes" in first
def test_generated_snapshot_passes_parser_and_quality_debt(tmp_path):
root = _repo_root(tmp_path)
snapshot = build_generation_snapshot(
repo=_repo(str(root)),
preflight=_preflight(),
workplans=_workplans(),
)
content = render_workplan(snapshot, owner="codex")
output = default_output_path(snapshot)
write_workplan_exclusive(output, content)
report = validate(root, skip_api=True)
assert report.failures == []
assert collect_quality_debt(
root, api_base=None, include_hub_intakes=False
) == []
assert output.name.startswith("FLEX-WP-0019-")
def test_snapshot_counts_archived_workplans_when_allocating_id(tmp_path):
root = _repo_root(tmp_path)
snapshot = build_generation_snapshot(
repo=_repo(str(root)),
preflight=_preflight(),
workplans=[
{
"slug": "flex-wp-0004",
"status": "active",
"backing_filename": "FLEX-WP-0004-live.md",
},
{
"slug": "flex-wp-0099",
"status": "archived",
"backing_filename": "260101-FLEX-WP-0099-old.md",
"backing_archived": True,
},
],
)
assert snapshot.workplan_id == "FLEX-WP-0100"
assert snapshot.archived_workplan_ids == ("FLEX-WP-0099",)
def test_snapshot_renders_active_work_and_private_forge_failure(tmp_path):
root = _repo_root(tmp_path)
report = _preflight(
safe_to_apply=False,
blockers=[
{
"code": "forge_unreadable",
"message": "private repository could not be inspected",
}
],
current={
"statehub": {"repo_id": REPO_ID, "host_paths": {}},
"forge_identity": {"verification_state": "unverified"},
"forge_available": False,
"forge": None,
},
active_work={
"workplans": [
{"id": "active-uuid", "slug": "FLEX-WP-0017", "status": "active"}
],
"tasks": [],
},
)
snapshot = build_generation_snapshot(
repo=_repo(str(root)), preflight=report, workplans=_workplans()
)
rendered = render_workplan(snapshot, owner="codex")
assert "status: proposed" in rendered
assert "`forge_unreadable`" in rendered
assert "`forge-identity-unverified`" in rendered
assert "`forge-repository-unreadable`" in rendered
assert "workplan `FLEX-WP-0017` / `active-uuid` is `active`" in rendered
assert "id: FLEX-WP-0019-T06\nstatus: wait" in rendered
def test_snapshot_supports_missing_local_path_from_indexed_workplans(tmp_path):
missing = tmp_path / "not-mounted"
snapshot = build_generation_snapshot(
repo=_repo(str(missing)),
preflight=_preflight(),
workplans=_workplans(),
)
rendered = render_workplan(snapshot, owner="codex")
assert snapshot.prefix == "FLEX-WP"
assert snapshot.path_available is False
assert "`local-path-unavailable`" in rendered
with pytest.raises(WorkplanGenerationError) as exc:
default_output_path(snapshot)
assert exc.value.code == "repository_path_unavailable"
def test_snapshot_preserves_old_alias_context(tmp_path):
root = _repo_root(tmp_path)
snapshot = build_generation_snapshot(
repo=_repo(str(root), aliases=["old-flex", "flex-auth"]),
preflight=_preflight(),
workplans=_workplans(),
)
rendered = render_workplan(snapshot, owner="codex")
assert 'protected/current aliases: `["flex-auth", "old-flex"]`' in rendered
assert "Retain the protected `flex-auth` alias" in rendered
def test_existing_uuid_fields_are_untouched_and_not_copied_to_new_plan(tmp_path):
root = _repo_root(tmp_path)
workplans = root / "workplans"
workplans.mkdir()
existing = workplans / "FLEX-WP-0001-existing.md"
original = """---
id: FLEX-WP-0001
type: workplan
title: Existing
domain: infotech
status: active
owner: codex
created: "2026-01-01"
state_hub_workstream_id: "55555555-5555-4555-8555-555555555555"
---
"""
existing.write_text(original)
snapshot = build_generation_snapshot(
repo=_repo(str(root)), preflight=_preflight(), workplans=[]
)
rendered = render_workplan(snapshot, owner="codex")
assert existing.read_text() == original
assert "55555555-5555-4555-8555-555555555555" not in rendered
assert "state_hub_workstream_id:" not in rendered
def test_exclusive_writer_refuses_existing_output(tmp_path):
output = tmp_path / "existing.md"
output.write_text("operator content\n")
with pytest.raises(WorkplanGenerationError) as exc:
write_workplan_exclusive(output, "replacement\n")
assert exc.value.code == "output_exists"
assert output.read_text() == "operator content\n"
def test_prefix_is_never_derived_from_new_slug(tmp_path):
root = tmp_path / "repo"
root.mkdir()
snapshot = build_generation_snapshot(
repo=_repo(str(root)),
preflight=_preflight(),
workplans=_workplans(),
)
assert snapshot.prefix == "FLEX-WP"
assert not snapshot.workplan_id.startswith("ACCESS-ENGINE")
def test_ambiguous_indexed_prefixes_fail_closed(tmp_path):
root = tmp_path / "repo"
root.mkdir()
with pytest.raises(WorkplanGenerationError) as exc:
build_generation_snapshot(
repo=_repo(str(root)),
preflight=_preflight(),
workplans=[
{"slug": "flex-wp-0001"},
{"slug": "other-wp-0001"},
],
)
assert exc.value.code == "workplan_prefix_ambiguous"

33
workplans/STATE-WP-0085-repository-lineage-preserving-rename.md
View file

@ -400,7 +400,7 @@ dashboard retains the pre-existing `/docs/intakes` broken-link warning.
```task
id: STATE-WP-0085-T06
status: todo
status: done
priority: high
state_hub_task_id: "154bdbb3-d3bf-5bac-85d7-eecd88545841"
```
@ -441,6 +441,37 @@ Acceptance:
- template snapshot tests cover repositories with active work, archived
workplans, missing local paths, private forge visibility, and old aliases.
Result (2026-08-29): added `statehub repo rename generate-workplan <old>
<new>` with a pure deterministic renderer over the registered repository,
captured rename preflight, indexed workplans, and visible repository metadata.
Prefix discovery uses the repository's declared convention when present and
otherwise requires one unambiguous prefix across active and archived workplan
records; it never derives a prefix from the proposed slug. Number allocation
includes archived records, so the fixed flex-auth snapshot selects
`FLEX-WP-0019` while retaining `FLEX-WP-` task/work-record identity.
The generated ready/proposed plan carries DoR evidence and eleven ordered
tasks for baseline capture, metadata preparation, separate product/runtime
naming decisions, consumer/CI/package/deployment/credential/fabric/SBOM/docs
inventory, renewed preflight and approval, Forgejo rename, State Hub rebind,
fresh-clone registration, relationship-aware verification, rollback, soak,
residuals, and old-checkout cleanup. Live rename and destructive cleanup tasks
start in `wait`. Every captured blocker, warning, active work item, alias, and
external handoff is rendered; an ownership ledger requires exact repository
slugs and prohibits the target plan from closing another repository's work.
Tokens are omitted. Existing UUID-bearing files remain untouched, and no new
`state_hub_workstream_id` or `state_hub_task_id` is invented.
Output uses exclusive create with no overwrite/force mode; private preflight
files retain the mode-0600 input guard. Missing checkouts can render from
indexed records only with an explicit output path and remain visibly blocked
for baseline capture. Snapshot/parser/quality tests cover deterministic
flex-auth output, active work, archived numbering, missing paths, unreadable
private Forge state, old aliases, ambiguous prefixes, UUID preservation, and
overwrite refusal. Verification: 801 Python tests and the 70-page dashboard
build pass; a clean wheel installation exposes the generator command. The
dashboard retains the pre-existing `/docs/intakes` broken-link warning.
## Prove failure recovery and telemetry continuity
```task