state-hub/tests/test_forge_projection.py

"""Deriving a projection from the forge (STATE-WP-0083-T01).

The properties that matter are identity and determinism: a forge-derived
projection must compute the same record identities as repo-manager, and the same
commit must always yield the same projection. Without both, the reset in T03
cannot be verified against anything.
"""

from __future__ import annotations

from pathlib import Path

import pytest

from api.services import forge_credential as fc

from api.services import forge_projection as fp


def test_identity_matches_the_fleet_derivation():
    """Same namespace as ADR-007, so overlay and forge agree on identity."""
    assert fp.derived_record_uuid("CUST-WP-0067") == "16249302-2767-55df-aec0-d92c2751c225"
    assert fp.derived_record_uuid("CUST-WP-0067-T01") == "f3608db4-20a5-58fb-a965-885eb14858af"


def _repo(tmp_path: Path) -> Path:
    root = tmp_path / "demo"
    (root / "workplans" / "archived").mkdir(parents=True)
    (root / "workplans" / "DEMO-WP-0001-a.md").write_text(
        "---\nid: DEMO-WP-0001\ntype: workplan\ntitle: \"First\"\nstatus: active\n---\n\n"
        "## Do the thing\n\n```task\nid: DEMO-WP-0001-T01\nstatus: todo\npriority: high\n```\n\n"
        "## Do the other\n\n```task\nid: DEMO-WP-0001-T02\nstatus: done\npriority: low\n```\n",
        encoding="utf-8",
    )
    (root / "workplans" / "archived" / "260101-DEMO-WP-0002-b.md").write_text(
        "---\nid: DEMO-WP-0002\ntype: workplan\ntitle: \"Second\"\nstatus: finished\n---\n\n# b\n",
        encoding="utf-8",
    )
    (root / "workplans" / "NOTES.md").write_text(
        "---\nid: NOT-A-WORKPLAN\ntype: note\n---\n\n# not a workplan\n", encoding="utf-8"
    )
    return root


def test_derives_workplans_tasks_and_archived_flag(tmp_path):
    p = fp.derive_from_checkout(_repo(tmp_path), "demo", "abc123")
    assert [w.record_id for w in p.workplans] == ["DEMO-WP-0001", "DEMO-WP-0002"]
    first, second = p.workplans
    assert first.status == "active" and first.archived is False
    assert second.archived is True
    assert p.task_count == 2
    assert [t.record_id for t in first.tasks] == ["DEMO-WP-0001-T01", "DEMO-WP-0001-T02"]


def test_ignores_files_that_are_not_workplans(tmp_path):
    """Selection is by `type: workplan`; anything else is not this hub's business."""
    p = fp.derive_from_checkout(_repo(tmp_path), "demo", "abc123")
    assert all(w.record_id != "NOT-A-WORKPLAN" for w in p.workplans)


def test_task_titles_fall_back_to_the_preceding_heading(tmp_path):
    p = fp.derive_from_checkout(_repo(tmp_path), "demo", "abc123")
    titles = [t.title for t in p.workplans[0].tasks]
    assert titles == ["Do the thing", "Do the other"]


def test_identifiers_are_derived_not_read_from_the_file(tmp_path):
    """A forge projection must not inherit whatever id a file happens to carry."""
    root = _repo(tmp_path)
    f = root / "workplans" / "DEMO-WP-0001-a.md"
    f.write_text(
        f.read_text(encoding="utf-8").replace(
            "status: active",
            'status: active\nstate_hub_workstream_id: "00000000-0000-4000-8000-000000000000"',
        ),
        encoding="utf-8",
    )
    p = fp.derive_from_checkout(root, "demo", "abc123")
    assert p.workplans[0].uuid == fp.derived_record_uuid("DEMO-WP-0001")
    assert p.workplans[0].uuid != "00000000-0000-4000-8000-000000000000"


def test_same_input_yields_identical_output(tmp_path):
    root = _repo(tmp_path)
    assert fp.derive_from_checkout(root, "demo", "abc").to_dict() == \
           fp.derive_from_checkout(root, "demo", "abc").to_dict()


def test_missing_workplans_directory_is_empty_not_an_error(tmp_path):
    (tmp_path / "bare").mkdir()
    p = fp.derive_from_checkout(tmp_path / "bare", "bare", "abc")
    assert p.workplans == [] and p.commit == "abc"


def test_clone_failure_is_reported_not_swallowed(monkeypatch):
    def boom(*a, **k):
        raise fp.ForgeDeriveError("repository not found")
    monkeypatch.setattr(fp, "_run_git", boom)
    with pytest.raises(fp.ForgeDeriveError, match="not found"):
        fp.derive_from_forge("nope")


