state-hub/api/services/forge_projection.py

"""Derive a repository's work-record projection from the forge (STATE-WP-0083-T01).

`ADR-012` decision 1 makes the forge the projection source. Central does its own
reading: it clones the repository's default branch and derives from that, rather
than accepting a projection computed elsewhere — which `ADR-010` decision 5
forbids.

This module is read-only. Deriving must be safe to run at any time, because it is
what makes the reset in `T03` verifiable: you can always ask what the projection
*should* be without changing anything.
"""

from __future__ import annotations

import os
import base64
import re
import subprocess
import tempfile
import uuid
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

import yaml

# Same derivation as ADR-007 / repo-manager, so a forge-derived projection and a
# preliminary overlay compute identical identities for the same record.
from api.services.task_record_id_backfill import qualify_task_id

_WORK_RECORD_NAMESPACE = uuid.UUID("a4058507-5c4a-5a00-ab06-fffa4fb46009")
_TASK_BLOCK_RE = re.compile(r"```task\s*\n(.*?)\n```", re.DOTALL)
_HEADING_RE = re.compile(r"^(#{1,4})\s+(.+?)$", re.MULTILINE)
DEFAULT_FORGE_BASE = "https://forgejo.coulomb.social/coulomb"


_IDENTIFIER_SLUG = re.compile(
    r"^[a-z0-9-]*-wp-\d{3,}(?:-t\d{2})?$"          # PREFIX-WP-0001, -T01
    r"|^[a-z0-9-]*adhoc[a-z0-9-]*?-\d{4}-\d{2}-\d{2}(?:-t\d{2})?$"  # daily ad-hoc, qualified either side
)


def _slug_is_identifier(slug: str) -> bool:
    """Whether a row's slug is a work-record identifier or a title slug.

    This is what separates a re-key from a rename when the path cannot. A row
    whose slug is an identifier is *claiming* to be that record, so if the forge
    derives a different identifier for the same file, they are two records and
    the old one retires. A row whose slug is a title (`three-phoenix-ha-cluster`
    for `RCLUSTER-WP-0007`) never claimed one: those are hub-first rows from
    before ADR-001, where the backing path is the only link there has ever been,
    and path matching is the only thing that can hold them together.
    """
    return bool(_IDENTIFIER_SLUG.match((slug or "").strip().lower()))


def _identity_is_derived(row: Any) -> bool:
    """Whether this row's UUID was derived from its identifier (ADR-007).

    Derived rows are UUIDv5 over the work-record namespace. Rows predating
    derived identity carry a random v4 UUID, and for those the identifier is a
    label rather than the identity — so the path fallback still applies, which
    is what keeps rename detection working for legacy records.
    """
    rid = getattr(row, "id", None)
    return getattr(rid, "version", None) == 5


def derived_record_uuid(record_id: str) -> str:
    return str(uuid.uuid5(_WORK_RECORD_NAMESPACE, f"helixforge\n{record_id}"))


class ForgeDeriveError(RuntimeError):
    """The repository could not be read from the forge."""


class ForgeUnreadableError(ForgeDeriveError):
    """Central is not permitted to read this repository — a policy, not a fault.

    `ADR-012`'s premise (the forge is the projection source) holds only for
    repositories central can read, and until `STATE-WP-0084` nothing said so:
    a private repository failed the same way a broken one did, so "cannot read"
    and "does not exist" were indistinguishable from the outside.

    They must never be confused, because they authorise opposite things. A
    repository that does not derive may have had its files removed deliberately;
    a repository we cannot read tells us nothing at all about its files. Only
    the first can justify retiring a record.

    Forgejo answers an unauthenticated request for a private repository with a
    404, so "not found" is classified as unreadable too. That is deliberate: the
    two cases are genuinely indistinguishable at this layer, and the safe
    reading of an ambiguous answer is the one that cannot destroy a record.
    """


# git says this in several ways depending on version, transport, and whether a
# credential helper is installed; all of them mean the same thing here.
_UNREADABLE_MARKERS = (
    "could not read username",
    "could not read password",
    "authentication failed",
    "terminal prompts disabled",
    "invalid username or password",
    "403 forbidden",
    "the requested url returned error: 403",
    "the requested url returned error: 401",
    "repository not found",
    "remote: not found",
    "does not appear to be a git repository",
)


def _is_unreadable(message: str) -> bool:
    low = message.lower()
    if "not found" in low and "fatal: repository" in low:
        return True
    return any(marker in low for marker in _UNREADABLE_MARKERS)


