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 datetime import date
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
from api.work_record_flavor import (
    FLAVOR_PROMOTION_REASONS, WORK_RECORD_FLAVORS,
    normalize_flavor, normalize_promotion_reason,
)

_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,6})\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
    description: str | None = None
    needs_human: bool = False
    intervention_note: str | None = None
    blocking_reason: str | None = None
    flavor: str | None = None
    decision_id: str | None = None
    blocked_on: str | None = None
    wait_until: date | None = None
    # Task-block ``depends_on`` targets (workplan or task record ids), in
    # source order, deduplicated. Unqualified ``T##`` targets are qualified
    # with the owning workplan.
    depends_on: list[str] = field(default_factory=list)


@dataclass
class DerivedWorkplan:
    record_id: str
    uuid: str
    title: str | None
    status: str | None
    relative_path: str
    archived: bool
    owner: str | None = None
    description: str | None = None
    tasks: list[DerivedTask] = field(default_factory=list)
    flavor: str | None = None
    flavor_promotion_reason: str | None = None
    flavor_promoted_from: str | None = None
    historical_preservation: bool = False
    # Frontmatter ``depends_on`` / ``depends_on_workplans`` / ``depends_on_tasks``
    # split as C-20 splits them: task record ids vs workplan record ids.
    depends_on_workplans: list[str] = field(default_factory=list)
    depends_on_tasks: list[str] = 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,
                    "owner": w.owner,
                    "description": w.description,
                    "flavor": w.flavor,
                    "flavor_promotion_reason": w.flavor_promotion_reason,
                    "flavor_promoted_from": w.flavor_promoted_from,
                    "relative_path": w.relative_path,
                    "archived": w.archived,
                    "historical_preservation": w.historical_preservation,
                    "depends_on_workplans": list(w.depends_on_workplans),
                    "depends_on_tasks": list(w.depends_on_tasks),
                    "tasks": [
                        {
                            "record_id": t.record_id,
                            "uuid": t.uuid,
                            "title": t.title,
                            "status": t.status,
                            "priority": t.priority,
                            "description": t.description,
                            "needs_human": t.needs_human,
                            "intervention_note": t.intervention_note,
                            "blocking_reason": t.blocking_reason,
                            "flavor": t.flavor,
                            "decision_id": t.decision_id,
                            "blocked_on": t.blocked_on,
                            "wait_until": t.wait_until.isoformat() if t.wait_until else None,
                            "depends_on": list(t.depends_on),
                        }
                        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 _clean_task_description(raw: str) -> str | None:
    """Match the file-authoritative consistency parser's section boundaries."""
    lines = raw.splitlines()
    while lines and (not lines[0].strip() or lines[0].strip() in {"---", "***", "___"}):
        lines.pop(0)
    while lines and (not lines[-1].strip() or lines[-1].strip() in {"---", "***", "___"}):
        lines.pop()
    return "\n".join(lines).strip() or None



def _source_flavor(value: Any, field_name: str = "flavor") -> str | None:
    flavor = normalize_flavor(value)
    if flavor is not None and flavor not in WORK_RECORD_FLAVORS:
        raise ForgeDeriveError(f"Unknown source {field_name}; no projection was applied")
    return flavor


def _source_promotion_reason(value: Any) -> str | None:
    reason = normalize_promotion_reason(value)
    if reason is not None and reason not in FLAVOR_PROMOTION_REASONS:
        raise ForgeDeriveError("Unknown source flavor_promotion_reason; no projection was applied")
    return reason


def _source_wait_until(value: Any) -> date | None:
    if value is None or value == "":
        return None
    if type(value) is date:
        return value
    text = str(value).strip()
    if re.fullmatch(r"\d{4}-\d{2}-\d{2}", text):
        try:
            return date.fromisoformat(text)
        except ValueError:
            pass
    raise ForgeDeriveError("Source wait_until must be a valid YYYY-MM-DD date; no projection was applied")


def _source_wait_text(value: Any) -> str | None:
    if value is None:
        return None
    if not isinstance(value, str):
        raise ForgeDeriveError("Source wait qualifier must be text; no projection was applied")
    return value.strip() or None


def _source_needs_human(value: Any) -> bool:
    if isinstance(value, bool):
        return value
    if value is None:
        return False
    text = str(value).strip().lower()
    if text in {"true", "yes", "1", "on"}:
        return True
    if text in {"false", "no", "0", "off", ""}:
        return False
    raise ForgeDeriveError("Source needs_human must be boolean; no projection was applied")


_TASK_TARGET_RE = re.compile(r"-T\d{2,3}$")
_SHORT_TASK_TARGET_RE = re.compile(r"^T\d{2,3}$", re.IGNORECASE)


def _source_dependency_list(value: Any, field_name: str) -> list[str]:
    """Return declared dependency targets, refusing shapes C-20 cannot read.

    A mapping or nested list is not a dependency declaration; dropping it would
    silently lose an edge, so the whole derive is refused instead.
    """
    if value is None or value == "":
        return []
    if isinstance(value, str):
        items = value.split(",")
    elif isinstance(value, list):
        if any(isinstance(item, (dict, list)) for item in value):
            raise ForgeDeriveError(f"Malformed source {field_name}; no projection was applied")
        items = [str(item) for item in value if item is not None]
    elif isinstance(value, dict):
        raise ForgeDeriveError(f"Malformed source {field_name}; no projection was applied")
    else:
        items = [str(value)]
    out: list[str] = []
    for item in items:
        target = item.strip().strip('"').strip("'").strip()
        if target and target not in out:
            out.append(target)
    return out


def _split_dependency_targets(meta: dict, owner_record_id: str) -> tuple[list[str], list[str]]:
    """Split declared targets into (workplan ids, task ids) like C-20 does."""
    declared = _source_dependency_list(meta.get("depends_on"), "depends_on")
    workplans = [t for t in _source_dependency_list(meta.get("depends_on_workplans"), "depends_on_workplans")]
    tasks = _source_dependency_list(meta.get("depends_on_tasks"), "depends_on_tasks")
    for target in declared:
        if _SHORT_TASK_TARGET_RE.match(target):
            target = qualify_task_id(target, owner_record_id) or target
        (tasks if _TASK_TARGET_RE.search(target) else workplans).append(target)
    tasks = [qualify_task_id(t, owner_record_id) or t if _SHORT_TASK_TARGET_RE.match(t) else t for t in tasks]
    dedupe = lambda items: list(dict.fromkeys(items))  # noqa: E731
    return dedupe(workplans), dedupe(tasks)


def _parse_tasks(body: str, workplan_id: str) -> list[DerivedTask]:
    headings = [
        (m.start(), len(m.group(1)), m.group(2).strip()) for m in _HEADING_RE.finditer(body)
    ]
    out: list[DerivedTask] = []
    task_matches = list(_TASK_BLOCK_RE.finditer(body))
    for index, m in enumerate(task_matches):
        try:
            block = yaml.safe_load(m.group(1).strip()) or {}
        except (yaml.YAMLError, ValueError) as exc:
            # PyYAML raises ValueError for impossible bare ISO dates.
            # Refuse the entire derive instead of dropping a stored gate.
            raise ForgeDeriveError("Malformed source task metadata; no projection was applied") from exc
        if not isinstance(block, dict):
            continue
        rid = str(block.get("id") or "").strip()
        if not rid:
            continue
        prev = [(pos, level, text) for pos, level, text in headings if pos < m.start()]
        heading = prev[-1] if prev else None
        title = block.get("title") or (heading[2] if heading else None)
        # Nested evidence belongs to this task. Stop at the next peer/ancestor
        # heading or task block, as the ordinary consistency path does.
        description_end = task_matches[index + 1].start() if index + 1 < len(task_matches) else len(body)
        following_heading = next((
            pos for pos, level, _text in headings
            if pos > m.end() and (heading is None or level <= heading[1])
        ), None)
        if following_heading is not None:
            description_end = min(description_end, following_heading)
        description = _clean_task_description(body[m.end():description_end])
        if description is None and block.get("description") is not None:
            description = str(block["description"])
        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),
                description=description,
                needs_human=_source_needs_human(block.get("needs_human", False)),
                intervention_note=(
                    str(block["intervention_note"])
                    if block.get("intervention_note") is not None
                    else (str(block["blocking_reason"]) if _source_needs_human(block.get("needs_human")) and block.get("blocking_reason") is not None else None)
                ),
                blocking_reason=(
                    str(block["blocking_reason"])
                    if block.get("blocking_reason") is not None
                    else None
                ),
                flavor=_source_flavor(block.get("flavor")),
                decision_id=_source_wait_text(block.get("decision_id")),
                blocked_on=_source_wait_text(block.get("blocked_on")),
                wait_until=_source_wait_until(block.get("wait_until")),
                depends_on=_task_dependency_targets(block, workplan_id),
            )
        )
    return out