class TestReset:
    """Applying the reset (STATE-WP-0083-T03).

    The properties worth guarding are the refusals, not the happy path. A reset
    that quietly retires a record someone still needs is worse than one that
    does nothing.
    """

    @staticmethod
    def _derived(*records):
        wps = []
        for rid, status, path in records:
            wps.append(
                fp.DerivedWorkplan(
                    record_id=rid, uuid=fp.derived_record_uuid(rid), title=rid,
                    status=status, relative_path=path, archived=False, tasks=[],
                )
            )
        return fp.DerivedProjection(repo_slug="demo", commit="c0ffee", workplans=wps)

    @pytest.mark.asyncio
    async def test_refuses_retirement_unless_acknowledged(self, monkeypatch):
        """A record that stops deriving may mean a deleted file — or a wrong branch."""
        session = _FakeSession(
            repo=_Repo(),
            rows=[_Row(slug="demo-wp-0001", status="active", path="workplans/a.md")],
        )
        out = await fp.reset_repository_projection(
            session, "demo", derived=self._derived()  # forge has nothing
        )
        assert out.status == "refused"
        assert out.refused and out.refused[0]["slug"] == "demo-wp-0001"
        assert out.retired == []
        assert session.committed is False

    @pytest.mark.asyncio
    async def test_retires_when_acknowledged(self):
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        session = _FakeSession(repo=_Repo(), rows=[row])
        out = await fp.reset_repository_projection(
            session, "demo", derived=self._derived(), acknowledge_retirements=True
        )
        assert out.status == "applied" and out.retired == ["demo-wp-0001"]
        assert row.projection_retired_at is not None
        assert row.projection_retired_reason == fp.RETIRE_REASON

    @pytest.mark.asyncio
    async def test_retirement_is_not_deletion(self):
        """Hub-native records reference workplans with RESTRICT; the row survives."""
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        session = _FakeSession(repo=_Repo(), rows=[row])
        await fp.reset_repository_projection(
            session, "demo", derived=self._derived(), acknowledge_retirements=True
        )
        assert row in session.rows
        assert session.deleted == []

    @pytest.mark.asyncio
    async def test_unretires_a_record_that_derives_again(self):
        from datetime import datetime, timezone
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        row.projection_retired_at = datetime.now(tz=timezone.utc)
        row.projection_retired_reason = fp.RETIRE_REASON
        session = _FakeSession(repo=_Repo(), rows=[row])
        out = await fp.reset_repository_projection(
            session, "demo",
            derived=self._derived(("DEMO-WP-0001", "active", "workplans/a.md")),
        )
        assert out.status == "applied"
        assert row.projection_retired_at is None

    @pytest.mark.asyncio
    async def test_unregistered_repository_is_refused_not_created(self):
        session = _FakeSession(repo=None, rows=[])
        out = await fp.reset_repository_projection(session, "nope", derived=self._derived())
        assert out.status == "refused"
        assert "not registered" in out.refused[0]["reason"]

    @pytest.mark.asyncio
    async def test_records_the_commit_it_derived_from(self):
        row = _Row(slug="demo-wp-0001", status="proposed", path="workplans/a.md")
        session = _FakeSession(repo=_Repo(), rows=[row])
        await fp.reset_repository_projection(
            session, "demo",
            derived=self._derived(("DEMO-WP-0001", "active", "workplans/a.md")),
        )
        assert row.derived_from_commit == "c0ffee"
        assert row.status == "active"


class _Repo:
    def __init__(self):
        import uuid as _u
        self.id = _u.uuid4()
        self.topic_id = None


class _Row:
    def __init__(self, slug, status, path, title=None):
        import uuid as _u
        self.id = _u.uuid4()
        self.slug = slug
        self.status = status
        # The title is a derived field the reset syncs; the fixture carried no
        # `title` at all, so it could not have caught the drift that left
        # cust-wp-0010 describing another workplan's work.
        self.title = title
        self.backing_relative_path = path
        self.backing_filename = path.rsplit("/", 1)[-1]
        self.backing_archived = False
        self.projection_retired_at = None
        self.projection_retired_reason = None
        self.derived_from_commit = None


class _FakeSession:
    """Stands in for AsyncSession: enough to prove intent without a database."""

    def __init__(self, repo, rows, foreign=None, slug_clash=None, task_rows=None):
        self._repo = repo
        self.rows = list(rows)
        self._foreign = list(foreign or [])
        self._slug_clash = list(slug_clash or [])
        self._task_rows = task_rows
        self.added = []
        self.deleted = []
        self.committed = False
        self._calls = 0

    async def execute(self, *_a, **_k):
        self._calls += 1
        repo, rows = self._repo, self.rows
        # 1 resolves the repo, 2 loads its workplans, 3 is the identifier
        # lookup, 4 the slug lookup. On the update-only path (no creates)
        # the 3rd call is the existing-task load instead.
        update_only = (
            self._task_rows is not None
            and not self._foreign
            and not self._slug_clash
        )
        if self._calls == 2:
            payload = rows
        elif update_only and self._calls == 3:
            payload = self._task_rows
        elif self._calls == 3:
            payload = self._foreign
        elif self._calls == 4:
            payload = self._slug_clash
        elif self._calls == 5:
            payload = self._task_rows or []
        else:
            payload = []

        class R:
            def scalar_one_or_none(self_inner):
                return repo

            def scalars(self_inner):
                return iter(payload)

        return R()

    def add(self, obj):
        self.added.append(obj)

    def delete(self, obj):
        self.deleted.append(obj)

    async def flush(self):
        return None

    async def commit(self):
        self.committed = True

    async def rollback(self):
        self.rolled_back = True