@dataclass
class DerivedTask:
    record_id: str
    uuid: str
    title: str | None
    status: str | None
    priority: str | None


@dataclass
class DerivedWorkplan:
    record_id: str
    uuid: str
    title: str | None
    status: str | None
    relative_path: str
    archived: bool
    tasks: list[DerivedTask] = field(default_factory=list)


@dataclass
class DerivedProjection:
    repo_slug: str
    commit: str
    workplans: list[DerivedWorkplan] = field(default_factory=list)
    # False when the checkout has no `workplans/` directory at all. An empty
    # projection then means "we did not find the records", which is not the same
    # claim as "this repository has no records" — and only the second one could
    # ever justify retiring anything.
    records_source_present: bool = True

    @property
    def task_count(self) -> int:
        return sum(len(w.tasks) for w in self.workplans)

    @property
    def retirement_eligible(self) -> bool:
        """Whether an absence in this projection is evidence of an absence.

        A projection that could not be read never reaches this: it raises. What
        this rules out is the quieter case — a clone that succeeded and returned
        nothing, which is what would have retired every record in a repository
        had a clone ever come back empty instead of failing (`STATE-WP-0084-T01`;
        the near-miss was `vergabe-teilnahme`).
        """
        return self.records_source_present

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema": "state-hub.forge-projection.v1",
            "repo_slug": self.repo_slug,
            # Provenance is not optional: a projection that cannot name the
            # commit it came from cannot be audited (ADR-012 decision 2).
            "commit": self.commit,
            "records_source_present": self.records_source_present,
            "retirement_eligible": self.retirement_eligible,
            "workplans": [
                {
                    "record_id": w.record_id,
                    "uuid": w.uuid,
                    "title": w.title,
                    "status": w.status,
                    "relative_path": w.relative_path,
                    "archived": w.archived,
                    "tasks": [
                        {
                            "record_id": t.record_id,
                            "uuid": t.uuid,
                            "title": t.title,
                            "status": t.status,
                            "priority": t.priority,
                        }
                        for t in w.tasks
                    ],
                }
                for w in self.workplans
            ],
        }


from api.services.forge_credential import forge_read_token  # noqa: F401

# Kept as module attributes so callers and tests that reached for them here
# still resolve after the sources moved to `forge_credential`.
FORGE_TOKEN_ENV = "FORGE_READ_TOKEN"
FORGE_TOKEN_FILE_ENV = "FORGE_READ_TOKEN_FILE"


def _credential_env(token: str | None) -> dict[str, str]:
    """Git config carrying the credential, passed by environment not argv.

    `-c http.extraHeader=...` would place the token in the process command line,
    where it is readable by anything that can run `ps` and lands in any log that
    records invocations. GIT_CONFIG_* achieves the same configuration without
    that exposure.
    """
    if not token:
        return {}
    header = base64.b64encode(f"x-access-token:{token}".encode()).decode()
    return {
        "GIT_CONFIG_COUNT": "1",
        "GIT_CONFIG_KEY_0": "http.extraHeader",
        "GIT_CONFIG_VALUE_0": f"Authorization: Basic {header}",
    }


def _run_git(
    *args: str, cwd: str | None = None, timeout: float = 120.0, token: str | None = None
) -> str:
    # Without this a clone of a private repository blocks on a username prompt
    # instead of failing, and an unattended derivation pass hangs rather than
    # reporting. Failing fast is what makes the unreadable case observable.
    env = {**os.environ, "GIT_TERMINAL_PROMPT": "0", "GIT_ASKPASS": "", "GCM_INTERACTIVE": "never"}
    env.update(_credential_env(token))
    proc = subprocess.run(
        ["git", *args], cwd=cwd, capture_output=True, text=True, timeout=timeout, env=env
    )
    if proc.returncode != 0:
        detail = (proc.stderr or proc.stdout).strip()[:400]
        if token:
            # Never let a credential reach an exception that is logged, stored
            # in a reset outcome, or returned over the API.
            detail = detail.replace(token, "***")
        raise ForgeDeriveError(detail)
    return proc.stdout.strip()


def _split_frontmatter(text: str) -> tuple[dict, str]:
    if not text.startswith("---"):
        return {}, text
    end = text.find("\n---", 3)
    if end == -1:
        return {}, text
    raw = text[3:end]
    body = text[end + 4 :]
    try:
        meta = yaml.safe_load(raw) or {}
    except yaml.YAMLError:
        return {}, text
    return (meta if isinstance(meta, dict) else {}), body