def _task_dependency_targets(block: dict, workplan_id: str) -> list[str]:
    workplans, tasks = _split_dependency_targets(block, workplan_id)
    return list(dict.fromkeys(workplans + tasks))


def _workplan_description(body: str) -> str | None:
    """Return bounded prose under ``## Goal`` when one is present."""
    match = re.search(r"^##\s+Goal\s*$", body, re.MULTILINE | re.IGNORECASE)
    if match is None:
        return None
    remainder = body[match.end() :]
    next_heading = re.search(r"^##\s+", remainder, re.MULTILINE)
    if next_heading is not None:
        remainder = remainder[: next_heading.start()]
    value = remainder.strip()
    return value[:4000] or None


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
    # Exact immutable histories are preserved by existing UUID, never allocated.
    from api.services.historical_projection import bound_sources
    try:
        historical = bound_sources(repo_root, repo_slug)
    except (ValueError, OSError, UnicodeError) as exc:
        raise ForgeDeriveError(str(exc)) from exc
    consumed = {entry["path"] for entry, _text in historical}
    groups = {}
    for entry, text in historical:
        meta, body = _split_frontmatter(text)
        parent_id = entry["parent_record_id"]
        # Split views carry the parent's existing UUID and repeat its task list.
        # Only the reviewed archived parent supplies the projected row.
        if entry["record_id"] != parent_id:
            continue
        if repo_slug == "activity-core":
            tasks = [DerivedTask(record_id=item["id"], uuid=item["state_hub_task_id"],
                                 title=item.get("title"), status=item.get("status"),
                                 priority=item.get("priority")) for item in meta["tasks"]]
        else:
            tasks = _parse_tasks(body, parent_id)
            bindings = [yaml.safe_load(m.group(1)) for m in _TASK_BLOCK_RE.finditer(body)]
            for task, binding in zip(tasks, bindings, strict=True):
                task.uuid = binding["state_hub_task_id"]
        groups[parent_id] = DerivedWorkplan(
            record_id=parent_id, uuid=entry["uuid"], title=meta.get("title"),
            status=meta.get("status"), relative_path=entry["path"],
            archived="/archived/" in entry["path"], tasks=tasks,
            historical_preservation=True,
        )
    proj.workplans.extend(groups.values())
    for path in sorted(wp_dir.rglob("*.md")):
        if path.relative_to(repo_root).as_posix() in consumed:
            continue
        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
        frontmatter_deps = _split_dependency_targets(meta, rid)
        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",
                owner=(str(meta["owner"]).strip() if meta.get("owner") else None),
                description=(
                    str(meta["description"]).strip()
                    if meta.get("description")
                    else _workplan_description(body)
                ),
                tasks=_parse_tasks(body, rid),
                flavor=_source_flavor(meta.get("flavor")),
                flavor_promotion_reason=_source_promotion_reason(meta.get("flavor_promotion_reason")),
                flavor_promoted_from=_source_flavor(meta.get("flavor_promoted_from"), "flavor_promoted_from"),
                depends_on_workplans=frontmatter_deps[0],
                depends_on_tasks=frontmatter_deps[1],
            )
        )
    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
    # More than one row/file claims one canonical identity. Choosing a winner
    # would turn a diagnostic into an implicit identity decision.
    ambiguous: list[dict[str, Any]] = field(default_factory=list)
    # 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
    # Source-owned dependency-edge preview (STATE-WP-0094-T05); None when the
    # caller supplied no hub edge state. Keys: create, remove, retain,
    # unresolved, external.
    edges: dict[str, list[dict[str, Any]]] | None = None

    @property
    def clean(self) -> bool:
        edge_changes = bool(self.edges and (self.edges["create"] or self.edges["remove"]))
        return not (self.missing or self.stale or self.differing or self.ambiguous or edge_changes)

    @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),
                "ambiguous": len(self.ambiguous),
            },
            "stale_withheld": self.stale_withheld,
            "missing": self.missing,
            "stale": self.stale,
            "differing": self.differing,
            "ambiguous": self.ambiguous,
            **({} if self.edges is None else {
                "edge_counts": {k: len(v) for k, v in self.edges.items()},
                "edges": self.edges,
            }),
        }