class TestCollisionRefusal:
    """A colliding identifier must refuse, not raise (STATE-WP-0083-T03).

    net-kingdom's ADHOC-2026-08-23 derives to an identifier another repository
    already holds — the case CUST-WP-0066 documents. The reset previously failed
    on a database constraint, which tells the caller nothing they can act on.
    """

    @staticmethod
    def _derived(rid="DEMO-WP-0001"):
        return fp.DerivedProjection(
            repo_slug="demo", commit="c0ffee",
            workplans=[fp.DerivedWorkplan(
                record_id=rid, uuid=fp.derived_record_uuid(rid), title=rid,
                status="active", relative_path="workplans/a.md",
                archived=False, tasks=[])],
        )

    @pytest.mark.asyncio
    async def test_refuses_when_the_identifier_belongs_elsewhere(self):
        derived = self._derived()
        foreign = _Row(slug="held-by-someone-else", status="finished", path="workplans/x.md")
        foreign.id = __import__("uuid").UUID(derived.workplans[0].uuid)
        session = _FakeSession(repo=_Repo(), rows=[], foreign=[foreign])
        out = await fp.reset_repository_projection(session, "demo", derived=derived)
        assert out.status == "refused"
        assert out.refused[0]["reason"].startswith("derived identifier already belongs")
        assert out.refused[0]["held_by_slug"] == "held-by-someone-else"
        assert out.created == [] and session.added == []

    @pytest.mark.asyncio
    async def test_acknowledging_retirements_does_not_authorise_a_collision(self):
        """Different decision, different authorisation."""
        derived = self._derived()
        foreign = _Row(slug="held-by-someone-else", status="finished", path="workplans/x.md")
        foreign.id = __import__("uuid").UUID(derived.workplans[0].uuid)
        session = _FakeSession(repo=_Repo(), rows=[], foreign=[foreign])
        out = await fp.reset_repository_projection(
            session, "demo", derived=derived, acknowledge_retirements=True
        )
        assert out.status == "refused"
        assert session.added == []

    @pytest.mark.asyncio
    async def test_no_collision_still_creates(self):
        derived = self._derived()
        session = _FakeSession(repo=_Repo(), rows=[], foreign=[])
        out = await fp.reset_repository_projection(session, "demo", derived=derived)
        assert out.status == "applied" and out.created == ["DEMO-WP-0001"]


class TestFleetReset:
    """The fleet form must be a loop over the repository form (T04).

    Its value is entirely in what it does with failure: a wide reset that stops
    at the first refusal is one nobody can run, because there is always one
    unresolved repository somewhere.
    """

    class _Factory:
        def __init__(self, sessions):
            self._sessions = list(sessions)

        def __call__(self):
            session = self._sessions.pop(0)
            class Ctx:
                async def __aenter__(self_inner):
                    return session
                async def __aexit__(self_inner, *a):
                    return False
            return Ctx()

    @staticmethod
    def _derived(rid):
        return fp.DerivedProjection(
            repo_slug="x", commit="c0ffee",
            workplans=[fp.DerivedWorkplan(
                record_id=rid, uuid=fp.derived_record_uuid(rid), title=rid,
                status="active", relative_path="workplans/a.md", archived=False, tasks=[])])

    @pytest.mark.asyncio
    async def test_a_refusal_does_not_stop_the_pass(self, monkeypatch):
        calls = []

        async def fake(session, slug, **kw):
            calls.append(slug)
            out = fp.ResetOutcome(repo_slug=slug, commit="c0ffee", status="applied")
            if slug == "bad":
                out.status = "refused"
                out.refused.append({"reason": "would be retired", "slug": "x"})
            else:
                out.updated.append("A-WP-0001")
            return out

        monkeypatch.setattr(fp, "reset_repository_projection", fake)
        s = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(3)]
        res = await fp.reset_fleet_projection(self._Factory(s), ["good", "bad", "also-good"])
        assert calls == ["good", "bad", "also-good"]
        d = res.to_dict()
        assert d["by_status"] == {"applied": 2, "refused": 1}
        assert d["totals"]["updated"] == 2

    @pytest.mark.asyncio
    async def test_an_error_does_not_stop_the_pass(self, monkeypatch):
        async def fake(session, slug, **kw):
            if slug == "boom":
                raise RuntimeError("clone failed")
            return fp.ResetOutcome(repo_slug=slug, commit="c", status="noop")

        monkeypatch.setattr(fp, "reset_repository_projection", fake)
        s = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(3)]
        res = await fp.reset_fleet_projection(self._Factory(s), ["a", "boom", "b"])
        d = res.to_dict()
        assert d["errored"] == 1 and "clone failed" in res.errors["boom"]
        assert set(res.results) == {"a", "b"}

    @pytest.mark.asyncio
    async def test_only_applied_repositories_are_committed(self, monkeypatch):
        async def fake(session, slug, **kw):
            out = fp.ResetOutcome(repo_slug=slug, commit="c", status="applied")
            if slug == "refuser":
                out.status = "refused"
            else:
                out.updated.append("A-WP-0001")
            return out

        monkeypatch.setattr(fp, "reset_repository_projection", fake)
        s = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(2)]
        await fp.reset_fleet_projection(self._Factory(s), ["applier", "refuser"])
        assert s[0].committed is True
        assert s[1].committed is False


class TestSlugCollisionRefusal:
    """slug carries its own unique constraint (STATE-WP-0083-T04).

    Checking the identifier alone left disaster-control raising IntegrityError:
    two repositories can derive different identifiers whose slugs still collide.
    """

    @pytest.mark.asyncio
    async def test_refuses_when_the_slug_belongs_elsewhere(self):
        derived = fp.DerivedProjection(
            repo_slug="demo", commit="c0ffee",
            workplans=[fp.DerivedWorkplan(
                record_id="REPO-WP-0001", uuid=fp.derived_record_uuid("REPO-WP-0001"),
                title="x", status="active", relative_path="workplans/a.md",
                archived=False, tasks=[])])
        clash = _Row(slug="repo-wp-0001", status="finished", path="workplans/other.md")
        session = _FakeSession(repo=_Repo(), rows=[], foreign=[], slug_clash=[clash])
        out = await fp.reset_repository_projection(session, "demo", derived=derived)
        assert out.status == "refused"
        assert out.refused[0]["reason"].startswith("slug already belongs")
        assert session.added == []


