Files
rfc-app/backend/tests/test_per_rfc_models.py
Ben Stull a255429e57 Post-v1: per-RFC model availability (UX half) folded into §6.6
First §19.2 candidate settled after v1. The heavier per-RFC-model
topic subdivided into UX (this) and credential delegation + funder
role (still §19.2). New §6.6 carries the rule: an optional `models:`
frontmatter field on the meta-repo RFC entry; absent inherits the
operator universe, populated narrows the picker to the intersection
with provisioned providers, `[]` opts the RFC out of AI entirely.
The first resolved entry is the RFC default. §18's ENABLED_MODELS is
reframed as the operator universe.

Code: migration 009 adds nullable cached_rfcs.models_json (NULL ≠ []
is load-bearing); entry.py grows the optional field with absent-vs-
empty round-tripping in parse/serialize; new models_resolver module
holds the rule; api_branches replaces /api/models with the slug-aware
/api/rfcs/{slug}/models and threads the chat + reask paths through
the resolver; api_prs §10.2 uses the resolver and extends the stub
fallback to the opt-out case; frontend passes slug to listModels.

Tests 106/106 green (96 prior + 10 in test_per_rfc_models.py). No
behavioral change for entries without `models:` — operator universe
preserved as default.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-25 05:42:15 -07:00

409 lines
17 KiB
Python