def _derived_identity_ambiguities(derived: DerivedProjection) -> list[dict[str, Any]]:
    """Return duplicate canonical identities declared by one forge projection."""
    workplans: dict[str, list[DerivedWorkplan]] = {}
    tasks: dict[str, list[tuple[DerivedWorkplan, DerivedTask]]] = {}
    for workplan in derived.workplans:
        workplans.setdefault(workplan.record_id.strip().lower(), []).append(workplan)
        for task in workplan.tasks:
            tasks.setdefault(task.record_id.strip().lower(), []).append((workplan, task))

    out: list[dict[str, Any]] = []
    for rows in workplans.values():
        if len(rows) > 1:
            out.append(
                {
                    "kind": "workplan",
                    "record_id": rows[0].record_id,
                    "source": "forge",
                    "paths": [row.relative_path for row in rows],
                }
            )
    for rows in tasks.values():
        if len(rows) > 1:
            out.append(
                {
                    "kind": "task",
                    "record_id": rows[0][1].record_id,
                    "source": "forge",
                    "workplans": [row[0].record_id for row in rows],
                }
            )
    return out


def diff_against_hub(
    derived: DerivedProjection,
    hub_workplans: list[dict[str, Any]],
    hub_tasks_by_workplan: dict[str, list[dict[str, Any]]],
    *,
    hub_edges: list[dict[str, Any]] | None = None,
    edge_targets: dict[str, list[str]] | None = None,
) -> 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.

    Dependency edges are previewed only when ``hub_edges`` is given: the
    ``workplan_dependencies`` rows whose from-workplan belongs to this
    repository (with ``source_owner``). ``edge_targets`` maps a target record id
    to its fleet-wide, non-retired candidate ids; a target absent from it but
    declared in this projection resolves to the id the receipt would create.
    """
    d = ProjectionDiff(repo_slug=derived.repo_slug, commit=derived.commit)
    d.ambiguous.extend(_derived_identity_ambiguities(derived))
    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()}
    preserved_by_uuid = {w.uuid: key for key, w in want_wp.items() if w.historical_preservation}
    for w in hub_workplans:
        if str(w["id"]) in preserved_by_uuid:
            have_wp[preserved_by_uuid[str(w["id"])]] = w
            continue
        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
        if w.historical_preservation:
            if cur.get("status") not in {"done", "finished", "archived", "completed"}:
                d.differing.append({"kind": "workplan", "record_id": w.record_id, "uuid": uid,
                                    "changed": {"historical_terminal_status": cur.get("status")}})
            continue
        changed = {}
        if (cur.get("status") or None) != (w.status or None):
            changed["status"] = {"hub": cur.get("status"), "forge": w.status}
        for field_name in ("flavor", "flavor_promotion_reason", "flavor_promoted_from"):
            expected = getattr(w, field_name)
            if cur.get(field_name) != expected:
                changed[field_name] = {"hub": cur.get(field_name), "forge": expected}
        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"]), [])
        if w.historical_preservation:
            by_uuid = {str(task["id"]): task for task in hub_rows}
            for task in w.tasks:
                current = by_uuid.get(task.uuid)
                if current is None:
                    d.missing.append({"kind": "task", "record_id": task.record_id,
                                      "uuid": task.uuid, "workplan": w.record_id})
                elif current.get("status") not in {"done", "cancel"}:
                    d.differing.append({"kind": "task", "record_id": task.record_id,
                                        "uuid": task.uuid, "changed": {
                                            "historical_terminal_status": current.get("status")}})
            continue
        # 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_groups: dict[str, list[dict[str, Any]]] = {}
        for task in hub_rows:
            have_groups.setdefault(
                _task_key(task.get("record_id"), task.get("title")), []
            ).append(task)
        for key, rows in have_groups.items():
            if len(rows) > 1:
                d.ambiguous.append(
                    {
                        "kind": "task",
                        "record_id": rows[0].get("record_id"),
                        "source": "hub",
                        "workplan": w.record_id,
                        "uuids": [str(row["id"]) for row in rows],
                        "match_key": key,
                    }
                )
        # Keep one representative so the rest of the diff remains useful, but
        # `ambiguous` makes the result non-clean and forbids treating it as an
        # actionable reset plan.
        have = {key: rows[0] for key, rows in have_groups.items()}
        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 not cur:
                continue
            changed = {}
            if (cur.get("status") or None) != (t.status or None):
                changed["status"] = {"hub": cur.get("status"), "forge": t.status}
            for field_name in ("needs_human", "intervention_note", "blocking_reason",
                               "flavor", "decision_id", "blocked_on", "wait_until"):
                expected = getattr(t, field_name)
                actual = cur.get(field_name, False if field_name == "needs_human" else None)
                if isinstance(actual, date):
                    actual = actual.isoformat()
                if isinstance(expected, date):
                    expected = expected.isoformat()
                if actual != expected:
                    changed[field_name] = {"hub": actual, "forge": expected}
            if changed:
                d.differing.append(
                    {"kind": "task", "record_id": t.record_id, "uuid": t.uuid,
                     "workplan": w.record_id, "changed": changed}
                )
    if hub_edges is not None:
        d.edges = _preview_source_edges(
            derived, want_wp, have_wp, hub_tasks_by_workplan, hub_edges, edge_targets or {}
        )
    return d


def _preview_source_edges(
    derived: DerivedProjection,
    want_wp: dict[str, DerivedWorkplan],
    have_wp: dict[str, dict[str, Any]],
    hub_tasks_by_workplan: dict[str, list[dict[str, Any]]],
    hub_edges: list[dict[str, Any]],
    edge_targets: dict[str, list[str]],
) -> dict[str, list[dict[str, Any]]]:
    """What receiving would do to source-owned edges, using the same planner."""
    from_ids: dict[str, str] = {}
    task_ids: dict[str, dict[str, str]] = {}
    pending: dict[str, str] = {}  # record id -> id the receipt would create
    for key, w in want_wp.items():
        cur = have_wp.get(key)
        wp_id = str(cur["id"]) if cur else w.uuid
        from_ids[key] = wp_id
        if not cur:
            pending[w.record_id] = wp_id
        hub_rows = hub_tasks_by_workplan.get(wp_id) or hub_tasks_by_workplan.get(w.uuid) or []
        by_record = {(r.get("record_id") or "").strip().lower(): str(r["id"])
                     for r in hub_rows if r.get("record_id")}
        for t in w.tasks:
            k = t.record_id.strip().lower()
            if k not in by_record:
                by_record[k] = t.uuid
                pending[t.record_id] = t.uuid
        task_ids[wp_id] = by_record
    targets: dict[str, tuple[str, Any]] = {}
    for target in source_edge_targets(want_wp):
        cands = sorted({str(c) for c in edge_targets.get(target, [])})
        if not cands and target in pending:
            cands = [pending[target]]
        if not cands:
            targets[target] = ("missing", None)
        elif len(cands) > 1:
            targets[target] = ("ambiguous", cands)
        else:
            targets[target] = ("ok", cands[0])
    desired, unresolved, held_sides = plan_source_edges(want_wp, from_ids, task_ids, targets)

    def _s(v: Any) -> str | None:
        return str(v) if v else None

    existing = [
        ((str(e["from_workplan_id"]), _s(e.get("from_task_id")), _s(e.get("to_workplan_id")),
          _s(e.get("to_task_id")), e.get("relationship_type") or "blocks"),
         str(e.get("id")), e.get("source_owner"))
        for e in hub_edges
    ]
    plan = classify_source_edges(desired, held_sides, existing, edge_source_owner(derived.repo_slug))
    return {
        "create": [_edge_dict(k, i) for k, i in plan["create"]],
        "remove": [_edge_dict(k, {"id": rid}) for k, rid in plan["remove"]],
        "retain": [_edge_dict(k, {"id": rid}) for k, rid in plan["retain"]],
        "unresolved": unresolved,
        "external": [_edge_dict(k, i) for k, i in plan["external"]],
    }


# ---------------------------------------------------------------------------
# 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)
    created_tasks: list[str] = field(default_factory=list)
    updated_tasks: list[str] = field(default_factory=list)
    cancelled_tasks: list[str] = field(default_factory=list)
    # Source-owned dependency edges (STATE-WP-0094-T05).
    created_edges: list[dict[str, Any]] = field(default_factory=list)
    removed_edges: list[dict[str, Any]] = field(default_factory=list)
    # Declared edges whose target is missing, ambiguous or invalid: recorded,
    # never silently dropped; the from-side's owned edges are retained.
    unresolved_edges: list[dict[str, Any]] = field(default_factory=list)
    # Declared edges already present as externally authored rows: preserved
    # and not adopted, so receiving can never remove them.
    external_edges: list[dict[str, Any]] = 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_tasks": len(self.created_tasks),
                "updated_tasks": len(self.updated_tasks),
                "cancelled_tasks": len(self.cancelled_tasks),
                "created_edges": len(self.created_edges),
                "removed_edges": len(self.removed_edges),
                "unresolved_edges": len(self.unresolved_edges),
                "external_edges": len(self.external_edges),
            },
            "created": self.created,
            "updated": self.updated,
            "retired": self.retired,
            "released": self.released,
            "created_tasks": self.created_tasks,
            "updated_tasks": self.updated_tasks,
            "cancelled_tasks": self.cancelled_tasks,
            "created_edges": self.created_edges,
            "removed_edges": self.removed_edges,
            "unresolved_edges": self.unresolved_edges,
            "external_edges": self.external_edges,
            "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"


def _uuid_key(value: Any) -> str:
    return str(uuid.UUID(str(value)))


async def _exact_retirement_blockers(session: Any, stale: list[Any]) -> list[dict[str, Any]]:
    """Block an exact retirement when any stale row still has tasks or edges.

    Zero means zero. Terminal tasks count, and so does a dependency edge in
    either direction. The caller retires nothing unless this returns empty.
    """
    if not stale:
        return []

    from sqlalchemy import or_, select

    from api.models.task import Task
    from api.models.workplan_dependency import WorkplanDependency

    stale_ids = [row.id for row in stale]
    tasks = list(
        (
            await session.execute(select(Task).where(Task.workplan_id.in_(stale_ids)))
        ).scalars()
    )
    edges = list(
        (
            await session.execute(
                select(WorkplanDependency).where(
                    or_(
                        WorkplanDependency.from_workplan_id.in_(stale_ids),
                        WorkplanDependency.to_workplan_id.in_(stale_ids),
                    )
                )
            )
        ).scalars()
    )
    refused: list[dict[str, Any]] = []
    for row in stale:
        key = str(row.id)
        row_tasks = [task for task in tasks if str(task.workplan_id) == key]
        row_edges = [
            edge
            for edge in edges
            if str(edge.from_workplan_id) == key
            or (edge.to_workplan_id is not None and str(edge.to_workplan_id) == key)
        ]
        if row_tasks:
            refused.append(
                {
                    "reason": "stale projection has tasks",
                    "uuid": key,
                    "slug": row.slug,
                    "task_ids": [str(task.id) for task in row_tasks],
                }
            )
        if row_edges:
            refused.append(
                {
                    "reason": "stale projection has dependency edges",
                    "uuid": key,
                    "slug": row.slug,
                    "dependency_ids": [str(edge.id) for edge in row_edges],
                }
            )
    if (tasks or edges) and not refused:
        refused.append(
            {
                "reason": "stale projection has tasks or dependency edges",
                "task_ids": [str(task.id) for task in tasks],
                "dependency_ids": [str(edge.id) for edge in edges],
            }
        )
    return refused


def _coerce_task_status(raw: str | None) -> Any:
    from api.models.task import TaskStatus

    if not raw:
        return None
    try:
        return TaskStatus(str(raw).strip().lower())
    except ValueError:
        return None


def _task_status_value(status: Any) -> str:
    return str(getattr(status, "value", status) or "")


def _sync_existing_workplan_tasks(
    session: Any,
    row: Any,
    derived_wp: DerivedWorkplan,
    hub_tasks: list[Any],
    outcome: ResetOutcome,
) -> None:
    """Match identified tasks of an existing workplan by record_id.

    Rows with no ``record_id`` predate the backfill and cannot be qualified
    from the file alone (CUST-WP-0068-T09). They are not cancelled or
    overwritten.
    """
    from api.models.task import Task, TaskPriority, TaskStatus

    want = {
        t.record_id.strip().lower(): t
        for t in derived_wp.tasks
        if t.record_id and t.record_id.strip()
    }
    matched: dict[str, Any] = {}
    claimed: set[str] = set()
    stale: list[Any] = []
    by_uuid = {str(ht.id): ht for ht in hub_tasks}
    for ht in hub_tasks:
        rid = (ht.record_id or "").strip().lower()
        if rid and rid in want:
            matched[rid] = ht
            claimed.add(str(ht.id))
        elif rid:
            stale.append(ht)

    for key, dt in want.items():
        ht = matched.get(key)
        if ht is None and dt.uuid in by_uuid:
            # Already on the hub under the derived UUID, but record_id was never
            # written (the registrar-minted case). Overwriting would collide.
            ht = by_uuid[dt.uuid]
            matched[key] = ht
            claimed.add(str(ht.id))
        if ht is None:
            kwargs: dict[str, Any] = {
                "id": uuid.UUID(dt.uuid),
                "workplan_id": row.id,
                "record_id": dt.record_id,
                "title": (dt.title or dt.record_id),
                "description": dt.description,
                "needs_human": dt.needs_human,
                "intervention_note": dt.intervention_note,
                "blocking_reason": dt.blocking_reason,
                "flavor": dt.flavor,
                "decision_id": dt.decision_id,
                "blocked_on": dt.blocked_on,
                "wait_until": dt.wait_until,
            }
            st = _coerce_task_status(dt.status)
            if st is not None:
                kwargs["status"] = st
            if dt.priority:
                try:
                    kwargs["priority"] = TaskPriority(dt.priority.strip().lower())
                except ValueError:
                    pass
            session.add(Task(**kwargs))
            outcome.created_tasks.append(dt.record_id)
            continue
        changed = False
        if not (ht.record_id or "").strip():
            ht.record_id = dt.record_id
            changed = True
        if dt.title and dt.title.strip() and ht.title != dt.title.strip():
            ht.title = dt.title.strip()
            changed = True
        if getattr(ht, "description", None) != dt.description:
            ht.description = dt.description
            changed = True
        st = _coerce_task_status(dt.status)
        if st is not None and ht.status != st:
            ht.status = st
            changed = True
        if getattr(ht, "needs_human", False) != dt.needs_human:
            ht.needs_human = dt.needs_human
            changed = True
        if getattr(ht, "intervention_note", None) != dt.intervention_note:
            ht.intervention_note = dt.intervention_note
            changed = True
        if getattr(ht, "blocking_reason", None) != dt.blocking_reason:
            ht.blocking_reason = dt.blocking_reason
            changed = True
        for field_name in ("flavor", "decision_id", "blocked_on", "wait_until"):
            expected = getattr(dt, field_name)
            if getattr(ht, field_name, None) != expected:
                setattr(ht, field_name, expected)
                changed = True
        if dt.priority:
            try:
                task_priority = TaskPriority(dt.priority.strip().lower())
            except ValueError:
                task_priority = None
            if task_priority is not None and getattr(ht, "priority", None) != task_priority:
                ht.priority = task_priority
                changed = True
        if changed:
            outcome.updated_tasks.append(dt.record_id)

    for ht in stale:
        if _task_status_value(ht.status) in {"wait", "todo", "progress"}:
            ht.status = TaskStatus.cancel
            outcome.cancelled_tasks.append(ht.record_id or str(ht.id))


async def reset_repository_projection(
    session: Any,
    repo_slug: str,
    *,
    acknowledge_retirements: bool = False,
    expected_retirement_uuids: list[uuid.UUID] | None = None,
    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, either by blanket acknowledgement or by
    the exact hub UUIDs expected to retire. The two authorizations cannot be
    combined. Omitting the UUID list keeps blanket acknowledgement unchanged;
    an empty list expects zero retirements.

    Scope: workplans, and their tasks. Tasks of existing workplans are matched
    by ``record_id`` (STATE-WP-0083-T06); rows with no ``record_id`` are left
    alone. A derived task the hub lacks is created; an identified hub task the
    forge no longer derives is cancelled if it is still open.
    """
    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

    source_ambiguities = _derived_identity_ambiguities(derived)
    if source_ambiguities:
        outcome.status = "refused"
        for ambiguity in source_ambiguities:
            outcome.refused.append(
                {
                    "reason": "forge projection contains duplicate canonical identity",
                    **ambiguity,
                }
            )
        outcome.notes.append(
            "Duplicate identities in the authoritative projection require a file-level "
            "decision; the reset will not choose a winner."
        )
        return outcome

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

    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()}

    # Release identifiers held by rows retired before retirement freed them.
    # This runs 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 refusing for
    # reasons having nothing to do with them.
    #
    # A row the forge still derives is exempt: stamping it would change the slug
    # out from under the matching below, so the record would fail to match its
    # own file and never be un-retired.
    for r in rows:
        if r.projection_retired_at is None or RETIRED_SLUG_MARK in (r.slug or ""):
            continue
        if (r.slug or "").strip().lower() in want or str(r.id) in want_by_uuid:
            continue
        before = r.slug
        r.slug = _tombstone_slug(r.slug or str(r.id), r.projection_retired_at)
        outcome.released.append(before)

    # How a row came to claim a key, strongest first. Two rows can claim the
    # same record — `cust-wp-0010` by its own identifier and
    # `workstream-lifecycle-documentation` by path, both pointing at
    # CUST-WP-0010's file. Assigning into `matched` unconditionally let the
    # later one win and dropped the other silently: never matched, so never
    # stale, so never reported by any pass.
    UUID_MATCH, SLUG_MATCH, PATH_MATCH, PREFIX_MATCH = 0, 1, 2, 3

    matched: dict[str, Any] = {}
    claim: dict[str, int] = {}
    displaced: list[Any] = []

    def _claim(key: str, row: Any, strength: int) -> None:
        held = claim.get(key)
        if held is None:
            matched[key], claim[key] = row, strength
            return
        if strength < held:
            # The new claim is stronger; the incumbent loses the record and
            # becomes a retirement candidate rather than disappearing.
            displaced.append(matched[key])
            matched[key], claim[key] = row, strength
        else:
            displaced.append(row)

    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:
            _claim(by_uuid, row, UUID_MATCH)
            continue
        if key in want:
            # The identifier itself derives again — including for a retired row,
            # which is then deliberately un-retired below.
            _claim(key, row, SLUG_MATCH)
            continue
        if row.projection_retired_at is None and RETIRED_SLUG_MARK in (row.slug or ""):
            # A stamped slug with a cleared flag is a resurrected row: the mark
            # is evidence a retirement happened and the flag says it did not.
            # It must land in `matched` under its own key so it shows up as
            # stale — skipping it outright, as a genuinely retired row is
            # skipped, is what left these reporting `noop` forever.
            matched[key] = row
            continue
        if row.projection_retired_at is not None:
            # Already retired: nothing to do, and nothing to re-decide.
            #
            # Past this point only the heuristics remain, and a retired row must
            # not be matched by those. Retirement is a
            # decision, and re-deriving the same file must not silently undo it.
            #
            # Releasing the identifier makes this necessary: the tombstoned slug
            # is no longer an identifier and a legacy row is not UUID-derived, so
            # both re-key guards fall through to path matching — which matches
            # the file the row was retired *for* and resurrects it, alongside the
            # correct record already created from that same file.
            #
            # A record that genuinely returns matches by UUID or by its own
            # identifier above, and is un-retired there.
            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:
                _claim(want_paths[_path_key(bp)], row, PATH_MATCH)
                continue
            cand = [k for k in want if key.startswith(k + "-")]
            if len(cand) == 1:
                _claim(cand[0], row, PREFIX_MATCH)
                continue
        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
    ] + [r for r in displaced if r.projection_retired_at is None]
    # Exact UUIDs and blanket acknowledgement are different authorizations.
    # A call that sets both retires nothing, before any retirement mutation.
    if expected_retirement_uuids is not None and acknowledge_retirements:
        outcome.status = "refused"
        outcome.refused.append(
            {
                "reason": "ambiguous authorization",
                "detail": (
                    "expected_retirement_uuids and acknowledge_retirements "
                    "cannot both be set"
                ),
            }
        )
        outcome.notes.append(
            "Nothing was retired. Pass either an exact retirement set or "
            "blanket acknowledgement, not both."
        )
        return outcome
    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 expected_retirement_uuids is not None:
        # `stale` is the unmatched live rows plus the displaced live rows.
        # A row that still matches, including a historical parent, is not in it.
        actual = {_uuid_key(row.id) for row in stale}
        expected = {_uuid_key(value) for value in expected_retirement_uuids}
        missing = sorted(actual - expected)
        extra = sorted(expected - actual)
        if missing or extra:
            outcome.status = "refused"
            outcome.refused.append(
                {
                    "reason": "expected retirement set does not match stale projections",
                    "missing": missing,
                    "extra": extra,
                }
            )
            outcome.notes.append(
                "Nothing was retired. The expected set must equal the stale "
                "set; partial retirement is not applied."
            )
            return outcome
        blockers = await _exact_retirement_blockers(session, stale)
        if blockers:
            outcome.status = "refused"
            outcome.refused.extend(blockers)
            outcome.notes.append(
                "Nothing was retired. A stale projection with tasks or "
                "dependency edges cannot be retired."
            )
            return outcome
    elif 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)

    existing_ids = [matched[k].id for k, _w in want.items() if k in matched]
    tasks_by_wp: dict[Any, list[Any]] = {}
    if existing_ids:
        loaded = list(
            (
                await session.execute(select(Task).where(Task.workplan_id.in_(existing_ids)))
            ).scalars()
        )
        for task_row in loaded:
            tasks_by_wp.setdefault(task_row.workplan_id, []).append(task_row)

    # Bound histories are receiving support for already existing records only.
    # Verify every parent and task before any mutation; preserve labels, status,
    # retirement stamps and all UUIDs. Missing rows cannot authorize recreation.
    for key, workplan in want.items():
        if not workplan.historical_preservation:
            continue
        row = matched.get(key)
        local = {str(task.id): task for task in tasks_by_wp.get(row.id, [])} if row else {}
        missing = [task.uuid for task in workplan.tasks if task.uuid not in local]
        nonterminal = [task.uuid for task in workplan.tasks
                       if task.uuid in local and local[task.uuid].status not in {"done", "cancel"}]
        if (row is None or str(row.id) != workplan.uuid or missing or nonterminal
                or row.status not in {"done", "finished", "archived", "completed"}):
            outcome.status = "refused"
            outcome.refused.append({
                "reason": "historical binding requires existing parent and task UUIDs",
                "workplan": workplan.record_id, "uuid": workplan.uuid,
                "missing_task_uuids": missing, "nonterminal_task_uuids": nonterminal,
            })
            return outcome

    # Canonical task identity is scoped to the matched current workplan while
    # reconciling. Duplicate rows attached to a retired/displaced workplan are
    # historical evidence and are deliberately not loaded above. Two rows with
    # one identity inside the matched workplan, however, are genuinely
    # ambiguous: the reset cannot know which row owns progress or dependencies.
    current_task_ambiguities: list[dict[str, Any]] = []
    for key, workplan in want.items():
        row = matched.get(key)
        if row is None:
            continue
        desired_record_ids = {
            task.record_id.strip().lower() for task in workplan.tasks
        }
        groups: dict[str, list[Any]] = {}
        for task_row in tasks_by_wp.get(row.id, []):
            record_id = (task_row.record_id or "").strip().lower()
            if record_id and record_id in desired_record_ids:
                groups.setdefault(record_id, []).append(task_row)
        for task_rows in groups.values():
            if len(task_rows) > 1:
                current_task_ambiguities.append(
                    {
                        "reason": "current workplan contains duplicate canonical task identity",
                        "kind": "task",
                        "record_id": task_rows[0].record_id,
                        "workplan": workplan.record_id,
                        "uuids": [str(task_row.id) for task_row in task_rows],
                    }
                )
    if current_task_ambiguities:
        outcome.status = "refused"
        outcome.refused.extend(current_task_ambiguities)
        outcome.notes.append(
            "Retired workplan copies do not block reset, but duplicate identities "
            "inside the matched current workplan require manual disposition."
        )
        return outcome

    # Check every task the reset would create in one query. A derived UUID held
    # by another workplan is a global identity collision. A repeated record_id
    # on a different workplan is not sufficient by itself: terminal rows from
    # older, wrong projections retain that label as history and tasks have no
    # projection-retirement marker. Letting the INSERT discover a UUID collision
    # would turn a deterministic refusal into an IntegrityError.
    creating_tasks: dict[str, tuple[DerivedWorkplan, DerivedTask]] = {}
    for key, workplan in want.items():
        row = matched.get(key)
        local_rows = tasks_by_wp.get(row.id, []) if row is not None else []
        local_record_ids = {
            (task_row.record_id or "").strip().lower()
            for task_row in local_rows
            if (task_row.record_id or "").strip()
        }
        local_uuids = {str(task_row.id) for task_row in local_rows}
        for task in workplan.tasks:
            if (
                task.record_id.strip().lower() not in local_record_ids
                and task.uuid not in local_uuids
            ):
                creating_tasks[task.uuid] = (workplan, task)

    if creating_tasks:
        task_holders = list(
            (
                await session.execute(
                    select(Task)
                    .where(
                        Task.id.in_(
                            [uuid.UUID(task_id) for task_id in creating_tasks]
                        ),
                    )
                )
            ).scalars()
        )
        if task_holders:
            outcome.status = "refused"
            for holder in task_holders:
                workplan, task = creating_tasks[str(holder.id)]
                outcome.refused.append(
                    {
                        "reason": "derived task identifier already belongs to another workplan",
                        "kind": "task",
                        "record_id": task.record_id,
                        "uuid": task.uuid,
                        "workplan": workplan.record_id,
                        "held_by_workplan_id": str(holder.workplan_id),
                    }
                )
            outcome.notes.append(
                "Task identity collisions require an identity decision; the reset "
                "will not overwrite or move the existing row."
            )
            return outcome

    received_rows: dict[str, Any] = {}
    for key, w in want.items():
        row = matched.get(key)
        if row is not None:
            received_rows[key] = row
        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,
                description=w.description,
                status=w.status or "proposed",
                owner=w.owner,
                flavor=w.flavor,
                flavor_promotion_reason=w.flavor_promotion_reason,
                flavor_promoted_from=w.flavor_promoted_from,
                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()
            received_rows[key] = row
            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,
                        description=t.description,
                        status=t.status or "todo",
                        priority=t.priority or "medium",
                        needs_human=t.needs_human,
                        intervention_note=t.intervention_note,
                        blocking_reason=t.blocking_reason,
                        flavor=t.flavor,
                        decision_id=t.decision_id,
                        blocked_on=t.blocked_on,
                        wait_until=t.wait_until,
                    )
                )
            outcome.created.append(w.record_id)
            continue

        if w.historical_preservation:
            outcome.notes.append("Preserved bound historical row and task UUIDs: " + w.record_id)
            continue

        changed = False
        # The title is derived like every other field, and not syncing it left
        # `cust-wp-0010` reading "Domain and Repository Goals" while its file
        # said "Workplan Lifecycle Documentation" — a record correctly
        # identified and correctly backed, describing the wrong work.
        #
        # An empty derived title is not an answer: three activity-core files
        # parse to no title at all, and blanking a real one is worse than
        # leaving it stale.
        if w.title and w.title.strip() and row.title != w.title.strip():
            row.title = w.title.strip()
            changed = True
        if w.description and getattr(row, "description", None) != w.description:
            row.description = w.description
            changed = True
        if getattr(row, "owner", None) != w.owner:
            row.owner = w.owner
            changed = True
        if w.status and row.status != w.status:
            row.status = w.status
            changed = True
        for field_name in ("flavor", "flavor_promotion_reason", "flavor_promoted_from"):
            expected = getattr(w, field_name)
            if getattr(row, field_name, None) != expected:
                setattr(row, field_name, expected)
                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)
        _sync_existing_workplan_tasks(
            session, row, w, tasks_by_wp.get(row.id, []), outcome
        )

    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)

    await _reconcile_source_edges(session, repo, repo_slug, derived, want, received_rows, outcome)

    if (
        outcome.created
        or outcome.updated
        or outcome.retired
        or outcome.released
        or outcome.created_tasks
        or outcome.updated_tasks
        or outcome.cancelled_tasks
        or outcome.created_edges
        or outcome.removed_edges
    ):
        outcome.status = "applied"
    return outcome