class TestUnreadableIsNotMissing:
    """STATE-WP-0084-T01.

    A repository central is not permitted to read and a repository whose
    records no longer derive authorise opposite things. Every test here exists
    to keep the retirement path unreachable from an answer that cannot support
    it — by construction, not by the reset happening to fail first.
    """

    @pytest.mark.parametrize(
        "stderr",
        [
            "fatal: could not read Username for 'https://forgejo.coulomb.social'",
            "remote: Invalid username or password.\nfatal: Authentication failed",
            "fatal: could not read Username for 'https://f': terminal prompts disabled",
            "fatal: repository 'https://forgejo.coulomb.social/rapp-openbao.git' not found",
            "fatal: unable to access '...': The requested URL returned error: 403",
        ],
    )
    def test_permission_shaped_failures_are_classified_unreadable(self, stderr, monkeypatch):
        def boom(*a, **k):
            raise fp.ForgeDeriveError(stderr)

        monkeypatch.setattr(fp, "_run_git", boom)
        with pytest.raises(fp.ForgeUnreadableError):
            fp.derive_from_forge("rapp-openbao")

    def test_a_genuine_fault_stays_a_plain_error(self, monkeypatch):
        def boom(*a, **k):
            raise fp.ForgeDeriveError("fatal: early EOF\nfatal: index-pack failed")

        monkeypatch.setattr(fp, "_run_git", boom)
        with pytest.raises(fp.ForgeDeriveError) as exc:
            fp.derive_from_forge("demo")
        assert not isinstance(exc.value, fp.ForgeUnreadableError)

    def test_unreadable_is_a_derive_error_so_old_callers_still_catch_it(self):
        assert issubclass(fp.ForgeUnreadableError, fp.ForgeDeriveError)

    def test_a_checkout_without_workplans_cannot_evidence_absence(self, tmp_path):
        (tmp_path / "bare").mkdir()
        p = fp.derive_from_checkout(tmp_path / "bare", "bare", "abc")
        assert p.workplans == []
        assert p.records_source_present is False
        assert p.retirement_eligible is False

    def test_a_real_checkout_can(self, tmp_path):
        p = fp.derive_from_checkout(_repo(tmp_path), "demo", "abc")
        assert p.records_source_present is True
        assert p.retirement_eligible is True

    def test_the_diff_withholds_stale_rather_than_computing_it(self):
        derived = fp.DerivedProjection(
            repo_slug="demo", commit="c0ffee", records_source_present=False
        )
        hub = [{"id": "11111111-1111-1111-1111-111111111111", "slug": "demo-wp-0001",
                "status": "active", "backing_relative_path": "workplans/a.md"}]
        d = fp.diff_against_hub(derived, hub, {})
        assert d.stale == []
        assert d.would_remove == 0
        assert d.stale_withheld

    @pytest.mark.asyncio
    async def test_an_empty_source_cannot_retire_even_when_acknowledged(self):
        derived = fp.DerivedProjection(
            repo_slug="demo", commit="c0ffee", records_source_present=False
        )
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        session = _FakeSession(repo=_Repo(), rows=[row])
        out = await fp.reset_repository_projection(
            session, "demo", derived=derived, acknowledge_retirements=True
        )
        assert out.status == "refused"
        assert out.retired == []
        assert row.projection_retired_at is None
        assert out.refused[0]["reason"].startswith("source produced no records")

    @pytest.mark.asyncio
    async def test_an_unreadable_repository_reports_unreadable_not_error(self, monkeypatch):
        def boom(*a, **k):
            raise fp.ForgeUnreadableError("rapp-openbao could not be read from the forge")

        monkeypatch.setattr(fp, "derive_from_forge", boom)
        session = _FakeSession(repo=_Repo(), rows=[])
        out = await fp.reset_repository_projection(session, "rapp-openbao")
        assert out.status == "unreadable"
        assert out.retired == [] and out.created == [] and out.updated == []

    @pytest.mark.asyncio
    async def test_the_fleet_keeps_unreadable_out_of_the_error_bucket(self, monkeypatch):
        async def fake(session, slug, **kw):
            if slug == "private":
                out = fp.ResetOutcome(repo_slug=slug, commit="", status="unreadable")
                out.refused.append({"reason": "repository could not be read from the forge",
                                    "slug": slug, "detail": "could not read Username"})
                return out
            if slug == "broken":
                raise RuntimeError("index-pack failed")
            out = fp.ResetOutcome(repo_slug=slug, commit="c0ffee", status="applied")
            out.updated.append("A-WP-0001")
            return out

        monkeypatch.setattr(fp, "reset_repository_projection", fake)
        sessions = [_FakeSession(repo=_Repo(), rows=[]) for _ in range(3)]
        outcome = await fp.reset_fleet_projection(
            TestFleetReset._Factory(sessions), ["ok", "private", "broken"]
        )
        assert list(outcome.unreadable) == ["private"]
        assert list(outcome.errors) == ["broken"]
        assert "private" not in outcome.results
        d = outcome.to_dict()
        assert d["unreadable_count"] == 1 and d["errored"] == 1
        assert d["repositories"] == 3


