Files
rfc-app/docs/superpowers/plans/2026-06-03-m3-backend-plan-a.md
T
Ben Stull dd72f913a3 docs: M3-backend Plan A implementation plan + spec refinements
Adds the bite-sized TDD implementation plan for Plan A (registry mirror +
runtime-config APIs + initial_state/unreviewed semantics; additive only).
Refines the spec with two planning discoveries: the re-stamp must be a
Python startup step (pure SQL can't read DEFAULT_PROJECT_ID), and the
PK-rebuild blast radius justifies splitting M3-backend into Plan A (this)
and Plan B (rebuild + project_id threading + re-stamp, before M4).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-03 22:23:55 -07:00

64 KiB

M3-backend Plan A — registry mirror + runtime config + review semantics

For agentic workers: REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (- [ ]) syntax for tracking.

Goal: Make the framework learn its projects from a git registry (REGISTRY_REPO), expose deployment + project config at runtime (GET /api/deployment, GET /api/projects/:id), and carry the initial_state/unreviewed review semantics — all additively, operating on the single default-id project (no PK rebuild, no re-stamp, no writer threading; those are Plan B).

Architecture: A self-contained app/registry.py mirrors REGISTRY_REPO/projects.yaml into the existing projects table + a new deployment singleton, driven by the existing webhook dispatcher and Reconciler.sweep() (Option A from the spec). META_REPO is retired (hard cut): the corpus mirror reads the default project's content_repo from the DB instead. Migration 027 is additive only. See spec: docs/superpowers/specs/2026-06-03-m3-backend-design.md.

Tech Stack: Python 3 / FastAPI / SQLite (raw sqlite3), pyyaml, pytest + httpx.MockTransport (the FakeGitea harness). Tests run from backend/ with python -m pytest.


Conventions for every task

  • Work on branch m3-backend-registry-spine (already checked out).
  • Run tests from the backend/ directory: cd backend && python -m pytest.
  • The shared integration fixtures tmp_env, app_with_fake_gitea, FakeGitea, sign_in_as, provision_user_row live in backend/tests/test_propose_vertical.py and are imported by other test files (e.g. from test_propose_vertical import app_with_fake_gitea, tmp_env # noqa: F401).
  • After Task 5, the whole suite must stay green; run python -m pytest (full suite) at the end of every subsequent task.

Task 1: Entry review fields (unreviewed / reviewed_at / reviewed_by)

Files:

  • Modify: backend/app/entry.py (dataclass ~29-53, parse ~56-85, serialize ~88-117)

  • Test: backend/tests/test_entry_review_fields.py

  • Step 1: Write the failing test

# backend/tests/test_entry_review_fields.py
"""§22.4c — the unreviewed/reviewed_at/reviewed_by entry frontmatter fields."""
from __future__ import annotations

from app import entry as entry_mod


def test_parse_defaults_unreviewed_false_when_absent():
    text = "---\nslug: ohm\ntitle: OHM\nstate: active\n---\n\nBody.\n"
    e = entry_mod.parse(text)
    assert e.unreviewed is False
    assert e.reviewed_at is None
    assert e.reviewed_by is None


def test_parse_reads_review_fields():
    text = (
        "---\nslug: ohm\ntitle: OHM\nstate: active\n"
        "unreviewed: true\nreviewed_at: '2026-06-03'\nreviewed_by: ben\n---\n\nBody.\n"
    )
    e = entry_mod.parse(text)
    assert e.unreviewed is True
    assert e.reviewed_at == "2026-06-03"
    assert e.reviewed_by == "ben"


def test_serialize_emits_review_fields_only_when_meaningful():
    # unreviewed False + no provenance → keys omitted (keeps document entries clean).
    e = entry_mod.Entry(slug="a", title="A", state="super-draft")
    assert "unreviewed" not in entry_mod.serialize(e)
    assert "reviewed_at" not in entry_mod.serialize(e)
    # unreviewed True → emitted.
    e2 = entry_mod.Entry(slug="b", title="B", state="active", unreviewed=True)
    assert "unreviewed: true" in entry_mod.serialize(e2)


def test_round_trip_preserves_review_fields():
    e = entry_mod.Entry(
        slug="b", title="B", state="active",
        unreviewed=False, reviewed_at="2026-06-03", reviewed_by="ben",
    )
    back = entry_mod.parse(entry_mod.serialize(e))
    assert back.reviewed_at == "2026-06-03"
    assert back.reviewed_by == "ben"
    assert back.unreviewed is False
  • Step 2: Run test to verify it fails

Run: cd backend && python -m pytest tests/test_entry_review_fields.py -v Expected: FAIL — Entry.__init__() got an unexpected keyword argument 'unreviewed'.

  • Step 3: Add the dataclass fields

In backend/app/entry.py, add to the Entry dataclass (after funder: str | None = None, before body: str = ""):

    # §22.4c: an `active` entry that landed without a human review gate
    # carries unreviewed=True until an owner clears it. Orthogonal to
    # `state`; only meaningful for active entries. reviewed_at/reviewed_by
    # are the provenance of the clear, paralleling graduated_at/by.
    unreviewed: bool = False
    reviewed_at: str | None = None
    reviewed_by: str | None = None
  • Step 4: Parse the fields

In parse(), add before the return Entry(:

    unreviewed = bool(fm.get("unreviewed") or False)

and add these keyword args to the Entry(...) constructor call:

        unreviewed=unreviewed,
        reviewed_at=fm.get("reviewed_at") or None,
        reviewed_by=fm.get("reviewed_by") or None,
  • Step 5: Serialize the fields

In serialize(), after the funder block (if entry.funder:), add:

    # §22.4c: emit unreviewed only when True (a super-draft / reviewed
    # active entry leaves the key absent → frontmatter stays minimal).
    if entry.unreviewed:
        fm["unreviewed"] = True
    if entry.reviewed_at:
        fm["reviewed_at"] = entry.reviewed_at
    if entry.reviewed_by:
        fm["reviewed_by"] = entry.reviewed_by
  • Step 6: Run tests to verify they pass

Run: cd backend && python -m pytest tests/test_entry_review_fields.py -v Expected: PASS (4 tests).

  • Step 7: Commit
git add backend/app/entry.py backend/tests/test_entry_review_fields.py
git commit -m "feat(entry): §22.4c unreviewed/reviewed_at/reviewed_by frontmatter fields"

Task 2: Migration 027 — additive registry/review schema

Files:

  • Create: backend/migrations/027_projects_runtime_config.sql

  • Test: backend/tests/test_migration_027.py

  • Step 1: Write the failing test

# backend/tests/test_migration_027.py
"""Migration 027 — additive §22 M3 schema (projects.type/initial_state, the
deployment singleton, cached_rfcs review columns). No table rebuilds in Plan A."""
from __future__ import annotations

import tempfile
from pathlib import Path

from app import db
from app.config import Config


def _fresh_config() -> Config:
    tmp = Path(tempfile.mkdtemp(prefix="mig027-")) / "t.db"
    return Config(
        gitea_url="x", gitea_bot_user="x", gitea_bot_token="x", gitea_org="x",
        meta_repo="meta", registry_repo="registry",
        oauth_client_id="x", oauth_client_secret="x", app_url="x",
        secret_key="x", database_path=tmp, owner_gitea_login="x",
        webhook_secret="x",
    )


def test_027_adds_project_type_and_initial_state():
    cfg = _fresh_config()
    db.run_migrations(cfg)
    conn = db.connect(cfg.database_path)
    cols = {r["name"]: r for r in conn.execute("PRAGMA table_info(projects)")}
    assert "type" in cols and cols["type"]["dflt_value"] == "'document'"
    assert "initial_state" in cols and cols["initial_state"]["dflt_value"] == "'super-draft'"
    conn.close()


def test_027_creates_deployment_singleton():
    cfg = _fresh_config()
    db.run_migrations(cfg)
    conn = db.connect(cfg.database_path)
    rows = list(conn.execute("SELECT id FROM deployment"))
    assert [r["id"] for r in rows] == [1]
    # The CHECK (id = 1) forbids a second row.
    try:
        conn.execute("INSERT INTO deployment (id) VALUES (2)")
        raised = False
    except Exception:
        raised = True
    assert raised
    conn.close()


def test_027_adds_review_columns_to_cached_rfcs():
    cfg = _fresh_config()
    db.run_migrations(cfg)
    conn = db.connect(cfg.database_path)
    cols = {r["name"] for r in conn.execute("PRAGMA table_info(cached_rfcs)")}
    assert {"unreviewed", "reviewed_at", "reviewed_by"} <= cols
    conn.close()
  • Step 2: Run test to verify it fails

Run: cd backend && python -m pytest tests/test_migration_027.py -v Expected: FAIL — type not in projects columns (migration doesn't exist yet).

  • Step 3: Write the migration
-- backend/migrations/027_projects_runtime_config.sql
-- §22 M3 (Plan A) — additive registry/runtime-config schema.
--
-- This is the additive half of M3's backend. It adds the project `type` and
-- `initial_state` columns (mirrored from the registry), the deployment
-- singleton (deployment name/tagline mirrored from the registry), and the
-- §22.4c review columns on cached_rfcs. NO table rebuilds: the §22.13 PK
-- rebuilds and the default->slug re-stamp ride a later migration (Plan B),
-- just before a second project can collide (see migration 026's DEFERRED
-- block and docs/superpowers/specs/2026-06-03-m3-backend-design.md §1/§6).

ALTER TABLE projects ADD COLUMN type TEXT NOT NULL DEFAULT 'document'
    CHECK (type IN ('document', 'specification', 'bdd'));
ALTER TABLE projects ADD COLUMN initial_state TEXT NOT NULL DEFAULT 'super-draft'
    CHECK (initial_state IN ('super-draft', 'active'));

-- Deployment-level identity (name, tagline) mirrored from the registry's
-- `deployment:` block. A singleton: the CHECK pins it to one row.
CREATE TABLE IF NOT EXISTS deployment (
    id           INTEGER PRIMARY KEY CHECK (id = 1),
    name         TEXT,
    tagline      TEXT,
    registry_sha TEXT,
    updated_at   TEXT NOT NULL DEFAULT (datetime('now'))
);
INSERT OR IGNORE INTO deployment (id) VALUES (1);

-- §22.4c review flag + provenance. unreviewed is git-truth (mirrored from
-- entry frontmatter); it survives a cache rebuild like `state` does.
ALTER TABLE cached_rfcs ADD COLUMN unreviewed INTEGER NOT NULL DEFAULT 0;
ALTER TABLE cached_rfcs ADD COLUMN reviewed_at TEXT;
ALTER TABLE cached_rfcs ADD COLUMN reviewed_by TEXT;
  • Step 4: Run tests to verify they pass

Run: cd backend && python -m pytest tests/test_migration_027.py -v Expected: PASS (3 tests).

  • Step 5: Run the full suite (no regressions from the additive migration)

Run: cd backend && python -m pytest -q Expected: PASS (the additive columns don't disturb existing queries).

  • Step 6: Commit
git add backend/migrations/027_projects_runtime_config.sql backend/tests/test_migration_027.py
git commit -m "feat(db): migration 027 — additive §22 M3 schema (type/initial_state, deployment, review cols)"

Task 3: Cache mirrors the review fields

Files:

  • Modify: backend/app/cache.py (_upsert_cached_rfc ~78-129)

  • Test: backend/tests/test_cache_review_fields.py

  • Step 1: Write the failing test

# backend/tests/test_cache_review_fields.py
"""§22.4c — _upsert_cached_rfc mirrors the review fields into cached_rfcs."""
from __future__ import annotations

from fastapi.testclient import TestClient

from test_propose_vertical import app_with_fake_gitea, tmp_env  # noqa: F401


def test_upsert_writes_review_fields(app_with_fake_gitea):
    from app import cache, db, entry as entry_mod

    app, _ = app_with_fake_gitea
    with TestClient(app):
        e = entry_mod.Entry(
            slug="rev", title="Rev", state="active",
            unreviewed=True, reviewed_at="2026-06-03", reviewed_by="ben",
        )
        cache._upsert_cached_rfc(e, body_sha="sha-rev")
        row = db.conn().execute(
            "SELECT unreviewed, reviewed_at, reviewed_by FROM cached_rfcs WHERE slug = 'rev'"
        ).fetchone()
        assert row["unreviewed"] == 1
        assert row["reviewed_at"] == "2026-06-03"
        assert row["reviewed_by"] == "ben"
  • Step 2: Run test to verify it fails

Run: cd backend && python -m pytest tests/test_cache_review_fields.py -v Expected: FAIL — the upsert doesn't write unreviewed (column stays default 0; reviewed_at stays NULL).

  • Step 3: Extend _upsert_cached_rfc

In backend/app/cache.py, change the INSERT column list, the VALUES placeholders, the ON CONFLICT(slug) DO UPDATE SET clause, and the params tuple to include the three review fields. The edited statement:

    db.conn().execute(
        """
        INSERT INTO cached_rfcs
            (slug, title, state, rfc_id, repo, proposed_by, proposed_at,
             graduated_at, graduated_by, owners_json, arbiters_json, tags_json,
             models_json, funder_login, body, body_sha,
             unreviewed, reviewed_at, reviewed_by,
             last_entry_commit_at, updated_at)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'))
        ON CONFLICT(slug) DO UPDATE SET
            title = excluded.title,
            state = excluded.state,
            rfc_id = excluded.rfc_id,
            repo = excluded.repo,
            proposed_by = excluded.proposed_by,
            proposed_at = excluded.proposed_at,
            graduated_at = excluded.graduated_at,
            graduated_by = excluded.graduated_by,
            owners_json = excluded.owners_json,
            arbiters_json = excluded.arbiters_json,
            tags_json = excluded.tags_json,
            models_json = excluded.models_json,
            funder_login = excluded.funder_login,
            body = excluded.body,
            body_sha = excluded.body_sha,
            unreviewed = excluded.unreviewed,
            reviewed_at = excluded.reviewed_at,
            reviewed_by = excluded.reviewed_by,
            last_entry_commit_at = datetime('now'),
            updated_at = datetime('now')
        """,
        (
            entry.slug,
            entry.title,
            entry.state,
            entry.id,
            entry.repo,
            entry.proposed_by,
            entry.proposed_at,
            entry.graduated_at,
            entry.graduated_by,
            json.dumps(entry.owners),
            json.dumps(entry.arbiters),
            json.dumps(entry.tags),
            models_json,
            funder_login,
            entry.body,
            body_sha,
            1 if entry.unreviewed else 0,
            entry.reviewed_at,
            entry.reviewed_by,
        ),
    )
  • Step 4: Run tests to verify they pass

Run: cd backend && python -m pytest tests/test_cache_review_fields.py -q Expected: PASS.

  • Step 5: Run the full suite

Run: cd backend && python -m pytest -q Expected: PASS.

  • Step 6: Commit
git add backend/app/cache.py backend/tests/test_cache_review_fields.py
git commit -m "feat(cache): mirror §22.4c review fields into cached_rfcs"

Task 4: app/registry.py — the registry mirror module

Files:

  • Create: backend/app/registry.py
  • Test: backend/tests/test_registry.py

The module factors into pure functions (parse_registry, apply_registry) that are unit-testable without I/O, plus refresh_registry (the I/O orchestrator, integration-tested in Task 5).

  • Step 1: Write the failing test
# backend/tests/test_registry.py
"""§22.2 registry parse + apply: validation, type-immutability, upsert."""
from __future__ import annotations

import tempfile
from pathlib import Path

import pytest

from app import db, registry
from app.config import Config


def _db():
    cfg = Config(
        gitea_url="x", gitea_bot_user="x", gitea_bot_token="x", gitea_org="x",
        meta_repo="meta", registry_repo="registry", oauth_client_id="x",
        oauth_client_secret="x", app_url="x", secret_key="x",
        database_path=Path(tempfile.mkdtemp(prefix="reg-")) / "t.db",
        owner_gitea_login="x", webhook_secret="x",
    )
    db.run_migrations(cfg)
    if db._CONN is not None:
        db._CONN.close()
        db._CONN = None
    db.init(cfg)
    return cfg


VALID = """
deployment:
  name: Open Human Model
  tagline: A model of human flourishing
projects:
  - id: default
    name: Open Human Model
    type: document
    content_repo: meta
    visibility: public
"""


def test_parse_valid_registry():
    doc = registry.parse_registry(VALID)
    assert doc.deployment_name == "Open Human Model"
    assert doc.deployment_tagline == "A model of human flourishing"
    assert len(doc.projects) == 1
    p = doc.projects[0]
    assert (p.id, p.type, p.content_repo, p.visibility) == ("default", "document", "meta", "public")
    # initial_state defaults from type when omitted.
    assert p.initial_state == "super-draft"


def test_parse_initial_state_defaults_per_type():
    doc = registry.parse_registry(
        "projects:\n  - {id: a, name: A, type: bdd, content_repo: a}\n"
    )
    assert doc.projects[0].initial_state == "active"  # bdd default


@pytest.mark.parametrize("bad,msg", [
    ("projects: []\n", "at least one"),
    ("projects:\n  - {id: 'Bad Slug', name: A, type: document, content_repo: a}\n", "valid slug"),
    ("projects:\n  - {id: a, name: A, type: nope, content_repo: a}\n", "invalid type"),
    ("projects:\n  - {id: a, name: A, type: document}\n", "content_repo"),
    ("projects:\n  - {id: a, name: A, type: document, content_repo: a, visibility: x}\n", "visibility"),
    ("projects:\n  - {id: a, name: A, type: document, content_repo: a}\n  - {id: a, name: B, type: document, content_repo: b}\n", "duplicate"),
])
def test_parse_rejects_invalid(bad, msg):
    with pytest.raises(registry.RegistryError) as e:
        registry.parse_registry(bad)
    assert msg in str(e.value)


def test_apply_upserts_projects_and_deployment():
    _db()
    doc = registry.parse_registry(VALID)
    registry.apply_registry(doc, registry_sha="regsha1")
    prow = db.conn().execute(
        "SELECT name, type, content_repo, visibility, initial_state, registry_sha FROM projects WHERE id='default'"
    ).fetchone()
    assert prow["name"] == "Open Human Model"
    assert prow["content_repo"] == "meta"
    assert prow["registry_sha"] == "regsha1"
    drow = db.conn().execute("SELECT name, tagline FROM deployment WHERE id=1").fetchone()
    assert drow["name"] == "Open Human Model"
    assert drow["tagline"] == "A model of human flourishing"


def test_apply_rejects_type_change_on_existing_project():
    _db()
    registry.apply_registry(registry.parse_registry(VALID), "s1")
    changed = VALID.replace("type: document", "type: specification")
    registry.apply_registry(registry.parse_registry(changed), "s2")  # logged + skipped, no raise
    t = db.conn().execute("SELECT type FROM projects WHERE id='default'").fetchone()["type"]
    assert t == "document"  # immutable — unchanged
  • Step 2: Run test to verify it fails

Run: cd backend && python -m pytest tests/test_registry.py -v Expected: FAIL — ModuleNotFoundError: No module named 'app.registry' (the M1 projects.py exists, but registry.py does not).

  • Step 3: Write app/registry.py
# backend/app/registry.py
"""§22.2 project registry mirror — the config-side analogue of
cache.refresh_meta_repo.

A deployment declares its projects in a `projects.yaml` at the root of
REGISTRY_REPO. This module mirrors that file into the `projects` cache table
and the `deployment` singleton. Per §22.2, `projects` rows flow from the
registry only — never from user actions. The mirror runs on the registry-repo
webhook and on every reconciler sweep (Option A wiring, Task 5).
"""
from __future__ import annotations

import base64
import json
import logging
import re
from dataclasses import dataclass, field

import yaml

from . import db
from .config import Config
from .gitea import Gitea

log = logging.getLogger(__name__)

VALID_TYPES = {"document", "specification", "bdd"}
VALID_VISIBILITY = {"gated", "public", "unlisted"}
VALID_INITIAL_STATE = {"super-draft", "active"}
# §22.4b: per-type default landing state.
_TYPE_DEFAULT_INITIAL_STATE = {
    "document": "super-draft",
    "specification": "super-draft",
    "bdd": "active",
}
_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]*$")


class RegistryError(Exception):
    """A registry document that fails validation, or a missing registry file.

    Raised by parse_registry/refresh_registry. The caller decides severity:
    fatal at startup (no last-good to serve), tolerated on a running deployment
    (keep the last-good projects rows). See Task 5 wiring.
    """


@dataclass
class ProjectEntry:
    id: str
    name: str
    type: str
    content_repo: str
    visibility: str
    initial_state: str
    config: dict = field(default_factory=dict)  # theme, enabled_models


@dataclass
class RegistryDoc:
    deployment_name: str
    deployment_tagline: str
    projects: list[ProjectEntry]


def parse_registry(text: str) -> RegistryDoc:
    """Parse + validate projects.yaml. Pure (no I/O). Raises RegistryError."""
    raw = yaml.safe_load(text) or {}
    dep = raw.get("deployment") or {}
    projects_raw = raw.get("projects") or []
    if not isinstance(projects_raw, list) or not projects_raw:
        raise RegistryError("registry must declare at least one project")
    seen: set[str] = set()
    entries: list[ProjectEntry] = []
    for p in projects_raw:
        pid = str(p.get("id") or "").strip()
        if not _SLUG_RE.match(pid):
            raise RegistryError(f"project id {pid!r} is not a valid slug")
        if pid in seen:
            raise RegistryError(f"duplicate project id {pid!r}")
        seen.add(pid)
        name = str(p.get("name") or "").strip()
        if not name:
            raise RegistryError(f"project {pid!r} missing name")
        ptype = str(p.get("type") or "").strip()
        if ptype not in VALID_TYPES:
            raise RegistryError(f"project {pid!r} has invalid type {ptype!r}")
        content_repo = str(p.get("content_repo") or "").strip()
        if not content_repo:
            raise RegistryError(f"project {pid!r} missing content_repo")
        vis = str(p.get("visibility") or "gated").strip()
        if vis not in VALID_VISIBILITY:
            raise RegistryError(f"project {pid!r} has invalid visibility {vis!r}")
        initial_state = str(
            p.get("initial_state") or _TYPE_DEFAULT_INITIAL_STATE[ptype]
        ).strip()
        if initial_state not in VALID_INITIAL_STATE:
            raise RegistryError(
                f"project {pid!r} has invalid initial_state {initial_state!r}"
            )
        cfg: dict = {}
        if p.get("theme") is not None:
            cfg["theme"] = p["theme"]
        if p.get("enabled_models") is not None:
            cfg["enabled_models"] = [str(m) for m in p["enabled_models"]]
        entries.append(
            ProjectEntry(pid, name, ptype, content_repo, vis, initial_state, cfg)
        )
    return RegistryDoc(
        deployment_name=str(dep.get("name") or "").strip(),
        deployment_tagline=str(dep.get("tagline") or "").strip(),
        projects=entries,
    )


def apply_registry(doc: RegistryDoc, registry_sha: str) -> None:
    """Upsert the parsed registry into projects + deployment. Idempotent.

    §22.4a: `type` is immutable — a change against an existing row is rejected
    (skip + log), never applied. Projects absent from the registry are left in
    place (archival is out of scope for M3; they simply stop refreshing).
    """
    conn = db.conn()
    for e in doc.projects:
        existing = conn.execute(
            "SELECT type FROM projects WHERE id = ?", (e.id,)
        ).fetchone()
        if existing is not None and existing["type"] not in (None, "", e.type):
            log.error(
                "registry: refusing immutable type change on project %s (%s -> %s)",
                e.id, existing["type"], e.type,
            )
            continue
        conn.execute(
            """
            INSERT INTO projects
                (id, name, type, content_repo, visibility, initial_state,
                 config_json, registry_sha, updated_at)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, datetime('now'))
            ON CONFLICT(id) DO UPDATE SET
                name = excluded.name,
                type = excluded.type,
                content_repo = excluded.content_repo,
                visibility = excluded.visibility,
                initial_state = excluded.initial_state,
                config_json = excluded.config_json,
                registry_sha = excluded.registry_sha,
                updated_at = datetime('now')
            """,
            (
                e.id, e.name, e.type, e.content_repo, e.visibility,
                e.initial_state, json.dumps(e.config), registry_sha,
            ),
        )
    conn.execute(
        """
        UPDATE deployment
           SET name = ?, tagline = ?, registry_sha = ?, updated_at = datetime('now')
         WHERE id = 1
        """,
        (doc.deployment_name, doc.deployment_tagline, registry_sha),
    )


async def refresh_registry(config: Config, gitea: Gitea) -> None:
    """Mirror REGISTRY_REPO/projects.yaml into projects + deployment.

    Idempotent. Raises RegistryError on a missing/invalid file and GiteaError
    on transport failure; the caller chooses fatal-vs-tolerated.
    """
    item = await gitea.get_contents(
        config.gitea_org, config.registry_repo, "projects.yaml", ref="main"
    )
    if not item or item.get("type") != "file":
        raise RegistryError(
            f"{config.gitea_org}/{config.registry_repo}/projects.yaml not found"
        )
    text = base64.b64decode(item["content"]).decode("utf-8")
    sha = item.get("last_commit_sha") or item.get("sha") or ""
    doc = parse_registry(text)
    apply_registry(doc, sha)
    log.info("registry: mirrored %d project(s) at %s", len(doc.projects), sha)
  • Step 4: Run tests to verify they pass

Run: cd backend && python -m pytest tests/test_registry.py -v Expected: PASS (all parametrized cases + apply tests).

  • Step 5: Commit
git add backend/app/registry.py backend/tests/test_registry.py
git commit -m "feat(registry): §22.2 projects.yaml parse/validate/apply module"

Task 5: Hard cut — retire META_REPO, wire the registry mirror

This is the integration task: it removes META_REPO, points the corpus mirror at the default project's content_repo from the DB, wires refresh_registry into startup + webhook + sweep, retires seed_default_project, and updates the shared test fixture to seed a registry (so every test's startup refresh_registry succeeds). It must land atomically — the suite is red between the config change and the fixture update.

Files:

  • Modify: backend/app/config.py

  • Modify: backend/app/projects.py

  • Modify: backend/app/cache.py (refresh_meta_repo/refresh_meta_branches/refresh_meta_pulls, Reconciler.sweep)

  • Modify: backend/app/webhooks.py

  • Modify: backend/app/main.py (lifespan)

  • Modify: backend/tests/test_propose_vertical.py (tmp_env, FakeGitea.__init__)

  • Modify: backend/tests/test_multi_project_spine_vertical.py (stale comment in one assertion)

  • Test: backend/tests/test_registry_wiring.py

  • Step 1: Write the failing test

# backend/tests/test_registry_wiring.py
"""Startup mirrors the registry; the registry webhook re-mirrors it."""
from __future__ import annotations

from fastapi.testclient import TestClient

from test_propose_vertical import app_with_fake_gitea, tmp_env  # noqa: F401


def test_startup_mirrors_registry_into_projects_and_deployment(app_with_fake_gitea):
    from app import db

    app, _ = app_with_fake_gitea
    with TestClient(app):
        # The fixture's projects.yaml declares the default project + a name.
        prow = db.conn().execute(
            "SELECT content_repo, type, initial_state FROM projects WHERE id='default'"
        ).fetchone()
        assert prow["content_repo"] == "meta"   # from the registry, not META_REPO
        assert prow["type"] == "document"
        drow = db.conn().execute("SELECT name FROM deployment WHERE id=1").fetchone()
        assert drow["name"]  # deployment name mirrored from the registry


def test_registry_webhook_remirrors(app_with_fake_gitea):
    import hashlib
    import hmac
    import json as _json

    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        from app import db
        # Edit projects.yaml in the fake registry repo: change the tagline.
        new_yaml = (
            "deployment:\n  name: OHM\n  tagline: Edited tagline\n"
            "projects:\n  - id: default\n    name: OHM\n    type: document\n"
            "    content_repo: meta\n    visibility: public\n"
        )
        fake.files[("wiggleverse", "registry", "main", "projects.yaml")] = {
            "content": new_yaml, "sha": "regsha2",
        }
        body = _json.dumps({"repository": {"full_name": "wiggleverse/registry"}}).encode()
        secret = "test-webhook-secret-for-signature-verification"
        sig = hmac.new(secret.encode(), body, hashlib.sha256).hexdigest()
        r = client.post(
            "/api/webhooks/gitea",
            content=body,
            headers={"X-Gitea-Event": "push", "X-Gitea-Signature": sig,
                     "Content-Type": "application/json"},
        )
        assert r.status_code == 200
        tagline = db.conn().execute("SELECT tagline FROM deployment WHERE id=1").fetchone()["tagline"]
        assert tagline == "Edited tagline"
  • Step 2: Run test to verify it fails

Run: cd backend && python -m pytest tests/test_registry_wiring.py -v Expected: FAIL — startup does not mirror the registry yet (type/deployment.name empty), and the registry webhook branch doesn't exist.

  • Step 3: Config — require REGISTRY_REPO, remove META_REPO, add DEFAULT_PROJECT_ID

In backend/app/config.py:

Replace the meta_repo: str and registry_repo: str dataclass fields with:

    registry_repo: str
    default_project_id: str = ""

(Remove the meta_repo field entirely.)

Replace the meta_repo_full property with:

    @property
    def registry_repo_full(self) -> str:
        return f"{self.gitea_org}/{self.registry_repo}"

In load_config(), remove the meta_repo=_optional("META_REPO", "meta"), line and the multi-line comment above registry_repo=, and set:

        registry_repo=_required("REGISTRY_REPO"),
        default_project_id=_optional("DEFAULT_PROJECT_ID"),

NOTE: The migration-027 test (test_migration_027.py) and test_registry.py construct Config(...) directly with meta_repo="meta". Remove the meta_repo="meta", argument from those two test helpers' Config(...) calls, since the field no longer exists.

  • Step 4: projects.py — default-project helpers; retire seed_default_project

In backend/app/projects.py, delete the seed_default_project function and replace the module body below DEFAULT_PROJECT_ID = "default" with these helpers:

def resolved_default_id(config: Config) -> str:
    """The id of the deployment's bootstrap/default project. Plan A: always
    'default' (the re-stamp to a config slug rides Plan B). The config knob is
    read here so Plan B can flip the resolution without touching call sites."""
    return config.default_project_id.strip() or DEFAULT_PROJECT_ID


def default_content_repo(config: Config) -> str | None:
    """The content repo the single-corpus mirror reads, from the default
    project's row (filled by the registry mirror). Replaces the retired
    META_REPO. None until the registry mirror has run."""
    row = db.conn().execute(
        "SELECT content_repo FROM projects WHERE id = ?",
        (resolved_default_id(config),),
    ).fetchone()
    return row["content_repo"] if row and row["content_repo"] else None


def project_initial_state(project_id: str) -> str:
    """§22.4b landing state for new entries in a project. Defaults to
    'super-draft' for an unknown/unset row (the safe, today's-flow default)."""
    row = db.conn().execute(
        "SELECT initial_state FROM projects WHERE id = ?", (project_id,)
    ).fetchone()
    if row is None or not row["initial_state"]:
        return "super-draft"
    return row["initial_state"]
  • Step 5: cache.py — corpus mirror reads content_repo from the DB; sweep mirrors the registry

In backend/app/cache.py, add to the imports at the top (the from . import line):

from . import db, entry as entry_mod, projects as projects_mod, registry as registry_mod

In refresh_meta_repo, replace org, repo = config.gitea_org, config.meta_repo with:

    org = config.gitea_org
    repo = projects_mod.default_content_repo(config)
    if not repo:
        log.warning("refresh_meta_repo: default project has no content_repo yet; skipping")
        return

Apply the same two-line repo-resolution change at the top of refresh_meta_branches and refresh_meta_pulls (search for config.meta_repo in those functions and replace each occurrence; each should resolve repo = projects_mod.default_content_repo(config) and early-return if falsy). Run grep -n "config.meta_repo" backend/app/cache.py and confirm zero remain.

In Reconciler.sweep, add the registry refresh as the first step inside the try: (before refresh_meta_repo), so the projects table is current before the corpus mirror reads content_repo:

        try:
            try:
                await registry_mod.refresh_registry(self._config, self._gitea)
            except Exception:
                # Running deployment: a malformed registry PR must not take the
                # sweep (or the deployment) down — keep the last-good rows.
                log.exception("reconciler: registry refresh failed; keeping last-good projects")
            await refresh_meta_repo(self._config, self._gitea)
            ...
  • Step 6: webhooks.py — resolve repo from DB; add the registry branch

In backend/app/webhooks.py, change the import line from . import cache, db to:

from . import cache, db, projects as projects_mod, registry as registry_mod

Replace the dispatch block (the repo_full = ... through the else: slug branch) with:

        repo_full = (payload.get("repository") or {}).get("full_name") or ""
        registry_full = f"{config.gitea_org}/{config.registry_repo}"
        meta_repo = projects_mod.default_content_repo(config)
        meta_full = f"{config.gitea_org}/{meta_repo}" if meta_repo else None
        try:
            if repo_full == registry_full:
                # §22.2: a registry-repo push re-mirrors the projects table.
                try:
                    await registry_mod.refresh_registry(config, gitea)
                except registry_mod.RegistryError:
                    # Tolerate a malformed registry PR on a running deployment.
                    log.exception("registry webhook: invalid projects.yaml; keeping last-good")
            elif meta_full and (repo_full == meta_full or not repo_full):
                await cache.refresh_meta_repo(config, gitea)
                await cache.refresh_meta_branches(config, gitea)
                await cache.refresh_meta_pulls(config, gitea)
            else:
                slug = _slug_for_repo(repo_full)
                if slug:
                    await cache.refresh_rfc_repo(config, gitea, slug)
                else:
                    log.info(
                        "webhook received for unknown repo: repo_full=%s event=%s "
                        "(no cached_rfcs row matched; hook may be on a fork or stale)",
                        repo_full, event,
                    )
        except Exception:
            log.exception("webhook refresh failed")
            raise HTTPException(status_code=500, detail="Refresh failed")
  • Step 7: main.py — drop seed_default_project; mirror the registry at startup (fatal on first boot)

In backend/app/main.py lifespan (async def lifespan), import registry: add registry as registry_mod, to the from . import (...) block. Then replace:

    db.init(config)
    projects.seed_default_project(config)  # §22.13 step 1
    gitea = Gitea(config)

with:

    db.init(config)
    gitea = Gitea(config)
    # §22.2: mirror the registry before anything reads projects/content_repo.
    # First boot has no last-good rows, so a missing/invalid registry is fatal
    # (loud-fail per separation-of-concerns); the reconciler sweep keeps it
    # fresh thereafter and tolerates a later bad PR.
    try:
        await registry_mod.refresh_registry(config, gitea)
    except Exception as e:
        raise RuntimeError(
            f"registry mirror failed at startup ({config.registry_repo_full}/projects.yaml): {e}"
        ) from e
    if projects.default_content_repo(config) is None:
        raise RuntimeError(
            f"registry does not describe the default project "
            f"{projects.resolved_default_id(config)!r} (no content_repo). "
            f"Add it to {config.registry_repo_full}/projects.yaml."
        )

Also update the startup log line near the end of lifespan — replace config.gitea_org, config.meta_repo with:

    log.info("RFC app started — registry %s", config.registry_repo_full)
  • Step 8: Update the shared test fixture to seed a registry

In backend/tests/test_propose_vertical.py:

In FakeGitea.__init__, after self._seed_repo("wiggleverse", "meta"), add:

        # §22 M3: the deployment's project registry. Startup refresh_registry
        # reads projects.yaml here; the single 'default' project's content_repo
        # points back at the seeded meta repo so the corpus mirror is unchanged.
        self._seed_repo("wiggleverse", "registry")
        self.files[("wiggleverse", "registry", "main", "projects.yaml")] = {
            "content": (
                "deployment:\n"
                "  name: Test Deployment\n"
                "  tagline: A test deployment\n"
                "projects:\n"
                "  - id: default\n"
                "    name: Test Deployment\n"
                "    type: document\n"
                "    content_repo: meta\n"
                "    visibility: public\n"
            ),
            "sha": "regsha0001",
        }

In the tmp_env fixture's env dict, add a registry repo name and drop the now-ignored META_REPO line. Change:

        "META_REPO": "meta",

to:

        "REGISTRY_REPO": "registry",
  • Step 9: Fix the one stale assertion comment

In backend/tests/test_multi_project_spine_vertical.py, the test_default_project_seeded_and_backfilled test still passes (content_repo is "meta"), but its inline comment references the retired META_REPO backfill. Update the comment on that assertion:

        # M3: content_repo now comes from the registry mirror (projects.yaml),
        # not the retired META_REPO startup backfill.
        assert row["content_repo"] == "meta"
  • Step 10: Run the wiring test, then the full suite

Run: cd backend && python -m pytest tests/test_registry_wiring.py -v Expected: PASS (2 tests).

Run: cd backend && grep -rn "config.meta_repo\|meta_repo_full\|seed_default_project" app/ Expected: no output (all references removed).

Run: cd backend && python -m pytest -q Expected: PASS (full suite green — the fixture now seeds the registry, so every app-startup test mirrors it and the corpus mirror reads content_repo='meta' exactly as before).

  • Step 11: Commit
git add backend/app/config.py backend/app/projects.py backend/app/cache.py \
        backend/app/webhooks.py backend/app/main.py backend/tests/test_propose_vertical.py \
        backend/tests/test_multi_project_spine_vertical.py backend/tests/test_registry_wiring.py \
        backend/tests/test_migration_027.py backend/tests/test_registry.py
git commit -m "feat(registry): hard-cut META_REPO -> REGISTRY_REPO; wire mirror into startup/webhook/sweep"

Task 6: Runtime-config APIs — GET /api/deployment + GET /api/projects/:id

Files:

  • Create: backend/app/api_deployment.py

  • Modify: backend/app/api.py (imports ~21-44; mount in make_router ~118-148)

  • Test: backend/tests/test_api_deployment.py

  • Step 1: Write the failing test

# backend/tests/test_api_deployment.py
"""§22.9/§22.5 — GET /api/deployment + GET /api/projects/:id with visibility."""
from __future__ import annotations

from fastapi.testclient import TestClient

from test_propose_vertical import (  # noqa: F401
    app_with_fake_gitea, tmp_env, provision_user_row, sign_in_as,
)


def _add_project(pid, name, vis, typ="document"):
    from app import db
    db.conn().execute(
        "INSERT OR REPLACE INTO projects (id, name, type, content_repo, visibility, initial_state) "
        "VALUES (?, ?, ?, ?, ?, 'super-draft')",
        (pid, name, typ, pid, vis),
    )


def test_deployment_lists_public_omits_gated_and_unlisted_for_anon(app_with_fake_gitea):
    app, _ = app_with_fake_gitea
    with TestClient(app) as client:
        _add_project("pub", "Public", "public")
        _add_project("gat", "Gated", "gated")
        _add_project("unl", "Unlisted", "unlisted")
        r = client.get("/api/deployment")
        assert r.status_code == 200
        body = r.json()
        assert body["name"] == "Test Deployment"
        ids = {p["id"] for p in body["projects"]}
        assert "pub" in ids and "default" in ids  # both public
        assert "gat" not in ids                    # gated, anon not a member
        assert "unl" not in ids                    # unlisted never enumerated


def test_projects_id_404_for_gated_non_member(app_with_fake_gitea):
    app, _ = app_with_fake_gitea
    with TestClient(app) as client:
        _add_project("gat", "Gated", "gated")
        assert client.get("/api/projects/gat").status_code == 404


def test_projects_id_returns_config_for_public(app_with_fake_gitea):
    app, _ = app_with_fake_gitea
    with TestClient(app) as client:
        from app import db
        db.conn().execute(
            "UPDATE projects SET config_json = ? WHERE id = 'default'",
            ('{"theme": {"accent": "#5b5bd6"}}',),
        )
        r = client.get("/api/projects/default")
        assert r.status_code == 200
        body = r.json()
        assert body["id"] == "default"
        assert body["type"] == "document"
        assert body["visibility"] == "public"
        assert body["theme"] == {"accent": "#5b5bd6"}


def test_projects_id_unlisted_readable_by_direct_id(app_with_fake_gitea):
    app, _ = app_with_fake_gitea
    with TestClient(app) as client:
        _add_project("unl", "Unlisted", "unlisted")
        assert client.get("/api/projects/unl").status_code == 200
  • Step 2: Run test to verify it fails

Run: cd backend && python -m pytest tests/test_api_deployment.py -v Expected: FAIL — 404 for /api/deployment (route not mounted).

  • Step 3: Write app/api_deployment.py
# backend/app/api_deployment.py
"""§22.9 runtime deployment/project config (replaces VITE_APP_NAME).

GET /api/deployment — the deployment name/tagline + the projects the caller can
see (§22.5: gated filtered by membership, unlisted omitted from enumeration).
GET /api/projects/:id — one project's runtime config + optional theme overlay,
gated behind the §22.5 read gate (404 for a non-member of a gated project).
"""
from __future__ import annotations

import json
from typing import Any

from fastapi import APIRouter, Request

from . import auth, db


def make_router() -> APIRouter:
    router = APIRouter()

    @router.get("/api/deployment")
    async def get_deployment(request: Request) -> dict[str, Any]:
        viewer = auth.current_user(request)
        dep = db.conn().execute(
            "SELECT name, tagline FROM deployment WHERE id = 1"
        ).fetchone()
        # §22.5: enumerate only public + (member-)gated; unlisted is never listed.
        visible = set(auth.visible_project_ids(viewer))
        rows = db.conn().execute(
            "SELECT id, name, type, visibility FROM projects "
            "WHERE visibility != 'unlisted' ORDER BY name"
        ).fetchall()
        projects = [
            {"id": r["id"], "name": r["name"], "type": r["type"], "visibility": r["visibility"]}
            for r in rows
            if r["id"] in visible
        ]
        return {
            "name": (dep["name"] if dep else "") or "",
            "tagline": (dep["tagline"] if dep else "") or "",
            "projects": projects,
        }

    @router.get("/api/projects/{project_id}")
    async def get_project(project_id: str, request: Request) -> dict[str, Any]:
        viewer = auth.current_user(request)
        # §22.5 read gate: a gated project 404s to a non-member (shape matches
        # an unknown id). unlisted is readable by direct id.
        auth.require_project_readable(viewer, project_id)
        row = db.conn().execute(
            "SELECT id, name, type, visibility, initial_state, config_json "
            "FROM projects WHERE id = ?",
            (project_id,),
        ).fetchone()
        if row is None:
            auth.require_project_readable(viewer, project_id)  # raises 404
        cfg = json.loads(row["config_json"] or "{}")
        dep = db.conn().execute("SELECT tagline FROM deployment WHERE id = 1").fetchone()
        return {
            "id": row["id"],
            "name": row["name"],
            "tagline": (dep["tagline"] if dep else "") or "",
            "type": row["type"],
            "visibility": row["visibility"],
            "initial_state": row["initial_state"],
            "theme": cfg.get("theme") or {},
        }

    return router

NOTE: require_project_readable (auth.py) returns gated→404 only when the row is missing OR the project is gated-and-not-member. For a truly unknown id, project_visibility returns 'gated', so a non-member gets 404 — the row is None guard then never reaches a real row. The redundant second require_project_readable call documents intent and is harmless.

  • Step 4: Mount the router in api.py

In backend/app/api.py, add api_deployment, to the from . import (...) block (alphabetical, near api_contributions). Then in make_router, after the other router.include_router(...) calls (after api_contributions), add:

    # §22.9 (M3): runtime deployment + per-project config (replaces VITE_APP_NAME).
    router.include_router(api_deployment.make_router())
  • Step 5: Run tests to verify they pass

Run: cd backend && python -m pytest tests/test_api_deployment.py -v Expected: PASS (4 tests).

  • Step 6: Run the full suite

Run: cd backend && python -m pytest -q Expected: PASS.

  • Step 7: Commit
git add backend/app/api_deployment.py backend/app/api.py backend/tests/test_api_deployment.py
git commit -m "feat(api): GET /api/deployment + /api/projects/:id (§22.9 runtime config)"

Task 7: Honor initial_state at propose time

Files:

  • Modify: backend/app/api.py (the propose_rfc endpoint ~797-845)
  • Test: backend/tests/test_initial_state_landing.py

SCOPE NOTE: Plan A sets the landed entry's state/unreviewed in frontmatter + cache per initial_state. The full active-entry lifecycle (skipping repo-creation/graduation for an active-landing project) is a bdd/type-surface concern deferred to M5; for document projects (the only live kind) initial_state is super-draft, so nothing changes in production.

  • Step 1: Write the failing test
# backend/tests/test_initial_state_landing.py
"""§22.4b — a project with initial_state='active' lands new entries active +
unreviewed; the default 'super-draft' project is unchanged."""
from __future__ import annotations

from fastapi.testclient import TestClient

from test_propose_vertical import (  # noqa: F401
    app_with_fake_gitea, tmp_env, provision_user_row, sign_in_as,
)


def _propose(client):
    return client.post("/api/rfcs/propose", json={
        "title": "Active Lander", "slug": "active-lander",
        "pitch": "Lands active.", "tags": [],
    })


def test_super_draft_default_unchanged(app_with_fake_gitea):
    from app import entry as entry_mod
    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        provision_user_row(user_id=1, login="ben", role="owner")
        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner")
        assert _propose(client).status_code == 200
        f = fake.files[("wiggleverse", "meta", "propose/active-lander", "rfcs/active-lander.md")]
        e = entry_mod.parse(f["content"])
        assert e.state == "super-draft"
        assert e.unreviewed is False


def test_active_initial_state_lands_active_unreviewed(app_with_fake_gitea):
    from app import db, entry as entry_mod
    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        db.conn().execute("UPDATE projects SET initial_state='active' WHERE id='default'")
        provision_user_row(user_id=1, login="ben", role="owner")
        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner")
        assert _propose(client).status_code == 200
        f = fake.files[("wiggleverse", "meta", "propose/active-lander", "rfcs/active-lander.md")]
        e = entry_mod.parse(f["content"])
        assert e.state == "active"
        assert e.unreviewed is True

NOTE: the branch name propose/active-lander matches what bot.open_idea_pr creates (confirm the exact prefix by reading bot.open_idea_pr; if it differs, adjust the file-key in the test). The assertion target is the entry file the bot wrote to the proposal branch.

  • Step 2: Run test to verify it fails

Run: cd backend && python -m pytest tests/test_initial_state_landing.py -v Expected: FAIL — the active case lands super-draft (the endpoint hardcodes it).

  • Step 3: Consult initial_state in propose_rfc

In backend/app/api.py, ensure projects is imported (add projects as projects_mod, to the from . import (...) block if absent). In propose_rfc, replace the entry = entry_mod.Entry( construction's state="super-draft", line. First, just before building the entry, add:

        # §22.4b: the target project's landing state. Through Plan A every
        # entry lands in the default project; M3-frontend routing carries a
        # non-default target later.
        target_project = auth.DEFAULT_PROJECT_ID
        landing_state = "active" if projects_mod.project_initial_state(target_project) == "active" else "super-draft"

Then change the Entry(...) constructor: replace state="super-draft", with:

            state=landing_state,
            unreviewed=(landing_state == "active"),
  • Step 4: Run tests to verify they pass

Run: cd backend && python -m pytest tests/test_initial_state_landing.py -v Expected: PASS (2 tests).

  • Step 5: Run the full suite

Run: cd backend && python -m pytest -q Expected: PASS.

  • Step 6: Commit
git add backend/app/api.py backend/tests/test_initial_state_landing.py
git commit -m "feat(propose): honor project initial_state (§22.4b active landing + unreviewed)"

Task 8: mark-reviewed action + catalog unreviewed filter

Files:

  • Modify: backend/app/bot.py (add a mark_reviewed write wrapper modeled on the graduation file-write)
  • Modify: backend/app/api.py (the list_rfcs catalog endpoint ~609-665; add the mark-reviewed endpoint near the other RFC endpoints)
  • Test: backend/tests/test_mark_reviewed.py

Before writing Step 3, read backend/app/bot.py's open_graduation_pr (≈700-767) and backend/app/api_graduation.py's _orchestrate (≈633-726) to copy the exact bot-write + inline-merge pattern (read file sha → update_file on a branch → open PR → wait_for_mergeable → merge → refresh cache). The mark-reviewed flow mirrors it: it rewrites the entry frontmatter with unreviewed: false + reviewed_at/reviewed_by, commits via the bot, merges, and refreshes cached_rfcs.

  • Step 1: Write the failing test
# backend/tests/test_mark_reviewed.py
"""§22.4c — owner/admin mark-reviewed clears the flag; catalog unreviewed filter."""
from __future__ import annotations

from fastapi.testclient import TestClient

from test_propose_vertical import (  # noqa: F401
    app_with_fake_gitea, tmp_env, provision_user_row, sign_in_as,
)


def _seed_unreviewed_active(fake, slug="feat"):
    """Put an active+unreviewed entry on the meta repo main + cache."""
    from app import cache, entry as entry_mod
    body = entry_mod.serialize(entry_mod.Entry(
        slug=slug, title="Feat", state="active", unreviewed=True,
        owners=["ben"], proposed_by="ben",
    ))
    fake.files[("wiggleverse", "meta", "main", f"rfcs/{slug}.md")] = {"content": body, "sha": "s1"}
    cache._upsert_cached_rfc(entry_mod.parse(body), body_sha="s1")


def test_catalog_unreviewed_filter(app_with_fake_gitea):
    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        _seed_unreviewed_active(fake, "feat")
        # a reviewed active entry should NOT appear under the filter
        from app import cache, entry as entry_mod
        ok = entry_mod.serialize(entry_mod.Entry(slug="ok", title="OK", state="active", owners=["ben"]))
        fake.files[("wiggleverse", "meta", "main", "rfcs/ok.md")] = {"content": ok, "sha": "s2"}
        cache._upsert_cached_rfc(entry_mod.parse(ok), body_sha="s2")
        r = client.get("/api/rfcs", params={"unreviewed": "true"})
        slugs = {i["slug"] for i in r.json()["items"]}
        assert slugs == {"feat"}


def test_mark_reviewed_clears_flag(app_with_fake_gitea):
    from app import db
    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        _seed_unreviewed_active(fake, "feat")
        provision_user_row(user_id=1, login="ben", role="owner")
        sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner")
        r = client.post("/api/projects/default/rfcs/feat/mark-reviewed")
        assert r.status_code == 200
        row = db.conn().execute(
            "SELECT unreviewed, reviewed_by FROM cached_rfcs WHERE slug='feat'"
        ).fetchone()
        assert row["unreviewed"] == 0
        assert row["reviewed_by"] == "ben"


def test_mark_reviewed_forbidden_for_non_superuser(app_with_fake_gitea):
    app, fake = app_with_fake_gitea
    with TestClient(app) as client:
        _seed_unreviewed_active(fake, "feat")
        provision_user_row(user_id=2, login="carol", role="contributor")
        sign_in_as(client, user_id=2, gitea_login="carol", display_name="Carol", role="contributor")
        r = client.post("/api/projects/default/rfcs/feat/mark-reviewed")
        assert r.status_code == 403
  • Step 2: Run test to verify it fails

Run: cd backend && python -m pytest tests/test_mark_reviewed.py -v Expected: FAIL — unreviewed query param ignored (filter returns both); mark-reviewed route 404.

  • Step 3: Add the catalog filter

In backend/app/api.py list_rfcs, change the signature to accept the query param and extend the WHERE clause. Replace the function signature line and the SQL:

    @router.get("/api/rfcs")
    async def list_rfcs(request: Request, unreviewed: str | None = None) -> dict[str, Any]:

and build the query with an optional unreviewed predicate:

        placeholders = ",".join("?" for _ in visible)
        params = list(visible)
        unreviewed_clause = ""
        if unreviewed is not None and unreviewed.lower() in ("1", "true", "yes"):
            unreviewed_clause = " AND unreviewed = 1 AND state = 'active'"
        rows = db.conn().execute(
            f"""
            SELECT slug, title, state, rfc_id, repo,
                   owners_json, arbiters_json, tags_json,
                   last_main_commit_at, last_entry_commit_at, updated_at
              FROM cached_rfcs
             WHERE state IN ('super-draft', 'active')
               AND project_id IN ({placeholders}){unreviewed_clause}
             ORDER BY COALESCE(last_main_commit_at, last_entry_commit_at) DESC
            """,
            params,
        ).fetchall()
  • Step 4: Add the mark-reviewed endpoint

In backend/app/api.py, near get_rfc (~667), add (it uses the bot wrapper from Step 5):

    @router.post("/api/projects/{project_id}/rfcs/{slug}/mark-reviewed")
    async def mark_reviewed(project_id: str, slug: str, request: Request) -> dict[str, Any]:
        """§22.4c — clear an active entry's `unreviewed` flag. Authority is the
        §22.7 project superuser (project_admin or deployment owner/admin), the
        same tier that graduates an entry."""
        viewer = auth.require_user(request)
        auth.require_project_readable(viewer, project_id)
        if not auth.is_project_superuser(viewer, project_id):
            raise HTTPException(403, "Only a project owner/admin can mark an entry reviewed")
        row = db.conn().execute(
            "SELECT state, unreviewed FROM cached_rfcs WHERE slug = ? AND project_id = ?",
            (slug, project_id),
        ).fetchone()
        if row is None:
            raise HTTPException(404, "Not found")
        if row["state"] != "active" or not row["unreviewed"]:
            raise HTTPException(409, "Entry is not an unreviewed active entry")
        try:
            await bot.mark_entry_reviewed(
                viewer.as_actor(),
                org=config.gitea_org,
                meta_repo=projects_mod.default_content_repo(config),
                slug=slug,
                reviewed_by=viewer.gitea_login,
                reviewed_at=entry_mod.today(),
            )
        except GiteaError as e:
            raise HTTPException(502, f"Gitea: {e.detail}")
        await cache.refresh_meta_repo(config, gitea)
        return {"ok": True}
  • Step 5: Add the bot.mark_entry_reviewed wrapper

In backend/app/bot.py, add a method modeled on open_graduation_pr but writing directly to main via the same authenticated update_file path (mark-reviewed is an owner gesture and needs no review PR — it parallels the graduation stamp, not the graduation flow). Read the existing bot write helpers first to match the On-behalf-of trailer + actions-log shape. The method:

    async def mark_entry_reviewed(
        self,
        actor,
        *,
        org: str,
        meta_repo: str,
        slug: str,
        reviewed_by: str,
        reviewed_at: str,
    ) -> None:
        """Clear §22.4c unreviewed on an active entry by rewriting its
        frontmatter on main. Mirrors the graduation stamp's bot-write shape."""
        from . import entry as entry_mod
        path = f"rfcs/{slug}.md"
        result = await self._gitea.read_file(org, meta_repo, path, ref="main")
        if result is None:
            raise GiteaError(404, f"{path} not found")
        text, sha = result
        e = entry_mod.parse(text)
        e.unreviewed = False
        e.reviewed_at = reviewed_at
        e.reviewed_by = reviewed_by
        await self._update_file_on_behalf(
            actor,
            org=org, repo=meta_repo, path=path,
            content=entry_mod.serialize(e), sha=sha, branch="main",
            message=f"Mark {slug} reviewed",
        )

NOTE: match the real helper names in bot.pyself._gitea, and the existing on-behalf write helper (it may be named differently than _update_file_on_behalf). Read bot.py first and use the actual private write method + actions-log call the other bot methods use, so mark-reviewed carries the same §6.5 trailer + actions row.

  • Step 6: Run tests to verify they pass

Run: cd backend && python -m pytest tests/test_mark_reviewed.py -v Expected: PASS (3 tests).

  • Step 7: Run the full suite

Run: cd backend && python -m pytest -q Expected: PASS.

  • Step 8: Commit
git add backend/app/api.py backend/app/bot.py backend/tests/test_mark_reviewed.py
git commit -m "feat(review): mark-reviewed action + §7 catalog unreviewed filter (§22.4c)"

Task 9: Release — env docs, changelog, version bump

Files:

  • Modify: backend/.env.example

  • Modify: CHANGELOG.md

  • Modify: VERSION, frontend/package.json

  • Step 1: Update backend/.env.example

Add REGISTRY_REPO (required) and DEFAULT_PROJECT_ID (optional), and remove META_REPO. Insert near the other Gitea settings:

# §22.2 — the project registry repo (REQUIRED). The framework reads
# `projects.yaml` at its root to learn which projects exist. The repo name is
# the deployment's choice; the app fails to start if this is unset.
REGISTRY_REPO=registry
# §22.13 — optional id for the bootstrap/default project. Reserved for the
# Plan B re-stamp; leave unset in Plan A (the default project id stays
# `default`). When set, it must match an `id` in projects.yaml.
# DEFAULT_PROJECT_ID=ohm

Delete the META_REPO=... line and its comment.

  • Step 2: Run a grep to confirm no doc still tells operators to set META_REPO

Run: cd backend && grep -rn "META_REPO" .env.example app/ ; echo "exit: $?" Expected: no matches in .env.example or app/ (a match in a migration-026 comment is fine — it's historical).

  • Step 3: Add the CHANGELOG entry with upgrade steps

In CHANGELOG.md, add a new top entry for 0.33.0 (a breaking, pre-1.0 minor). Match the file's existing entry format; the body must include an upgrade-steps block:

## 0.33.0 — §22 M3 (backend, Plan A): registry mirror + runtime config

### Added
- Project **registry mirror** (`app/registry.py`): the framework now learns its
  projects from `REGISTRY_REPO/projects.yaml`, mirrored into the `projects`
  table + a new `deployment` singleton on the §4 webhook + reconciler.
- `GET /api/deployment` (name, tagline, visible projects) and
  `GET /api/projects/:id` (name, type, visibility, initial_state, theme) —
  runtime config replacing the build-time `VITE_APP_NAME` (frontend cut lands in
  M3-frontend).
- `§22.4b initial_state` honored at propose time; `§22.4c unreviewed` flag,
  mark-reviewed action (`POST /api/projects/:id/rfcs/:slug/mark-reviewed`), and a
  `?unreviewed=true` catalog filter.
- Migration `027` (additive): `projects.type`/`initial_state`, the `deployment`
  table, and `cached_rfcs.unreviewed`/`reviewed_at`/`reviewed_by`.

### Breaking
- **`META_REPO` is retired; `REGISTRY_REPO` is required.** The corpus mirror now
  reads each project's `content_repo` from the registry.

### Upgrade steps
1. Create a registry repo under your deployment's Gitea org.
2. Author `projects.yaml` at its root: a `deployment:` block (`name`, `tagline`)
   and one `projects:` entry for your existing corpus — `id: default`, `name`,
   `type: document`, `content_repo: <your old META_REPO value>`,
   `visibility: public`. (Keep `id: default` for this release; the pretty-slug
   re-stamp lands in the next backend slice, before any `/p/` URL is public.)
3. Set `REGISTRY_REPO=<your registry repo name>` and remove `META_REPO`.
4. Add a Gitea webhook on the registry repo pointing at `/api/webhooks/gitea`
   (same secret as the corpus webhook).
5. Deploy. Migration `027` runs; the registry mirror reconciles at startup.
   Verify `GET /api/deployment` returns your project and `/api/health` is green.
  • Step 4: Bump the version

Set the version to 0.33.0 in both files (they must match per SPEC §20):

printf '0.33.0\n' > VERSION

In frontend/package.json, change "version": "0.32.0", to "version": "0.33.0",.

  • Step 5: Verify version match + full suite

Run: cat VERSION && grep '"version"' frontend/package.json Expected: both show 0.33.0.

Run: cd backend && python -m pytest -q Expected: PASS (full suite).

  • Step 6: Commit
git add backend/.env.example CHANGELOG.md VERSION frontend/package.json
git commit -m "release: 0.33.0 — §22 M3 backend Plan A (registry mirror + runtime config)"

Self-review checklist (completed by plan author)

Spec coverage (Plan A scope only):

  • Registry format + registry.py mirror (spec §2) → Task 4 + Task 5.
  • GET /api/deployment + /api/projects/:id (spec §4) → Task 6.
  • initial_state honoring (spec §5c) → Task 7.
  • unreviewed parse/serialize (spec §5a) → Task 1; cache mirror (§5b) → Task 3; mark-reviewed (§5d) → Task 8; catalog filter query side (§5e) → Task 8.
  • Migration 027 additive columns + deployment table (spec §1a/§1b) → Task 2.
  • Config hard cut, META_REPO retirement (spec §3) → Task 5.
  • Versioning + upgrade steps (spec §8) → Task 9.
  • Deferred to Plan B (per spec §1 decision 6): the 12-table PK rebuilds (§1c), the default→slug re-stamp (§1d), project_id writer threading, two-project isolation tests. Not in this plan by design.

Placeholder scan: No "TBD"/"add error handling"/"similar to" — every code step shows the code. Two explicit "read X first to match the real helper name" notes (Task 8 bot wrapper, Task 7 branch-name) are deliberate: they point at existing code the worker must mirror exactly rather than guessing a signature.

Type/name consistency: Entry.unreviewed/reviewed_at/reviewed_by (Task 1) are read by _upsert_cached_rfc (Task 3), the registry's ProjectEntry/RegistryDoc/RegistryError/parse_registry/apply_registry/refresh_registry (Task 4) are consumed by the wiring (Task 5), projects.default_content_repo/resolved_default_id/project_initial_state (Task 5) are used by cache/webhooks/propose (Tasks 5/7) and mark-reviewed (Task 8). Names are consistent across tasks.