def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]:
    headings = [
        (m.start(), m.group(2).strip()) for m in _HEADING_RE.finditer(body)
    ]
    out: list[DerivedTask] = []
    for m in _TASK_BLOCK_RE.finditer(body):
        try:
            block = yaml.safe_load(m.group(1).strip()) or {}
        except yaml.YAMLError:
            continue
        if not isinstance(block, dict):
            continue
        rid = str(block.get("id") or "").strip()
        if not rid:
            continue
        title = block.get("title")
        if not title:
            prev = [t for pos, t in headings if pos < m.start()]
            title = prev[-1] if prev else None
        out.append(
            DerivedTask(
                # A bare `T01` is not an identifier: it is unique only within
                # its workplan, so `uuid5("T01")` is the same UUID for every
                # workplan in the fleet. llm-connect's 91 task blocks derive
                # just 49 distinct UUIDs unqualified, and creating its
                # workplans fails on a duplicate task primary key.
                #
                # Qualifying with the owning workplan is the same rule
                # `task_record_id_backfill.qualify_task_id` applies to stored
                # ids; both must agree or the backfill and the projection
                # disagree about what a task is called.
                record_id=qualify_task_id(rid, workplan_id) or rid,
                # Derived, not read from the file: the forge projection must not
                # inherit an identifier the file happens to carry.
                uuid=derived_record_uuid(qualify_task_id(rid, workplan_id) or rid),
                title=title,
                status=(str(block["status"]).strip() if block.get("status") else None),
                priority=(str(block["priority"]).strip() if block.get("priority") else None),
            )
        )
    return out


def derive_from_checkout(repo_root: Path, repo_slug: str, commit: str) -> DerivedProjection:
    """Derive a projection from an already-materialised checkout."""
    proj = DerivedProjection(repo_slug=repo_slug, commit=commit)
    wp_dir = repo_root / "workplans"
    if not wp_dir.is_dir():
        proj.records_source_present = False
        return proj
    for path in sorted(wp_dir.rglob("*.md")):
        if path.name.startswith("."):
            continue
        try:
            text = path.read_text(encoding="utf-8")
        except (OSError, UnicodeDecodeError):
            continue
        meta, body = _split_frontmatter(text)
        if str(meta.get("type") or "").strip() != "workplan":
            continue
        rid = str(meta.get("id") or "").strip()
        if not rid:
            continue
        proj.workplans.append(
            DerivedWorkplan(
                record_id=rid,
                uuid=derived_record_uuid(rid),
                title=(str(meta["title"]).strip() if meta.get("title") else None),
                status=(str(meta["status"]).strip() if meta.get("status") else None),
                relative_path=str(path.relative_to(repo_root).as_posix()),
                archived=path.parent.name == "archived",
                tasks=_parse_tasks(body, rid),
            )
        )
    proj.workplans.sort(key=lambda w: w.record_id)
    return proj


def derive_from_forge(
    repo_slug: str, *, forge_base: str = DEFAULT_FORGE_BASE, ref: str | None = None
) -> DerivedProjection:
    """Clone the repository's default branch from the forge and derive from it.

    A fresh shallow clone every time, deliberately: the source is what the forge
    holds now, and a reused working copy is how a projection ends up reflecting
    someone's local state instead (ADR-012 context).
    """
    url = f"{forge_base.rstrip('/')}/{repo_slug}.git"
    token = forge_read_token()
    with tempfile.TemporaryDirectory(prefix=f"forge-{repo_slug}-") as tmp:
        args = ["clone", "--depth", "1", "--quiet"]
        if ref:
            args += ["--branch", ref]
        try:
            _run_git(*args, url, tmp, token=token)
        except subprocess.TimeoutExpired as exc:
            raise ForgeDeriveError(f"clone timed out for {repo_slug}") from exc
        except ForgeDeriveError as exc:
            # Classify before propagating. A caller that cannot tell "not
            # permitted" from "broken" will eventually treat one as the other.
            if _is_unreadable(str(exc)):
                raise ForgeUnreadableError(
                    f"{repo_slug} could not be read from the forge: {exc}"
                ) from exc
            raise
        commit = _run_git("rev-parse", "HEAD", cwd=tmp)
        return derive_from_checkout(Path(tmp), repo_slug, commit)


# ---------------------------------------------------------------------------
# Comparison against what the hub currently holds (STATE-WP-0083-T02)
# ---------------------------------------------------------------------------



_ARCHIVE_PREFIX_RE = re.compile(r"^\d{6}-")


def _path_key(path: str) -> str:
    """Normalise a workplan path so an archived copy matches its live one."""
    name = path.rsplit("/", 1)[-1]
    return _ARCHIVE_PREFIX_RE.sub("", name).strip().lower()