class TestForgeCredential:
    """Optional forge read credential (STATE-WP-0084-T03).

    Absent is a valid state: a hub with no credential must still derive every
    public repository. The credential must never reach argv, a log, or an
    exception — the places a secret leaks without anyone deciding to leak it.
    """

    @pytest.fixture(autouse=True)
    def _clear_cache(self):
        """The resolved credential is cached for 5 minutes in production.

        That cache is deliberate — a fleet reset of 121 repositories must not
        authenticate to OpenBao 121 times — so tests clear it rather than
        disable it, and exercise the same code path production uses.
        """
        fc.reset_cache()
        yield
        fc.reset_cache()

    def test_absent_credential_is_none_not_empty_string(self, monkeypatch):
        monkeypatch.delenv(fc.TOKEN_ENV, raising=False)
        monkeypatch.delenv(fc.TOKEN_FILE_ENV, raising=False)
        assert fc.forge_read_token() is None

    def test_a_file_is_preferred_over_the_environment(self, tmp_path, monkeypatch):
        """Kubernetes rotates a mounted file without a redeploy."""
        f = tmp_path / "token"
        f.write_text("from-file\n", encoding="utf-8")
        monkeypatch.setenv(fc.TOKEN_ENV, "from-env")
        monkeypatch.setenv(fc.TOKEN_FILE_ENV, str(f))
        assert fc.forge_read_token() == "from-file"

    def test_an_unreadable_token_file_does_not_fall_back_silently(self, monkeypatch):
        """Falling back to a stale env value would hide a broken mount."""
        monkeypatch.setenv(fc.TOKEN_FILE_ENV, "/nonexistent/token")
        monkeypatch.setenv(fc.TOKEN_ENV, "from-env")
        assert fc.forge_read_token() is None

    def test_openbao_is_the_last_resort_not_the_first(self, tmp_path, monkeypatch):
        """A file or env value must not trigger a network call."""
        f = tmp_path / "token"
        f.write_text("local", encoding="utf-8")
        monkeypatch.setenv(fc.TOKEN_FILE_ENV, str(f))
        monkeypatch.setattr(
            fc, "_from_openbao", lambda: pytest.fail("OpenBao consulted unnecessarily")
        )
        assert fc.forge_read_token() == "local"

    def test_openbao_failure_resolves_to_absent_not_an_exception(self, monkeypatch):
        """A hub that cannot reach OpenBao must still derive public repos.

        Raising here would turn "nine repositories are unreadable" into "the
        whole derivation pass failed" — the outcome STATE-WP-0084-T01 exists to
        prevent.
        """
        monkeypatch.delenv(fc.TOKEN_FILE_ENV, raising=False)
        monkeypatch.delenv(fc.TOKEN_ENV, raising=False)
        monkeypatch.setenv(fc.OPENBAO_ADDR_ENV, "https://openbao.invalid")
        monkeypatch.setenv(fc.SECRET_PATH_ENV, "kv/data/forge")
        monkeypatch.setenv(fc.OPENBAO_ROLE_ENV, "state-hub-forge-derivation")
        monkeypatch.setenv(fc.OPENBAO_JWT_PATH_ENV, "/nonexistent/sa-token")
        assert fc.forge_read_token() is None

    def test_openbao_unwraps_kv_v2(self, tmp_path, monkeypatch):
        jwt = tmp_path / "sa"
        jwt.write_text("jwt-value", encoding="utf-8")
        monkeypatch.delenv(fc.TOKEN_FILE_ENV, raising=False)
        monkeypatch.delenv(fc.TOKEN_ENV, raising=False)
        monkeypatch.setenv(fc.OPENBAO_ADDR_ENV, "https://openbao.test")
        monkeypatch.setenv(fc.SECRET_PATH_ENV, "kv/data/forge")
        monkeypatch.setenv(fc.OPENBAO_ROLE_ENV, "state-hub-forge-derivation")
        monkeypatch.setenv(fc.OPENBAO_JWT_PATH_ENV, str(jwt))

        class R:
            def __init__(self, payload):
                self._p = payload

            def raise_for_status(self):
                return None

            def json(self):
                return self._p

        class C:
            def __enter__(self):
                return self

            def __exit__(self, *a):
                return False

            def post(self, url, json):
                assert json["jwt"] == "jwt-value"
                assert json["role"] == "state-hub-forge-derivation"
                return R({"auth": {"client_token": "bao-token"}})

            def get(self, url, headers):
                assert headers["X-Vault-Token"] == "bao-token"
                return R({"data": {"data": {"token": "forge-secret"}}})

        monkeypatch.setattr(fc.httpx, "Client", lambda **kw: C())
        assert fc.forge_read_token() == "forge-secret"

    def test_credential_never_appears_in_argv(self, monkeypatch):
        """`-c http.extraHeader=` would put the token in every ps listing."""
        seen = {}

        class P:
            returncode = 0
            stdout = ""
            stderr = ""

        def fake_run(cmd, **kw):
            seen["cmd"] = cmd
            seen["env"] = kw.get("env") or {}
            return P()

        monkeypatch.setattr(fp.subprocess, "run", fake_run)
        fp._run_git("clone", "url", "dir", token="s3cret")
        assert not any("s3cret" in part for part in seen["cmd"])
        assert seen["env"]["GIT_CONFIG_COUNT"] == "1"
        assert "s3cret" not in seen["env"]["GIT_CONFIG_KEY_0"]

    def test_credential_is_redacted_from_failures(self, monkeypatch):
        class P:
            returncode = 128
            stdout = ""
            stderr = "fatal: auth failed using s3cret"

        monkeypatch.setattr(fp.subprocess, "run", lambda *a, **k: P())
        with pytest.raises(fp.ForgeDeriveError) as exc:
            fp._run_git("clone", token="s3cret")
        assert "s3cret" not in str(exc.value) and "***" in str(exc.value)

    def test_no_credential_still_runs(self, monkeypatch):
        class P:
            returncode = 0
            stdout = "ok"
            stderr = ""
        seen = {}

        def fake_run(cmd, **kw):
            seen["env"] = kw.get("env") or {}
            return P()

        monkeypatch.setattr(fp.subprocess, "run", fake_run)
        assert fp._run_git("status") == "ok"
        assert "GIT_CONFIG_COUNT" not in seen["env"]