# ---------------------------------------------------------------------------
# Source-owned dependency edges (STATE-WP-0094-T05)
# docs/forge-dependency-edge-receiving.md
# ---------------------------------------------------------------------------

EDGE_SOURCE_PREFIX = "forge:"
_CLOSED_TASK_STATUSES = {"done", "cancel"}


def _edge_workplan_closed(status: str | None) -> bool:
    from api.workplan_status import CLOSED_WORKPLAN_STATUSES, normalize_workplan_status

    try:
        return normalize_workplan_status(status or "") in CLOSED_WORKPLAN_STATUSES
    except ValueError:
        return False


def _edge_task_closed(status: str | None) -> bool:
    from api.task_status import normalize_task_status

    try:
        return normalize_task_status(status or "todo", default="todo") in _CLOSED_TASK_STATUSES
    except ValueError:
        return False


def edge_source_owner(repo_slug: str) -> str:
    return f"{EDGE_SOURCE_PREFIX}{repo_slug}"


def _iter_edge_declarations(want: dict[str, DerivedWorkplan]):
    """Yield ``(key, workplan, task_or_None, target)`` for every desired declaration.

    Shared by receiving and the dry-run preview so both apply the same rules:
    historical workplans declare nothing; task-block edges of closed workplans
    and closed tasks are history (C-20 parity).
    """
    for key, w in want.items():
        if w.historical_preservation:
            continue
        for target in w.depends_on_workplans + w.depends_on_tasks:
            yield key, w, None, target
        if _edge_workplan_closed(w.status):
            continue
        for t in w.tasks:
            if not t.depends_on or _edge_task_closed(t.status):
                continue
            for target in t.depends_on:
                yield key, w, t, target