@dataclass
class ProjectionDiff:
    """What a reset would change, computed without changing anything.

    This is what makes `ADR-012` decision 7's "verifiable" real: the reset can
    always be inspected before it runs, and its result compared against the forge
    afterwards.
    """

    repo_slug: str
    commit: str
    missing: list[dict[str, Any]] = field(default_factory=list)   # forge has, hub lacks
    stale: list[dict[str, Any]] = field(default_factory=list)     # hub has, forge lacks
    differing: list[dict[str, Any]] = field(default_factory=list) # both, fields differ
    # Set when the source could not support a claim of absence, so `stale` was
    # deliberately left empty rather than computed (`STATE-WP-0084-T01`).
    stale_withheld: str | None = None

    @property
    def clean(self) -> bool:
        return not (self.missing or self.stale or self.differing)

    @property
    def would_remove(self) -> int:
        return len(self.stale)

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema": "state-hub.projection-diff.v1",
            "repo_slug": self.repo_slug,
            "commit": self.commit,
            "clean": self.clean,
            "counts": {
                "missing": len(self.missing),
                "stale": len(self.stale),
                "differing": len(self.differing),
            },
            "stale_withheld": self.stale_withheld,
            "missing": self.missing,
            "stale": self.stale,
            "differing": self.differing,
        }


def diff_against_hub(
    derived: DerivedProjection,
    hub_workplans: list[dict[str, Any]],
    hub_tasks_by_workplan: dict[str, list[dict[str, Any]]],
) -> ProjectionDiff:
    """Compare a derived projection with the hub's current records.

    Pure: takes the hub's state as data rather than reading it, so the comparison
    is testable without a database and cannot accidentally mutate anything.
    """
    d = ProjectionDiff(repo_slug=derived.repo_slug, commit=derived.commit)
    if not derived.retirement_eligible:
        # Compute what is missing and what differs as usual — those only ever
        # add or correct. Absence is the one conclusion this source cannot
        # support, so it is not drawn at all rather than drawn and then filtered.
        d.stale_withheld = (
            "the checkout has no workplans/ directory, so an absent record is "
            "unexplained rather than evidence of removal"
        )

    # Match on canonical identity, never on UUID. Most hub records still carry
    # pre-ADR-007 random identifiers, so a UUID-keyed comparison reports every
    # record as simultaneously missing and stale — and a reset built on that
    # would destroy and recreate the entire projection. The canonical record id
    # is what is stable across the identifier migration; the backing file is the
    # fallback when a hub slug was derived from a filename rather than an id.
    def _wp_key(record_id: str | None, slug: str | None, path: str | None) -> str:
        if record_id:
            return record_id.strip().lower()
        if path:
            return _path_key(path)
        return (slug or "").strip().lower()

    want_wp = {w.record_id.strip().lower(): w for w in derived.workplans}
    have_wp: dict[str, dict[str, Any]] = {}
    want_paths = {_path_key(w.relative_path): k for k, w in want_wp.items()}
    for w in hub_workplans:
        slug = str(w.get("slug") or "")
        key = slug.strip().lower()
        if key not in want_wp:
            # Slugs were not always the canonical id; fall back to the file.
            bp = w.get("backing_relative_path")
            if bp and _path_key(bp) in want_paths:
                key = want_paths[_path_key(bp)]
            else:
                cand = [k for k in want_wp if slug.lower().startswith(k + "-")]
                if len(cand) == 1:
                    key = cand[0]
        have_wp[key] = w

    for key, w in want_wp.items():
        if key not in have_wp:
            d.missing.append({"kind": "workplan", "record_id": w.record_id, "uuid": w.uuid})
    for key, w in have_wp.items():
        if key not in want_wp:
            if d.stale_withheld:
                continue
            uid = str(w["id"])
            d.stale.append(
                {
                    "kind": "workplan",
                    "uuid": uid,
                    "slug": w.get("slug"),
                    "status": w.get("status"),
                    # A stale record with no backing file is the case that must
                    # never be destroyed silently (ADR-012 decision 7).
                    "has_backing_file": bool(
                        w.get("backing_filename") or w.get("backing_relative_path")
                    ),
                }
            )
    for key, w in want_wp.items():
        cur = have_wp.get(key)
        uid = str(cur["id"]) if cur else w.uuid
        if not cur:
            continue
        changed = {}
        if (cur.get("status") or None) != (w.status or None):
            changed["status"] = {"hub": cur.get("status"), "forge": w.status}
        cur_path = cur.get("backing_relative_path") or None
        if cur_path != w.relative_path:
            changed["backing_relative_path"] = {"hub": cur_path, "forge": w.relative_path}
        if changed:
            d.differing.append(
                {"kind": "workplan", "record_id": w.record_id, "uuid": uid, "changed": changed}
            )

    for w in derived.workplans:
        hub_rows = hub_tasks_by_workplan.get(w.uuid, [])
        if not hub_rows:
            cur = have_wp.get(w.record_id.strip().lower())
            if cur:
                hub_rows = hub_tasks_by_workplan.get(str(cur["id"]), [])
        # Match on the canonical record id where the hub has one. Rows created
        # before STATE-WP-0083-T06 fall back to title, which is why those are
        # reported rather than acted on: a renamed heading is indistinguishable
        # from a replaced task under title matching.
        def _task_key(record_id: str | None, title: str | None) -> str:
            if record_id:
                return record_id.strip().lower()
            return "title:" + (title or "").strip().lower()

        have = {
            _task_key(t.get("record_id"), t.get("title")): t for t in hub_rows
        }
        want = {_task_key(t.record_id, t.title): t for t in w.tasks}
        for key, t in want.items():
            if key not in have:
                d.missing.append(
                    {"kind": "task", "record_id": t.record_id, "uuid": t.uuid,
                     "workplan": w.record_id}
                )
        for key, t in have.items():
            if key not in want:
                if d.stale_withheld:
                    continue
                uid = str(t["id"])
                d.stale.append(
                    {"kind": "task", "uuid": uid, "title": t.get("title"),
                     "status": t.get("status"), "workplan": w.record_id,
                     "has_backing_file": False}
                )
        for key, t in want.items():
            cur = have.get(key)
            if cur and (cur.get("status") or None) != (t.status or None):
                d.differing.append(
                    {"kind": "task", "record_id": t.record_id, "uuid": t.uuid,
                     "workplan": w.record_id,
                     "changed": {"status": {"hub": cur.get("status"), "forge": t.status}}}
                )
    return d