class TestRekeyIsNotRename:
    """A changed identifier is a new record, not a moved file.

    The ad-hoc requalification (CUST-WP-0066) keeps the filename by design, so
    for those records a re-key never changes the path. Matching on path there
    cannot distinguish the two cases, and choosing rename updates a row whose
    UUID still encodes the old identifier — leaving the file and the hub
    disagreeing about what the record is called.
    """

    class _Row:
        def __init__(self, id, slug, path):
            self.id = id
            self.slug = slug
            self.backing_relative_path = path
            self.projection_retired_at = None
            self.status = "finished"

    def test_a_derived_row_is_not_matched_by_path(self):
        """uuid5 means identity is a function of the identifier."""
        import uuid as _u
        old = self._Row(
            _u.UUID(fp.derived_record_uuid("ADHOC-2026-08-25")),
            "adhoc-2026-08-25",
            "workplans/ADHOC-2026-08-25.md",
        )
        assert fp._identity_is_derived(old)

    def test_a_legacy_row_keeps_path_matching(self):
        """v4 rows predate derived identity; there the identifier is a label."""
        import uuid as _u
        legacy = self._Row(_u.uuid4(), "repo-wp-0001", "workplans/REPO-WP-0001.md")
        assert not fp._identity_is_derived(legacy)

    def test_the_two_uuids_actually_differ(self):
        """Guards the premise: if they were equal the distinction is moot."""
        assert fp.derived_record_uuid("ADHOC-2026-08-25") != fp.derived_record_uuid(
            "CUST-WP-ADHOC-2026-08-25"
        )

    @pytest.mark.parametrize(
        "slug",
        [
            "adhoc-2026-07-02",                # grandfathered daily ad-hoc
            "cust-wp-adhoc-2026-08-25",        # qualified daily ad-hoc
            "net-kingdom-adhoc-2026-07-02",    # repo-name-qualified legacy form
            "adhoc-llmc-2026-06-02",           # another legacy qualification
            "rcluster-wp-0007",                # ordinary workplan
            "cust-wp-0066-t01",                # task
        ],
    )
    def test_identifier_slugs_are_recognised(self, slug):
        """A slug that claims to be an identifier is one, whatever its UUID.

        25 of 44 ad-hoc rows are legacy v4, so a UUID-version test alone would
        leave them path-matched and silently diverged from their files.
        """
        assert fp._slug_is_identifier(slug)

    @pytest.mark.parametrize(
        "slug",
        ["three-phoenix-ha-cluster", "testdrive-jsui-publication", "state-hub-v0.1", ""],
    )
    def test_title_slugs_keep_path_matching(self, slug):
        """Hub-first rows never claimed an identifier; the path is their only link."""
        assert not fp._slug_is_identifier(slug)


class TestRetirementReleasesTheIdentifier:
    """`slug` is unique table-wide, so retirement must free it.

    Retirement that only sets a timestamp leaves the identifier locked to a
    record nothing derives, and the repository that owns it can never claim it.
    """

    def test_the_identifier_is_released(self):
        from datetime import datetime, timezone
        when = datetime(2026, 8, 28, tzinfo=timezone.utc)
        assert fp._tombstone_slug("repo-wp-0001", when) == "repo-wp-0001@retired-20260828"

    def test_re_retiring_does_not_stack_marks(self):
        """Otherwise the 100-char column overflows after a few passes."""
        from datetime import datetime, timezone
        a = fp._tombstone_slug("repo-wp-0001", datetime(2026, 8, 28, tzinfo=timezone.utc))
        b = fp._tombstone_slug(a, datetime(2026, 9, 1, tzinfo=timezone.utc))
        assert b == "repo-wp-0001@retired-20260901"
        assert b.count(fp.RETIRED_SLUG_MARK) == 1

    def test_a_long_slug_stays_within_the_column(self):
        from datetime import datetime, timezone
        out = fp._tombstone_slug("x" * 140, datetime(2026, 8, 28, tzinfo=timezone.utc))
        assert len(out) <= 100

    class _Retired:
        def __init__(self, slug, when, id=None):
            import uuid as _u
            self.id = id or _u.uuid4()
            self.slug = slug
            self.projection_retired_at = when
            self.projection_retired_reason = "x"
            self.backing_relative_path = None
            self.status = "finished"

    def test_rows_retired_before_the_stamp_existed_are_repaired(self):
        """`stale` excludes already-retired rows, so nothing else revisits them.

        The 32 rows retired earlier today still hold their identifiers; without
        this they would block the owning repository forever.
        """
        from datetime import datetime, timezone
        when = datetime(2026, 8, 28, tzinfo=timezone.utc)
        row = self._Retired("repo-wp-0001", when)
        assert fp.RETIRED_SLUG_MARK not in row.slug
        row.slug = fp._tombstone_slug(row.slug, row.projection_retired_at)
        assert row.slug == "repo-wp-0001@retired-20260828"

    def test_an_already_stamped_row_is_left_alone(self):
        """Otherwise every reset rewrites the date and churns the row."""
        from datetime import datetime, timezone
        row = self._Retired(
            "repo-wp-0001@retired-20260828", datetime(2026, 8, 28, tzinfo=timezone.utc)
        )
        assert fp.RETIRED_SLUG_MARK in row.slug

    def test_a_release_is_committed_even_when_the_repo_refuses(self):
        """The fleet driver commits only on `applied` and rolls back otherwise.

        Moving the release ahead of the refusal returns achieves nothing if the
        rollback then discards it — which is the state railiance-cluster and
        inter-hub were in.
        """
        import inspect
        src = inspect.getsource(fp.reset_fleet_projection)
        assert 'result.status == "applied" or result.released' in src


