state-hub/repository_rename_workplan.py

"""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 and attach a
  `state-hub.repository-rename-handoff.v1` payload following
  `state-hub/docs/schemas/repository-rename-handoff-v1.schema.json`. It names
  the source repository, old/new slug, affected paths or graph IDs, required
  re-ingest and verification, owning workplan/task, and non-secret evidence.
- 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 containing
  `fresh_clone: true`, Forge numeric `forge_repository_id`, exact `head_commit`,
  clone path, registration result, and consistency result. Remote coordinates
  contain no embedded credential.

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 is `verified`
with 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 its repository-rename handoff payload and owner work-record ID.
- 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",
]