def source_edge_targets(want: dict[str, DerivedWorkplan]) -> set[str]:
    """Every target record id a receipt of ``want`` would resolve."""
    return {target for _, _, _, target in _iter_edge_declarations(want)}


def plan_source_edges(
    want: dict[str, DerivedWorkplan],
    from_workplan_ids: dict[str, Any],
    task_ids: dict[Any, dict[str, Any]],
    targets: dict[str, tuple[str, Any]],
) -> tuple[dict[tuple, dict[str, Any]], list[dict[str, Any]], set[tuple]]:
    """Pure: desired edge keys, unresolved declarations and held from-sides.

    ``from_workplan_ids`` maps a ``want`` key to the from workplan's id (keys
    absent are skipped); ``task_ids`` maps that id to ``{record_id.lower(): task
    id}``; ``targets`` maps a target record id to ``("ok", id)`` or
    ``("missing"|"ambiguous", detail)`` (absent means missing).
    Edge key: ``(from_workplan_id, from_task_id, to_workplan_id, to_task_id, rel)``.
    """
    desired: dict[tuple, dict[str, Any]] = {}
    unresolved: list[dict[str, Any]] = []
    # From-sides with an unresolved declaration keep every owned edge: the
    # unresolved target may be one of them (e.g. newly ambiguous).
    held_sides: set[tuple] = set()
    for key, w, t, target in _iter_edge_declarations(want):
        wp_id = from_workplan_ids.get(key)
        if wp_id is None:
            continue
        from_task_id = None
        if t is not None:
            from_task_id = task_ids.get(wp_id, {}).get(t.record_id.strip().lower())
            if from_task_id is None:
                held_sides.add((wp_id, "*"))
                unresolved.append({"workplan": w.record_id, "from": t.record_id,
                                   "target": target, "reason": "from-task-missing"})
                continue
        side = (wp_id, from_task_id)
        is_task = bool(_TASK_TARGET_RE.search(target))
        state, value = targets.get(target, ("missing", None))
        base = {"workplan": w.record_id, "from": t.record_id if t is not None else w.record_id,
                "target": target}
        if state == "ok" and ((not is_task and value == wp_id) or (is_task and value == from_task_id)):
            state, value = "self", None
        if state != "ok":
            held_sides.add(side)
            unresolved.append({**base, "reason": state, **({"candidates": value} if value else {})})
            continue
        rel = "starts_after" if is_task else "blocks"
        edge_key = (wp_id, from_task_id, None if is_task else value, value if is_task else None, rel)
        desired.setdefault(edge_key, {**base, "relationship_type": rel})
    return desired, unresolved, held_sides