# ---------------------------------------------------------------------------
# Applying the reset (STATE-WP-0083-T03)
# ---------------------------------------------------------------------------


@dataclass
class ResetOutcome:
    repo_slug: str
    commit: str
    status: str                      # applied | refused | noop | unreadable
    created: list[str] = field(default_factory=list)
    updated: list[str] = field(default_factory=list)
    retired: list[str] = field(default_factory=list)
    refused: list[dict[str, Any]] = field(default_factory=list)
    # Identifiers freed from rows retired before retirement released them.
    released: list[str] = field(default_factory=list)
    notes: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema": "state-hub.projection-reset.v1",
            "repo_slug": self.repo_slug,
            "commit": self.commit,
            "status": self.status,
            "counts": {
                "created": len(self.created),
                "updated": len(self.updated),
                "retired": len(self.retired),
                "refused": len(self.refused),
                "released": len(self.released),
            },
            "created": self.created,
            "updated": self.updated,
            "retired": self.retired,
            "released": self.released,
            "refused": self.refused,
            "notes": self.notes,
        }


RETIRED_SLUG_MARK = "@retired-"


def _tombstone_slug(slug: str, when: datetime) -> str:
    """Release the identifier a retired row was holding.

    `slug` is unique across the whole table, so retirement that only sets a
    timestamp leaves the identifier locked to a record nothing derives any
    more — and the repository that legitimately owns it can never claim it.
    That is what kept repo-seed refused after core-hub's inherited REPO-WP rows
    were retired.

    The row, its UUID, and its progress events are untouched; only the
    human-facing identifier is stamped, so history stays attached to the record
    it happened under. Re-retiring an already-stamped row must not stack marks,
    or the column overflows after a few passes.
    """
    base = (slug or "").split(RETIRED_SLUG_MARK)[0]
    stamped = f"{base}{RETIRED_SLUG_MARK}{when:%Y%m%d}"
    return stamped[:100]


RETIRE_REASON = "no longer derived from the forge"