class TestUuidMatchWins:
    """A row whose UUID is uuid5 of the identifier IS that record.

    `testdrive-jsui-publication` and `three-phoenix-ha-cluster` carry legacy
    title slugs while their UUIDs are derived from MARKITECT-WP-0002 and
    RCLUSTER-WP-0007. The slug-identifier rule alone reads them as re-keys and
    proposes retiring records that were correct all along.
    """

    def test_a_title_slugged_row_is_still_its_derived_record(self):
        import uuid as _u
        rid = "MARKITECT-WP-0002"
        row_id = _u.UUID(fp.derived_record_uuid(rid))
        # The slug says one thing, the UUID says another; the UUID is identity.
        assert not fp._slug_is_identifier("testdrive-jsui-publication")
        assert str(row_id) == fp.derived_record_uuid(rid)

    def test_uuid_match_is_checked_before_the_rekey_rules(self):
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        uuid_at = src.index("want_by_uuid.get(str(row.id))")
        rekey_at = src.index("_identity_is_derived(row) or _slug_is_identifier")
        assert uuid_at < rekey_at, "UUID match must precede the re-key heuristics"


class TestTaskIdentityIsQualified:
    """A bare `T01` is unique within its workplan, not in the fleet.

    Unqualified, `uuid5("T01")` is the same UUID for every workplan that has a
    T01 — llm-connect's 91 task blocks derive 49 distinct UUIDs, and creating
    its workplans dies on `duplicate key value violates unique constraint
    "tasks_pkey"`.
    """

    BODY = """
```task
id: T01
status: done
```

```task
id: T02
status: todo
```
"""

    def test_short_ids_are_qualified_by_their_workplan(self):
        a = fp._parse_tasks(self.BODY, "LLM-WP-0001")
        b = fp._parse_tasks(self.BODY, "LLM-WP-0002")
        assert [t.record_id for t in a] == ["LLM-WP-0001-T01", "LLM-WP-0001-T02"]
        assert {t.uuid for t in a}.isdisjoint({t.uuid for t in b})

    def test_an_already_qualified_id_is_left_alone(self):
        body = "```task\nid: LLM-WP-0009-T03\nstatus: todo\n```"
        assert fp._parse_tasks(body, "LLM-WP-0001")[0].record_id == "LLM-WP-0009-T03"

    def test_it_agrees_with_the_backfill(self):
        """Both must apply the same rule or they disagree on a task's name."""
        from api.services.task_record_id_backfill import qualify_task_id
        got = fp._parse_tasks(self.BODY, "LLM-WP-0001")[0].record_id
        assert got == qualify_task_id("T01", "LLM-WP-0001")

    def test_uuid_follows_the_qualified_id(self):
        t = fp._parse_tasks(self.BODY, "LLM-WP-0001")[0]
        assert t.uuid == fp.derived_record_uuid("LLM-WP-0001-T01")


class TestRetirementIsNotUndoneByPath:
    """A retired row is matched by UUID or not at all.

    Releasing the identifier removes the evidence that made the row a re-key:
    the tombstoned slug is no longer an identifier, and a legacy row is not
    UUID-derived, so both guards fall through to path matching — which matches
    the very file the row was retired for and resurrects it beside the correct
    record created from that same file.
    """

    def test_a_tombstoned_slug_is_not_an_identifier(self):
        """This is the signal the tombstone destroys."""
        assert fp._slug_is_identifier("adhoc-2026-06-01")
        assert not fp._slug_is_identifier("adhoc-2026-06-01@retired-20260827")

    def test_retired_rows_are_skipped_before_path_matching(self):
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        skip_at = src.index("if row.projection_retired_at is not None:")
        path_at = src.index("if bp and _path_key(bp) in want_paths:")
        assert skip_at < path_at, "retired rows must be skipped before path matching"

    def test_uuid_match_still_precedes_the_skip(self):
        """A record that genuinely returns must still be un-retired."""
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        assert src.index("want_by_uuid.get(str(row.id))") < src.index(
            "if row.projection_retired_at is not None:"
        )

    def test_a_row_that_derives_again_is_not_tombstoned(self):
        """Releasing it would change the slug out from under the matching.

        The row would then fail to match its own file and never be un-retired —
        which is how five rows ended up tombstoned but not retired.
        """
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        rel = src.index("outcome.released.append(before)")
        guard = src.index('in want or str(r.id) in want_by_uuid')
        assert guard < rel, "the release must exempt records the forge still derives"

    def test_a_stamped_slug_counts_as_retired_even_if_the_flag_is_clear(self):
        """Otherwise a resurrected row path-matches its file and reports noop forever."""
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        assert "projection_retired_at is None and RETIRED_SLUG_MARK in" in src

    def test_a_resurrected_row_becomes_stale_rather_than_being_skipped(self):
        """Skipping it outright is what left six rows reporting noop forever.

        `stale` is computed from `matched`, so a row that never enters `matched`
        is invisible: not retired, so not stale; path-matching its own file, so
        never reported.
        """
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        zombie = src.index("projection_retired_at is None and RETIRED_SLUG_MARK in")
        tail = src[zombie:zombie + 700]
        assert "matched[key] = row" in tail, "a resurrected row must enter `matched`"


class TestCompetingClaims:
    """Two rows can claim one record; the weaker must not vanish.

    `cust-wp-0010` claimed CUST-WP-0010 by its own identifier while
    `workstream-lifecycle-documentation` claimed it by path. 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.
    """

    def test_claims_are_ranked_not_overwritten(self):
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        assert "def _claim(" in src
        assert "displaced.append" in src

    def test_a_displaced_row_becomes_stale(self):
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        stale_at = src.index("stale = [")
        assert "displaced" in src[stale_at:stale_at + 400]

    def test_identifier_beats_path(self):
        """A row naming the record outranks one that merely shares its file."""
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        assert src.index("UUID_MATCH, SLUG_MATCH, PATH_MATCH, PREFIX_MATCH = 0, 1, 2, 3") > 0
        # Strength is ordered strongest-first, so a lower number wins.
        assert "if strength < held:" in src