def _edge_dict(edge_key: tuple, info: dict[str, Any] | None = None) -> dict[str, Any]:
    from_wp, from_task, to_wp, to_task, rel = edge_key
    return {
        **(info or {}),
        "from_workplan_id": str(from_wp),
        "from_task_id": str(from_task) if from_task else None,
        "to_workplan_id": str(to_wp) if to_wp else None,
        "to_task_id": str(to_task) if to_task else None,
        "relationship_type": rel,
    }


def classify_source_edges(
    desired: dict[tuple, dict[str, Any]],
    held_sides: set[tuple],
    existing: list[tuple[tuple, Any, str | None]],
    owner: str,
) -> dict[str, list]:
    """Pure: split desired/existing edges into create/external/remove/retain.

    ``existing`` lists ``(edge key, row id, source_owner)`` for every row whose
    from-workplan belongs to the repository. Several rows may share a key (an
    owned row and an API-authored duplicate); they are never collapsed, so an
    external duplicate can neither hide nor protect an owned row.
    """
    owned_keys = {k for k, _, o in existing if o == owner}
    any_keys = {k for k, _, _ in existing}
    plan: dict[str, list] = {"create": [], "external": [], "remove": [], "retain": []}
    for edge_key in sorted(desired, key=str):
        if edge_key in owned_keys:
            continue
        bucket = "external" if edge_key in any_keys else "create"
        plan[bucket].append((edge_key, desired[edge_key]))
    for edge_key, row_id, src_owner in sorted(existing, key=lambda r: (str(r[0]), str(r[1]))):
        if src_owner != owner or edge_key in desired:
            continue
        if (edge_key[0], edge_key[1]) in held_sides or (edge_key[0], "*") in held_sides:
            plan["retain"].append((edge_key, row_id))
        else:
            plan["remove"].append((edge_key, row_id))
    return plan