async def reset_repository_projection(
    session: Any,
    repo_slug: str,
    *,
    acknowledge_retirements: bool = False,
    forge_base: str = DEFAULT_FORGE_BASE,
    derived: DerivedProjection | None = None,
) -> ResetOutcome:
    """Reconcile one repository's workplan projection against the forge.

    Creates what the forge has and the hub lacks, updates what differs, and
    retires what no longer derives. It does not delete: hub-native records
    reference workplans with `ON DELETE RESTRICT`, and destroying a progress
    event to tidy a derived projection would lose hub-native truth to fix a
    derived-state problem (`ADR-012` decision 7 as amended).

    Retirement is refused by default. A record that stops deriving may mean the
    file was removed deliberately — or that someone pointed this at the wrong
    branch. The caller must say which.

    Scope: workplans, and the tasks of workplans being created. Tasks of
    *existing* workplans are left alone, because hub task rows carry no
    canonical identifier and can only be matched by title — renaming a heading
    would otherwise destroy and recreate its record (`T06`).
    """
    from datetime import datetime, timezone

    from sqlalchemy import select

    from api.models.managed_repo import ManagedRepo
    from api.models.task import Task
    from api.models.workplan import Workplan

    if derived is None:
        try:
            derived = derive_from_forge(repo_slug, forge_base=forge_base)
        except ForgeUnreadableError as exc:
            # Not an error: a statement about what central is permitted to see.
            # Reported as its own status so a caller cannot mistake it for a
            # repository whose records stopped deriving (`STATE-WP-0084-T01`).
            out = ResetOutcome(repo_slug=repo_slug, commit="", status="unreadable")
            out.refused.append(
                {"reason": "repository could not be read from the forge", "slug": repo_slug,
                 "detail": str(exc)[:300]}
            )
            out.notes.append(
                "Nothing was changed and nothing was retired. This says nothing "
                "about whether the repository's records still exist."
            )
            return out
    outcome = ResetOutcome(repo_slug=repo_slug, commit=derived.commit, status="noop")

    repo = (
        await session.execute(select(ManagedRepo).where(ManagedRepo.slug == repo_slug))
    ).scalar_one_or_none()
    if repo is None:
        outcome.status = "refused"
        outcome.refused.append({"reason": "repository is not registered", "slug": repo_slug})
        return outcome

    rows = list(
        (
            await session.execute(select(Workplan).where(Workplan.repo_id == repo.id))
        ).scalars()
    )

    # Release identifiers held by rows retired before retirement freed them.
    # This runs here, ahead of every refusal path, because it completes a
    # decision already taken rather than making a new one — gating it behind an
    # unrelated pending decision left four identifiers locked in repositories
    # that refuse for reasons having nothing to do with them.
    for r in rows:
        if r.projection_retired_at is not None and RETIRED_SLUG_MARK not in (r.slug or ""):
            before = r.slug
            r.slug = _tombstone_slug(r.slug or str(r.id), r.projection_retired_at)
            outcome.released.append(before)

    want = {w.record_id.strip().lower(): w for w in derived.workplans}
    want_paths = {_path_key(w.relative_path): k for k, w in want.items()}

    # The derived UUID is the strongest key there is: if a row's id equals
    # uuid5 of a wanted record's identifier, that row *is* that record, whatever
    # its slug says. Checking it first is what stops a correct record with a
    # legacy title slug — `testdrive-jsui-publication` for MARKITECT-WP-0002 —
    # from being read as a re-key and proposed for retirement.
    want_by_uuid = {w.uuid: k for k, w in want.items()}

    matched: dict[str, Any] = {}
    for row in rows:
        key = (row.slug or "").strip().lower()
        by_uuid = want_by_uuid.get(str(row.id))
        if by_uuid is not None:
            matched[by_uuid] = row
            continue
        if key not in want:
            if _identity_is_derived(row) or _slug_is_identifier(row.slug or ""):
                # ADR-007: a record identified by an identifier *is* that
                # identifier. Two cases reach here — a derived (v5) row, and a
                # legacy row whose slug is still a work-record identifier. Both
                # are re-keys, not renames.
                #
                # ADR-007: a derived record's identity *is* a function of its
                # identifier, so a changed identifier means a different record —
                # the old one retires and the new one is created. Matching it to
                # the file by path instead would update the row in place while
                # its UUID still encodes the old identifier, leaving the file
                # and the hub disagreeing about what the record is called.
                #
                # This is not hypothetical: the ad-hoc requalification
                # (CUST-WP-0066) deliberately keeps the filename, so for those
                # records a re-key *never* changes the path. Path matching
                # cannot tell a re-key from a rename there, and silently chose
                # rename for all 30 of them.
                #
                # The row keeps its own slug as key, so it lands in `stale` and
                # becomes a retirement candidate rather than vanishing.
                matched[key] = row
                continue
            bp = row.backing_relative_path
            if bp and _path_key(bp) in want_paths:
                key = want_paths[_path_key(bp)]
            else:
                cand = [k for k in want if key.startswith(k + "-")]
                key = cand[0] if len(cand) == 1 else key
        matched[key] = row

    # An identifier this repository would create may already belong to another
    # repository. Two repositories creating the same daily identifier on the
    # same day is a documented case (CUST-WP-0066), and the derivation is
    # deliberately deterministic, so the collision is real rather than
    # incidental. Refuse and say so: a constraint violation is a stack trace,
    # a refusal is something the caller can rule on.
    creating = [w for k, w in want.items() if k not in matched]
    if creating:
        foreign = list(
            (
                await session.execute(
                    select(Workplan).where(
                        Workplan.id.in_([uuid.UUID(w.uuid) for w in creating]),
                        Workplan.repo_id != repo.id,
                    )
                )
            ).scalars()
        )
        # `slug` carries its own unique constraint across the whole table, so an
        # identifier check alone is not enough: two repositories can derive
        # different identifiers whose slugs still collide. Missing this is what
        # left disaster-control raising IntegrityError after the identifier
        # refusal was added.
        slug_clash = list(
            (
                await session.execute(
                    select(Workplan).where(
                        Workplan.slug.in_([w.record_id.lower() for w in creating]),
                        Workplan.repo_id != repo.id,
                    )
                )
            ).scalars()
        )
        if slug_clash:
            held = {(r.slug or "").lower(): r for r in slug_clash}
            outcome.status = "refused"
            for w in creating:
                row = held.get(w.record_id.lower())
                if row is None:
                    continue
                outcome.refused.append(
                    {
                        "reason": "slug already belongs to another repository",
                        "record_id": w.record_id,
                        "slug": w.record_id.lower(),
                        "held_by_id": str(row.id),
                    }
                )
            outcome.notes.append(
                "Identifier collision is an identity decision, not a projection "
                "one; acknowledging retirements does not authorise it."
            )
            return outcome

        if foreign:
            owned = {str(r.id): r for r in foreign}
            outcome.status = "refused"
            for w in creating:
                held = owned.get(w.uuid)
                if held is None:
                    continue
                outcome.refused.append(
                    {
                        "reason": "derived identifier already belongs to another repository",
                        "record_id": w.record_id,
                        "uuid": w.uuid,
                        "held_by_slug": held.slug,
                    }
                )
            outcome.notes.append(
                "Identifier collision is an identity decision, not a projection "
                "one; acknowledging retirements does not authorise it."
            )
            return outcome

    stale = [
        r for k, r in matched.items()
        if k not in want and r.projection_retired_at is None
    ]
    if stale and not derived.retirement_eligible:
        # Acknowledgement cannot authorise this. The caller is confirming that
        # records which stopped deriving should be retired; here nothing has
        # been shown to have stopped deriving, because the source produced no
        # records to compare against. Consenting to a conclusion is not the
        # same as the evidence for it existing.
        outcome.status = "refused"
        for r in stale:
            outcome.refused.append(
                {
                    "reason": "source produced no records; absence is unexplained",
                    "slug": r.slug,
                    "status": r.status,
                    "backing_relative_path": r.backing_relative_path,
                }
            )
        outcome.notes.append(
            "The checkout has no workplans/ directory. Retirement withheld "
            "regardless of acknowledgement (STATE-WP-0084-T01)."
        )
        return outcome
    if stale and not acknowledge_retirements:
        outcome.status = "refused"
        for r in stale:
            outcome.refused.append(
                {
                    "reason": "would be retired; the forge no longer derives it",
                    "slug": r.slug,
                    "status": r.status,
                    "backing_relative_path": r.backing_relative_path,
                }
            )
        outcome.notes.append(
            "Re-run with acknowledgement to retire these. Nothing was changed."
        )
        return outcome

    now = datetime.now(tz=timezone.utc)

    for key, w in want.items():
        row = matched.get(key)
        if row is None:
            row = Workplan(
                id=uuid.UUID(w.uuid),
                repo_id=repo.id,
                topic_id=repo.topic_id,
                slug=w.record_id.lower(),
                title=w.title or w.record_id,
                status=w.status or "proposed",
                backing_filename=w.relative_path.rsplit("/", 1)[-1],
                backing_relative_path=w.relative_path,
                backing_archived=w.archived,
                derived_from_commit=derived.commit,
            )
            session.add(row)
            await session.flush()
            for t in w.tasks:
                # Safe only because nothing exists to mis-match against: this
                # workplan is new to the hub.
                session.add(
                    Task(
                        id=uuid.UUID(t.uuid),
                        workplan_id=row.id,
                        record_id=t.record_id,
                        title=t.title or t.record_id,
                        status=t.status or "todo",
                        priority=t.priority or "medium",
                    )
                )
            outcome.created.append(w.record_id)
            continue

        changed = False
        if w.status and row.status != w.status:
            row.status = w.status
            changed = True
        if row.backing_relative_path != w.relative_path:
            row.backing_relative_path = w.relative_path
            row.backing_filename = w.relative_path.rsplit("/", 1)[-1]
            row.backing_archived = w.archived
            changed = True
        if row.projection_retired_at is not None:
            # It derives again; un-retire rather than leaving a contradiction.
            row.projection_retired_at = None
            row.projection_retired_reason = None
            changed = True
        if row.derived_from_commit != derived.commit:
            row.derived_from_commit = derived.commit
            changed = True
        if changed:
            outcome.updated.append(w.record_id)

    for r in stale:
        outcome.retired.append(r.slug or str(r.id))
        r.projection_retired_at = now
        r.projection_retired_reason = RETIRE_REASON
        r.slug = _tombstone_slug(r.slug or str(r.id), now)

    if outcome.created or outcome.updated or outcome.retired or outcome.released:
        outcome.status = "applied"
    outcome.notes.append(
        "Tasks of existing workplans were not touched; hub tasks carry no "
        "canonical identifier (STATE-WP-0083-T06)."
    )
    return outcome