class TestTitleIsDerivedToo:
    """The title is a derived field and must follow the file.

    Not syncing it left `cust-wp-0010` correctly identified and correctly backed
    while describing a different workplan's work.
    """

    def test_title_is_synced(self):
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        assert "row.title = w.title.strip()" in src

    def test_an_empty_derived_title_does_not_blank_a_real_one(self):
        """Three activity-core files parse to no title; blanking is worse than stale."""
        import inspect
        src = inspect.getsource(fp.reset_repository_projection)
        i = src.index("row.title = w.title.strip()")
        guard = src[max(0, i - 200):i]
        assert "w.title and w.title.strip()" in guard


class TestTitleSyncBehaviour:
    """Exercised against the fake session rather than asserted on source."""

    @pytest.mark.asyncio
    async def test_a_drifted_title_is_corrected(self):
        row = _Row(slug="demo-wp-0001", status="active",
                   path="workplans/a.md", title="Some Other Workplan")
        session = _FakeSession(repo=_Repo(), rows=[row])
        out = await fp.reset_repository_projection(
            session, "demo",
            derived=TestReset()._derived(("DEMO-WP-0001", "active", "workplans/a.md")),
        )
        assert out.status == "applied"
        assert row.title != "Some Other Workplan"

    @pytest.mark.asyncio
    async def test_an_empty_derived_title_leaves_the_existing_one(self):
        row = _Row(slug="demo-wp-0001", status="active",
                   path="workplans/a.md", title="Real Title")
        d = TestReset()._derived(("DEMO-WP-0001", "active", "workplans/a.md"))
        d.workplans[0].title = None
        session = _FakeSession(repo=_Repo(), rows=[row])
        await fp.reset_repository_projection(session, "demo", derived=d)
        assert row.title == "Real Title"


class _TaskRow:
    def __init__(self, record_id, status="todo", title="t", workplan_id=None):
        import uuid as _u
        self.id = _u.uuid4()
        self.record_id = record_id
        self.status = status
        self.title = title
        self.priority = "medium"
        self.workplan_id = workplan_id


class TestExistingWorkplanTasks:
    """Tasks of an existing workplan are matched by record_id (CUST-WP-0068-T09)."""

    def _derived(self, *tasks):
        dts = [
            fp.DerivedTask(
                record_id=tid,
                uuid=fp.derived_record_uuid(tid),
                title=title,
                status=st,
                priority="medium",
            )
            for tid, title, st in tasks
        ]
        return fp.DerivedProjection(
            repo_slug="demo",
            commit="c0ffee",
            workplans=[
                fp.DerivedWorkplan(
                    record_id="DEMO-WP-0001",
                    uuid=fp.derived_record_uuid("DEMO-WP-0001"),
                    title="DEMO-WP-0001",
                    status="active",
                    relative_path="workplans/a.md",
                    archived=False,
                    tasks=dts,
                )
            ],
        )

    @pytest.mark.asyncio
    async def test_creates_a_missing_identified_task(self):
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        session = _FakeSession(repo=_Repo(), rows=[row], task_rows=[])
        out = await fp.reset_repository_projection(
            session, "demo",
            derived=self._derived(("DEMO-WP-0001-T01", "Do it", "todo")),
        )
        assert out.created_tasks == ["DEMO-WP-0001-T01"]
        assert session.added
        assert session.added[0].record_id == "DEMO-WP-0001-T01"

    @pytest.mark.asyncio
    async def test_cancels_an_open_task_the_file_no_longer_derives(self):
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        stale = _TaskRow("DEMO-WP-0001-T09", status="todo", workplan_id=row.id)
        session = _FakeSession(repo=_Repo(), rows=[row], task_rows=[stale])
        out = await fp.reset_repository_projection(
            session, "demo",
            derived=self._derived(("DEMO-WP-0001-T01", "Do it", "todo")),
        )
        assert stale.status.value == "cancel" or stale.status == "cancel" or str(stale.status).endswith("cancel")
        assert "DEMO-WP-0001-T09" in out.cancelled_tasks
        assert session.deleted == []

    @pytest.mark.asyncio
    async def test_sets_record_id_when_the_derived_uuid_already_exists(self):
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        existing = _TaskRow(None, status="done", title="Clear the stale hub rows", workplan_id=row.id)
        derived = self._derived(("DEMO-WP-0001-T01", "Clear the stale hub rows", "done"))
        existing.id = __import__("uuid").UUID(derived.workplans[0].tasks[0].uuid)
        session = _FakeSession(repo=_Repo(), rows=[row], task_rows=[existing])
        out = await fp.reset_repository_projection(session, "demo", derived=derived)
        assert out.created_tasks == []
        assert existing.record_id == "DEMO-WP-0001-T01"
        assert session.added == []

    @pytest.mark.asyncio
    async def test_leaves_unidentified_tasks_alone(self):
        row = _Row(slug="demo-wp-0001", status="active", path="workplans/a.md")
        orphan = _TaskRow(None, status="todo", title="legacy", workplan_id=row.id)
        session = _FakeSession(repo=_Repo(), rows=[row], task_rows=[orphan])
        await fp.reset_repository_projection(
            session, "demo",
            derived=self._derived(("DEMO-WP-0001-T01", "Do it", "todo")),
        )
        assert orphan.status == "todo"