async def _resolve_edge_target(session: Any, record_id: str, cache: dict) -> tuple[str, Any]:
    """Resolve a declared target fleet-wide to ``("ok", uuid)`` or a reason.

    Workplans match by ADR-007 derived UUID or record-id slug; tasks by derived
    UUID or ``record_id``. Projection-retired workplans (and their tasks) are
    not targets. More than one candidate is ambiguous: no winner is chosen.
    """
    if record_id in cache:
        return cache[record_id]
    from sqlalchemy import func, or_, select

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

    derived = uuid.UUID(derived_record_uuid(record_id))
    if _TASK_TARGET_RE.search(record_id):
        stmt = (
            select(Task.id)
            .join(Workplan, Workplan.id == Task.workplan_id)
            .where(or_(Task.id == derived, Task.record_id == record_id))
            .where(Workplan.projection_retired_at.is_(None))
        )
    else:
        stmt = select(Workplan.id).where(
            or_(Workplan.id == derived, func.lower(Workplan.slug) == record_id.lower())
        ).where(Workplan.projection_retired_at.is_(None))
    ids = sorted({row[0] for row in (await session.execute(stmt)).all()}, key=str)
    if not ids:
        result: tuple[str, Any] = ("missing", None)
    elif len(ids) > 1:
        result = ("ambiguous", [str(i) for i in ids])
    else:
        result = ("ok", ids[0])
    cache[record_id] = result
    return result