# ---------------------------------------------------------------------------
# Fleet form (STATE-WP-0083-T04)
# ---------------------------------------------------------------------------


@dataclass
class FleetResetOutcome:
    results: dict[str, dict[str, Any]] = field(default_factory=dict)
    errors: dict[str, str] = field(default_factory=dict)
    # Kept apart from `errors` on purpose. Nine repositories sitting in an error
    # bucket read as nine broken repositories; they were nine we were not
    # allowed to read, which is a different thing to go and fix
    # (`STATE-WP-0083-T04`, 2026-08-26).
    unreadable: dict[str, str] = field(default_factory=dict)

    def to_dict(self) -> dict[str, Any]:
        by_status: dict[str, int] = {}
        for r in self.results.values():
            by_status[r["status"]] = by_status.get(r["status"], 0) + 1
        return {
            "schema": "state-hub.fleet-projection-reset.v1",
            "repositories": len(self.results) + len(self.errors) + len(self.unreadable),
            "by_status": by_status,
            "errored": len(self.errors),
            "unreadable_count": len(self.unreadable),
            "unreadable": self.unreadable,
            "totals": {
                k: sum(r["counts"][k] for r in self.results.values())
                for k in ("created", "updated", "retired", "released", "refused")
            },
            "results": self.results,
            "errors": self.errors,
        }


