state-hub/api/services/task_record_id_backfill.py

"""Backfill canonical record ids onto existing task rows (STATE-WP-0083-T06).

Only the repository files hold the mapping. A file task declares both its
canonical id and the projection UUID it was registered under:

    ```task
    id: CUST-WP-0067-T01
    state_hub_task_id: "f3608db4-..."
    ```

so the pairing can be read directly rather than guessed from titles. Anything a
file does not claim is left alone: a task row whose canonical id cannot be
established keeps `record_id` null, and the reset continues to refuse to act on
it. An unknown identity must stay unknown rather than be inferred.
"""

from __future__ import annotations

import re
from dataclasses import dataclass, field
from pathlib import Path
from typing import Any

_TASK_BLOCK_RE = re.compile(r"```task\s*\n(.*?)\n```", re.DOTALL)
_ID_RE = re.compile(r"^id:\s*(\S+)", re.MULTILINE)
_UUID_RE = re.compile(r'state_hub_task_id:\s*"?([0-9a-f-]{36})"?')
_FRONTMATTER_ID_RE = re.compile(r"^id:\s*(\S+)", re.MULTILINE)
# A task id written as a bare "T01" is unique only inside its own workplan.
# Storing it as a canonical identifier makes every workplan's first task share
# one identity — 51 such ids were assigned to 148 rows before this was caught.
_SHORT_TASK_ID_RE = re.compile(r"^T\d+$", re.IGNORECASE)


def qualify_task_id(task_id: str, workplan_id: str | None) -> str | None:
    """Return a canonical task id, or None when identity cannot be established."""
    task_id = task_id.strip()
    if not _SHORT_TASK_ID_RE.fullmatch(task_id):
        return task_id
    if not workplan_id:
        # Unqualifiable: leaving it unidentified is correct, since an identity
        # that is not unique is worse than none.
        return None
    return f"{workplan_id.strip()}-{task_id.upper()}"


@dataclass
class BackfillReport:
    scanned_files: int = 0
    pairs_found: int = 0
    updated: int = 0
    already_set: int = 0
    conflicts: list[dict[str, str]] = field(default_factory=list)
    unmatched_uuids: int = 0

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema": "state-hub.task-record-id-backfill.v1",
            "scanned_files": self.scanned_files,
            "pairs_found": self.pairs_found,
            "updated": self.updated,
            "already_set": self.already_set,
            "unmatched_uuids": self.unmatched_uuids,
            "conflicts": self.conflicts,
        }


def collect_pairs(roots: list[Path]) -> tuple[dict[str, str], BackfillReport]:
    """Map projection UUID -> canonical record id, from workplan files."""
    report = BackfillReport()
    pairs: dict[str, str] = {}
    for root in roots:
        wp_dir = root / "workplans"
        if not wp_dir.is_dir():
            continue
        for path in sorted(wp_dir.rglob("*.md")):
            if path.name.startswith("."):
                continue
            try:
                text = path.read_text(encoding="utf-8")
            except (OSError, UnicodeDecodeError):
                continue
            report.scanned_files += 1
            head = text.split("---", 2)[1] if text.startswith("---") and text.count("---") >= 2 else ""
            wp_match = _FRONTMATTER_ID_RE.search(head)
            workplan_id = wp_match.group(1).strip() if wp_match else None
            for block in _TASK_BLOCK_RE.finditer(text):
                body = block.group(1)
                rid = _ID_RE.search(body)
                uid = _UUID_RE.search(body)
                if not rid or not uid:
                    continue
                record_id = qualify_task_id(rid.group(1), workplan_id)
                if record_id is None:
                    continue
                task_uuid = uid.group(1)
                prior = pairs.get(task_uuid)
                if prior and prior != record_id:
                    # One UUID claimed by two canonical ids: a duplicate
                    # registration. Recording it and skipping is the only safe
                    # option — picking one would fabricate an identity.
                    report.conflicts.append(
                        {"uuid": task_uuid, "first": prior, "second": record_id}
                    )
                    continue
                pairs[task_uuid] = record_id
    report.pairs_found = len(pairs)
    return pairs, report


async def backfill_task_record_ids(
    session: Any, roots: list[Path], *, dry_run: bool = True
) -> BackfillReport:
    from sqlalchemy import select

    from api.models.task import Task

    pairs, report = collect_pairs(roots)
    if not pairs:
        return report

    rows = list((await session.execute(select(Task))).scalars())
    by_id = {str(r.id): r for r in rows}
    for task_uuid, record_id in pairs.items():
        row = by_id.get(task_uuid)
        if row is None:
            report.unmatched_uuids += 1
            continue
        if row.record_id == record_id:
            report.already_set += 1
            continue
        if row.record_id and row.record_id != record_id:
            report.conflicts.append(
                {"uuid": task_uuid, "first": row.record_id, "second": record_id}
            )
            continue
        report.updated += 1
        if not dry_run:
            row.record_id = record_id
    return report


async def backfill_from_forge(
    session: Any,
    repo_slugs: list[str],
    *,
    forge_base: str | None = None,
    dry_run: bool = True,
) -> BackfillReport:
    """Backfill from repositories cloned out of the forge.

    The local-path variant above needs a workstation checkout, which central
    does not have and should not depend on: `ADR-012` decision 1 makes the forge
    the projection source, and a backfill sourced from someone's laptop would
    reintroduce exactly the coupling that ADR removes.

    Central can clone the forge directly, so it reads the pairing from the same
    place it derives everything else.
    """
    import tempfile

    from api.services.forge_projection import (
        DEFAULT_FORGE_BASE,
        ForgeDeriveError,
        _run_git,
    )

    base = forge_base or DEFAULT_FORGE_BASE
    report = BackfillReport()
    pairs: dict[str, str] = {}

    for slug in repo_slugs:
        url = f"{base.rstrip('/')}/{slug}.git"
        with tempfile.TemporaryDirectory(prefix=f"backfill-{slug}-") as tmp:
            try:
                _run_git("clone", "--depth", "1", "--quiet", url, tmp)
            except (ForgeDeriveError, Exception):
                # A repository that cannot be read contributes nothing. It must
                # not silently reduce what the rest can identify.
                continue
            repo_pairs, repo_report = collect_pairs([Path(tmp)])
            report.scanned_files += repo_report.scanned_files
            report.conflicts.extend(repo_report.conflicts)
            for uid, rid in repo_pairs.items():
                prior = pairs.get(uid)
                if prior and prior != rid:
                    report.conflicts.append({"uuid": uid, "first": prior, "second": rid})
                    continue
                pairs[uid] = rid

    report.pairs_found = len(pairs)
    if not pairs:
        return report

    from sqlalchemy import select

    from api.models.task import Task

    rows = list((await session.execute(select(Task))).scalars())
    by_id = {str(r.id): r for r in rows}
    for uid, rid in pairs.items():
        row = by_id.get(uid)
        if row is None:
            report.unmatched_uuids += 1
            continue
        if row.record_id == rid:
            report.already_set += 1
            continue
        if row.record_id and row.record_id != rid:
            report.conflicts.append({"uuid": uid, "first": row.record_id, "second": rid})
            continue
        report.updated += 1
        if not dry_run:
            row.record_id = rid
    return report