"""Integration coverage for the §19.2 "per-RFC model availability — UX
half" candidate folded into §6.6.
The meta-repo entry's optional `models:` frontmatter narrows what AI
models contributors can pick from on a given RFC. The resolution rule
is uniform across every AI surface:
- Absent (`models:` key omitted) → inherit the operator universe.
- Populated list → intersection with the operator's provisioned
providers, preserving the entry's order. First entry is the RFC
default.
- Empty list (`models: []`) → AI surfaces refuse honestly.
The tests below prove each branch through the API surface and verify
the cache round-trip from meta-repo frontmatter to picker option list.
"""
from __future__ import annotations
import asyncio
import json
from test_propose_vertical import ( # noqa: F401
FakeGitea,
app_with_fake_gitea,
provision_user_row,
sign_in_as,
tmp_env,
)
from test_rfc_view_vertical import FakeProvider, seed_active_rfc # noqa: F401
from test_super_draft_vertical import seed_super_draft # noqa: F401
SEED_BODY = (
"Open Human Model is a framework for representing humans.\n\n"
"It defines consent, trait, and agency in compatible terms."
)
def _set_models_json(slug: str, value: str | None) -> None:
"""Write the resolved cached_rfcs.models_json directly. Mirrors what
`_upsert_cached_rfc` would do on the next reconciler sweep, without
requiring a full meta-repo refresh in tests focused on the resolver."""
from app import db
db.conn().execute(
"UPDATE cached_rfcs SET models_json = ? WHERE slug = ?",
(value, slug),
)
def _install_two_providers(app) -> None:
"""Swap two fake providers — `claude` and `gemini` — into the
shared providers dict. The dict is held by reference inside the
router closures (see Slice 2's pattern in test_rfc_view_vertical),
so the mutation propagates."""
app.state.providers.clear()
app.state.providers["claude"] = FakeProvider("TITLE: A\nDESCRIPTION: B")
app.state.providers["gemini"] = FakeProvider("TITLE: G\nDESCRIPTION: H")
# ---------------------------------------------------------------------------
# /api/rfcs/<slug>/models — the picker option-list surface
# ---------------------------------------------------------------------------
def test_absent_frontmatter_inherits_operator_universe(app_with_fake_gitea):
"""When the meta-repo entry has no `models:` field, the picker's
option list is the operator's full provisioned universe."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner", email="ben@test")
_install_two_providers(app)
# models_json is NULL — set_models_json not called.
r = client.get("/api/rfcs/ohm/models")
assert r.status_code == 200, r.text
body = r.json()
ids = [m["id"] for m in body["models"]]
assert ids == ["claude", "gemini"]
assert body["default"] == "claude"
def test_populated_frontmatter_narrows_picker_to_intersection(app_with_fake_gitea):
"""A populated `models:` list narrows the picker. Operator order
does not matter — the entry's stated order is preserved."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner", email="ben@test")
_install_two_providers(app)
_set_models_json("ohm", json.dumps(["gemini"]))
r = client.get("/api/rfcs/ohm/models")
body = r.json()
assert [m["id"] for m in body["models"]] == ["gemini"]
assert body["default"] == "gemini"
def test_empty_frontmatter_disables_ai_surfaces(app_with_fake_gitea):
"""`models: []` — the RFC opts out of AI per §6.6. The picker
surface returns an empty list and an empty default."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner", email="ben@test")
_install_two_providers(app)
_set_models_json("ohm", json.dumps([]))
r = client.get("/api/rfcs/ohm/models")
body = r.json()
assert body["models"] == []
assert body["default"] == ""
def test_intersection_with_provisioned_providers(app_with_fake_gitea):
"""An entry that lists models the operator no longer provisions
— they're silently hidden from the picker. The frontmatter list
is preserved so a later operator change can restore them; the
cache row is untouched by this test."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner", email="ben@test")
_install_two_providers(app)
# Frontmatter names a model the operator doesn't have plus one
# it does — only the intersection comes out.
_set_models_json("ohm", json.dumps(["llama-3", "claude"]))
r = client.get("/api/rfcs/ohm/models")
body = r.json()
assert [m["id"] for m in body["models"]] == ["claude"]
assert body["default"] == "claude"
def test_intersection_empty_same_as_opt_out(app_with_fake_gitea):
"""When the frontmatter names only models the operator doesn't
provision, the resolved list is empty and the surface refuses
cleanly — same shape as the explicit `models: []` opt-out."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
sign_in_as(client, user_id=1, gitea_login="ben",
display_name="Ben", role="owner", email="ben@test")
_install_two_providers(app)
_set_models_json("ohm", json.dumps(["llama-3", "gpt-5"]))
r = client.get("/api/rfcs/ohm/models")
assert r.json()["models"] == []
# ---------------------------------------------------------------------------
# §10.2 PR draft: model selection follows the RFC default
# ---------------------------------------------------------------------------
def test_pr_draft_uses_rfc_default_model(app_with_fake_gitea):
"""The §10.2 draft picks the RFC's default — the first entry of
the resolved list. We rig the providers so `claude` returns one
title and `gemini` returns a different one, then prove the entry's
pinned model is the one we get back."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
provision_user_row(user_id=2, login="alice", role="contributor")
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
# Two fakes, distinct outputs.
app.state.providers.clear()
app.state.providers["claude"] = FakeProvider(
"TITLE: from claude\nDESCRIPTION: claude wrote this."
)
app.state.providers["gemini"] = FakeProvider(
"TITLE: from gemini\nDESCRIPTION: gemini wrote this."
)
_set_models_json("ohm", json.dumps(["gemini"]))
sign_in_as(client, user_id=2, gitea_login="alice",
display_name="Alice", role="contributor")
r = client.post("/api/rfcs/ohm/branches/main/promote-to-branch", json={})
branch = r.json()["branch_name"]
# Make the branch differ from main so the draft endpoint passes
# its "commits ahead" precondition.
from app import db
from test_rfc_view_vertical import FakeGitea # noqa: F401
# Promote-to-branch starts a branch off main with main's content;
# the simplest way to give it a commit-ahead is via a manual edit
# flush. But for this test, an inline commit on the fake is
# cleaner.
repo_full = "wiggleverse/rfc-0001-ohm"
owner, repo = repo_full.split("/", 1)
new_body = SEED_BODY + "\n\nA further sentence."
fake.files[(owner, repo, branch, "RFC.md")] = {
"content": new_body, "sha": fake._next_sha(),
}
fake.branches[(owner, repo)][branch]["sha"] = fake.files[
(owner, repo, branch, "RFC.md")
]["sha"]
db.conn().execute(
"""
UPDATE cached_branches SET head_sha = ?
WHERE rfc_slug = 'ohm' AND branch_name = ?
""",
(fake.branches[(owner, repo)][branch]["sha"], branch),
)
r = client.post(f"/api/rfcs/ohm/branches/{branch}/pr-draft")
assert r.status_code == 200, r.text
# `gemini` is the resolved default — its title comes back.
assert r.json()["title"] == "from gemini"
def test_pr_draft_falls_back_to_stub_on_opt_out(app_with_fake_gitea):
"""When the RFC opts out of AI (`models: []`), §10.2 falls back to
its deterministic stub — the title is the stub format, not an
LLM-generated string."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
provision_user_row(user_id=2, login="alice", role="contributor")
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
_install_two_providers(app)
_set_models_json("ohm", json.dumps([]))
sign_in_as(client, user_id=2, gitea_login="alice",
display_name="Alice", role="contributor")
r = client.post("/api/rfcs/ohm/branches/main/promote-to-branch", json={})
branch = r.json()["branch_name"]
from app import db
repo_full = "wiggleverse/rfc-0001-ohm"
owner, repo = repo_full.split("/", 1)
new_body = SEED_BODY + "\n\nA further sentence."
fake.files[(owner, repo, branch, "RFC.md")] = {
"content": new_body, "sha": fake._next_sha(),
}
fake.branches[(owner, repo)][branch]["sha"] = fake.files[
(owner, repo, branch, "RFC.md")
]["sha"]
db.conn().execute(
"""
UPDATE cached_branches SET head_sha = ?
WHERE rfc_slug = 'ohm' AND branch_name = ?
""",
(fake.branches[(owner, repo)][branch]["sha"], branch),
)
r = client.post(f"/api/rfcs/ohm/branches/{branch}/pr-draft")
assert r.status_code == 200, r.text
body = r.json()
# The stub uses "Edits to <RFC title>" as the title (per Slice 3).
assert body["title"] == "Edits to OHM"
# ---------------------------------------------------------------------------
# Chat surface refuses cleanly when AI is unavailable for the RFC
# ---------------------------------------------------------------------------
def test_chat_refuses_when_rfc_opted_out_of_ai(app_with_fake_gitea):
"""A chat turn on a branch of an RFC with `models: []` refuses
with 503 — same shape as "no providers configured." The refusal
surfaces as a property of the RFC, not a server failure."""
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=2, login="alice", role="contributor")
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
_install_two_providers(app)
_set_models_json("ohm", json.dumps([]))
sign_in_as(client, user_id=2, gitea_login="alice",
display_name="Alice", role="contributor")
r = client.post("/api/rfcs/ohm/branches/main/promote-to-branch", json={})
branch = r.json()["branch_name"]
view = client.get(f"/api/rfcs/ohm/branches/{branch}").json()
thread_id = view["main_thread_id"]
r = client.post(
f"/api/rfcs/ohm/branches/{branch}/threads/{thread_id}/chat",
json={"text": "tighten this", "model": "claude"},
)
assert r.status_code == 503
# ---------------------------------------------------------------------------
# Cache round-trip: frontmatter `models:` flows into models_json
# ---------------------------------------------------------------------------
def test_meta_repo_frontmatter_models_round_trips_through_cache(app_with_fake_gitea):
"""The production path: a meta-repo entry whose frontmatter carries
`models: [claude]` lands in `cached_rfcs.models_json` after the
next `refresh_meta_repo` sweep."""
from fastapi.testclient import TestClient
from app import cache as cache_mod, db
app, fake = app_with_fake_gitea
with TestClient(app) as client:
provision_user_row(user_id=1, login="ben", role="owner")
# Seed a super-draft entry directly on meta-main with a
# `models:` frontmatter field. seed_super_draft doesn't take
# this field, so we write the meta-repo file ourselves and
# let the cache reconciler parse it.
entry_text = (
"---\n"
"slug: ohm\n"
"title: OHM\n"
"state: super-draft\n"
"id: null\n"
"repo: null\n"
"proposed_by: ben\n"
"proposed_at: 2026-05-23\n"
"graduated_at: null\n"
"graduated_by: null\n"
"owners: []\n"
"arbiters: []\n"
"tags: []\n"
"models:\n"
" - claude\n"
"---\n\n"
"The pitch goes here.\n"
)
sha = fake._next_sha()
fake.files[("wiggleverse", "meta", "main", "rfcs/ohm.md")] = {
"content": entry_text, "sha": sha,
}
fake.branches[("wiggleverse", "meta")]["main"]["sha"] = sha
asyncio.run(cache_mod.refresh_meta_repo(app.state.config, app.state.gitea))
row = db.conn().execute(
"SELECT models_json FROM cached_rfcs WHERE slug = 'ohm'"
).fetchone()
assert row is not None
assert row["models_json"] is not None
assert json.loads(row["models_json"]) == ["claude"]
# And an entry without `models:` lands NULL.
no_models_text = entry_text.replace("models:\n - claude\n", "")
sha2 = fake._next_sha()
fake.files[("wiggleverse", "meta", "main", "rfcs/beta.md")] = {
"content": no_models_text.replace("slug: ohm", "slug: beta")
.replace("title: OHM", "title: Beta"),
"sha": sha2,
}
fake.branches[("wiggleverse", "meta")]["main"]["sha"] = sha2
asyncio.run(cache_mod.refresh_meta_repo(app.state.config, app.state.gitea))
row = db.conn().execute(
"SELECT models_json FROM cached_rfcs WHERE slug = 'beta'"
).fetchone()
assert row is not None
assert row["models_json"] is None
# ---------------------------------------------------------------------------
# entry.py: absent / empty / populated round-trip the parser+serializer
# ---------------------------------------------------------------------------
def test_entry_models_field_absent_vs_empty_distinguished():
"""The absent vs. empty distinction is load-bearing per §6.6.
parse(serialize(x)) preserves None as None and [] as []."""
from app import entry as entry_mod
absent = entry_mod.Entry(slug="x", title="X")
assert absent.models is None
text_absent = entry_mod.serialize(absent)
assert "models:" not in text_absent
round_absent = entry_mod.parse(text_absent)
assert round_absent.models is None
opted_out = entry_mod.Entry(slug="y", title="Y", models=[])
text_empty = entry_mod.serialize(opted_out)
assert "models: []" in text_empty
round_empty = entry_mod.parse(text_empty)
assert round_empty.models == []
populated = entry_mod.Entry(slug="z", title="Z", models=["claude", "gemini"])
text_full = entry_mod.serialize(populated)
round_full = entry_mod.parse(text_full)
assert round_full.models == ["claude", "gemini"]