async def reset_fleet_projection(
    session_factory: Any,
    repo_slugs: list[str],
    *,
    acknowledge_retirements: bool = False,
    forge_base: str = DEFAULT_FORGE_BASE,
) -> FleetResetOutcome:
    """Reset every repository, one at a time, sharing the per-repository path.

    The fleet form is a loop over the repository form and nothing else
    (`ADR-012` decision 7). The rarely-run wide operation must be the frequently
    run narrow one, or the wide one is trusted on the strength of never having
    been exercised.

    A repository that refuses or errors is recorded and the pass continues.
    Aborting on the first refusal would mean one unresolved repository blocks
    reconstruction everywhere — which in practice means permanently.

    Each repository gets its own session, so one failure cannot roll back
    another's work or leave a poisoned transaction behind.
    """
    outcome = FleetResetOutcome()
    for slug in repo_slugs:
        try:
            async with session_factory() as session:
                result = await reset_repository_projection(
                    session,
                    slug,
                    acknowledge_retirements=acknowledge_retirements,
                    forge_base=forge_base,
                )
                # A released identifier is a repair that stands on its own, so
                # it must survive a refusal in the same repository — otherwise
                # moving the release ahead of the refusal returns achieves
                # nothing and the rollback quietly undoes it.
                if result.status == "applied" or result.released:
                    await session.commit()
                else:
                    await session.rollback()
                if result.status == "unreadable":
                    detail = next(
                        (r.get("detail", "") for r in result.refused), ""
                    )
                    outcome.unreadable[slug] = detail or "could not be read from the forge"
                else:
                    outcome.results[slug] = result.to_dict()
        except ForgeUnreadableError as exc:
            # Reachable when the caller supplied its own derivation path.
            outcome.unreadable[slug] = str(exc)[:300]
        except Exception as exc:  # noqa: BLE001 - one repo must not end the pass
            outcome.errors[slug] = f"{type(exc).__name__}: {exc}"[:300]
    return outcome