async def _reconcile_source_edges(
    session: Any,
    repo: Any,
    repo_slug: str,
    derived: DerivedProjection,
    want: dict[str, DerivedWorkplan],
    received_rows: dict[str, Any],
    outcome: ResetOutcome,
) -> None:
    """Reconcile ``workplan_dependencies`` rows owned by this repository's source.

    Ownership: a row is source-owned iff ``source_owner == forge:<repo-slug>``
    and its from-side belongs to this repository. Only those rows are created or
    removed here. Rows with any other (or no) owner are externally authored and
    preserved, even when they duplicate a declared edge.
    """
    from sqlalchemy import select

    from api.models.task import Task
    from api.models.workplan import Workplan
    from api.models.workplan_dependency import WorkplanDependency

    owner = edge_source_owner(repo_slug)
    await session.flush()

    from_ids = {key: row.id for key, row in received_rows.items() if key in want}
    targets: dict[str, tuple[str, Any]] = {}
    for target in sorted(source_edge_targets(want)):
        targets[target] = await _resolve_edge_target(session, target, targets)
    task_ids: dict[Any, dict[str, Any]] = {}
    for key, w, t, _ in _iter_edge_declarations(want):
        wp_id = from_ids.get(key)
        if t is None or wp_id is None or wp_id in task_ids:
            continue
        task_ids[wp_id] = {
            (tr.record_id or "").strip().lower(): tr.id
            for tr in (await session.execute(select(Task).where(Task.workplan_id == wp_id))).scalars()
        }
    desired, unresolved, held_sides = plan_source_edges(want, from_ids, task_ids, targets)
    outcome.unresolved_edges.extend(unresolved)

    repo_workplan_ids = list(
        (await session.execute(select(Workplan.id).where(Workplan.repo_id == repo.id))).scalars()
    )
    existing_rows = list(
        (
            await session.execute(
                select(WorkplanDependency).where(
                    WorkplanDependency.from_workplan_id.in_(repo_workplan_ids)
                )
            )
        ).scalars()
    ) if repo_workplan_ids else []
    rows_by_id = {e.id: e for e in existing_rows}
    plan = classify_source_edges(
        desired,
        held_sides,
        [((e.from_workplan_id, e.from_task_id, e.to_workplan_id, e.to_task_id, e.relationship_type),
          e.id, e.source_owner) for e in existing_rows],
        owner,
    )
    for edge_key, info in plan["external"]:
        outcome.external_edges.append(_edge_dict(edge_key, info))
    for edge_key, info in plan["create"]:
        from_wp, from_task, to_wp, to_task, rel = edge_key
        session.add(
            WorkplanDependency(
                from_workplan_id=from_wp,
                from_task_id=from_task,
                to_workplan_id=to_wp,
                to_task_id=to_task,
                relationship_type=rel,
                description=f"{info['from']} depends_on {info['target']}",
                source_owner=owner,
                source_commit=derived.commit,
            )
        )
        outcome.created_edges.append(_edge_dict(edge_key, info))
    for edge_key, row_id in plan["retain"]:
        outcome.notes.append(
            f"Retained source-owned edge {row_id}: its from-side has an unresolved declaration."
        )
    for edge_key, row_id in plan["remove"]:
        await session.delete(rows_by_id[row_id])
        outcome.removed_edges.append(_edge_dict(edge_key, {"id": str(row_id)}))
    # Integrity failures surface here, inside the caller's transaction.
    await session.flush()


# ---------------------------------------------------------------------------
# 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