benstull/rfc-app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Slice 2: the §8 active-RFC view in full

Browse Source 
Per the §19.1 brief: the three-column shape (§8.1) opens on main
in discuss mode (§8.2), supports the §8.3 discuss-vs-contribute
flip on non-main branches, hosts §8.4's per-branch chat with AI
participation (§18's <change> protocol → §8.14 changes rows), the
§8.8 change-card panel with §8.9 accept/decline/edit-before-accept,
the §8.10 tracked-change markup + DiffView toggle, the §8.11
manual-edit flushes with the stale-change mechanic, the §8.12
range and paragraph sub-threads, the §8.13 flag affordance, and
the §8.14 discuss-mode buffer.

Backend: bot.py grew per-RFC-repo write ops (cut_branch_from_main,
commit_accepted_change with the structured original/proposed/reason
body and Change-Id + Source-Message-Id + On-behalf-of trailers,
commit_manual_flush, ensure_rfc_repo_seed). cache.py grew
refresh_rfc_repo and the webhook dispatches on repository.full_name.
providers.py and chat.py port the §18 carryovers — multi-provider
LLM abstraction and SSE-streaming chat against the §5 threads /
thread_messages / changes schema. api_branches.py mounts the §17
branches/<branch>/* and threads/<thread_id>/* routes with the §6
/ §11 permission checks inline.

Frontend: RFCView.jsx rebuilt as the §8 surface; Editor.jsx,
ChatPanel.jsx, ChangePanel.jsx, PromptBar.jsx, SelectionTooltip.jsx,
DiffView.jsx, ModelPicker.jsx, modelStyles.js lifted from the
prototype and adapted to the canonical schema.

Covered by `backend/tests/test_rfc_view_vertical.py` — eleven new
integration tests against an extended FakeGitea (PUT contents,
POST orgs/{org}/repos, seed_rfc_repo): main-view read,
promote-to-branch, accept (with and without edit-before-accept),
decline, manual flush + system message, flag creation, visibility
flip, anonymous read-but-no-contribute, stale-change refusal, and
the chat-streaming path with a fake provider injected. The 5
Slice 1 tests continue to pass alongside.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ben Stull
2026-05-24 04:35:14 -07:00
parent 779ba6db59
commit 3bc8fe92af
24 changed files with 5433 additions and 151 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/tests/test_propose_vertical.py
+54 -6
View File
@@ -34,22 +34,38 @@ import pytest
class FakeGitea:
"""A narrow in-memory simulation of the Gitea API the slice uses."""
"""A narrow in-memory simulation of the Gitea API the slices exercise.
Slice 2 extends the seam to cover per-RFC repos: PUT contents
(update file), POST orgs/{org}/repos (create repo), and branch
listing with commit timestamps. The simulator is intentionally
minimal — only the routes the production paths actually call.
"""
def __init__(self):
# files: (owner, repo, branch, path) -> {"content": str, "sha": str}
self.files: dict[tuple[str, str, str, str], dict] = {}
# branches: (owner, repo) -> {branch_name -> {"sha": str}}
# branches: (owner, repo) -> {branch_name -> {"sha": str, "ts": str}}
self.branches: dict[tuple[str, str], dict[str, dict]] = {}
# pulls: (owner, repo) -> list[pull-dict]
self.pulls: dict[tuple[str, str], list[dict]] = {}
# repos: set of (owner, repo)
self.repos: set[tuple[str, str]] = set()
self._pr_counter = 0
self._commit_counter = 0
self._seed_repo("wiggleverse", "meta")
def _seed_repo(self, owner, repo):
self.branches[(owner, repo)] = {"main": {"sha": "initial"}}
self.branches[(owner, repo)] = {"main": {"sha": "initial", "ts": "2026-05-23T00:00:00Z"}}
self.pulls[(owner, repo)] = []
self.repos.add((owner, repo))
def seed_rfc_repo(self, owner, repo, *, rfc_md_body):
"""Convenience: seed a per-RFC repo with an RFC.md on main."""
self._seed_repo(owner, repo)
sha = self._next_sha()
self.files[(owner, repo, "main", "RFC.md")] = {"content": rfc_md_body, "sha": sha}
self.branches[(owner, repo)]["main"] = {"sha": sha, "ts": "2026-05-23T00:00:00Z"}
def _next_sha(self):
self._commit_counter += 1
@@ -62,8 +78,29 @@ class FakeGitea:
payload = json.loads(body) if body else {}
# GET /repos/{owner}/{repo}
if method == "GET" and re.fullmatch(r"/repos/[^/]+/[^/]+", path):
return httpx.Response(200, json={"name": path.split("/")[-1]})
m_repo = re.fullmatch(r"/repos/([^/]+)/([^/]+)", path)
if method == "GET" and m_repo:
owner, repo = m_repo.groups()
if (owner, repo) in self.repos:
return httpx.Response(200, json={"name": repo, "full_name": f"{owner}/{repo}"})
return httpx.Response(404, json={"message": "not found"})
# POST /orgs/{org}/repos
m = re.fullmatch(r"/orgs/([^/]+)/repos", path)
if method == "POST" and m:
org = m.group(1)
name = payload["name"]
self._seed_repo(org, name)
return httpx.Response(201, json={"name": name, "full_name": f"{org}/{name}"})
# GET /repos/{owner}/{repo}/branches (list)
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/branches", path)
if method == "GET" and m:
owner, repo = m.groups()
items = []
for name, b in self.branches.get((owner, repo), {}).items():
items.append({"name": name, "commit": {"id": b["sha"], "timestamp": b.get("ts")}})
return httpx.Response(200, json=items)
# GET /repos/{owner}/{repo}/branches/{branch}
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/branches/([^/]+)", path)
@@ -126,9 +163,20 @@ class FakeGitea:
content = base64.b64decode(payload["content"]).decode()
sha = self._next_sha()
self.files[(owner, repo, branch, fpath)] = {"content": content, "sha": sha}
self.branches[(owner, repo)][branch]["sha"] = sha
self.branches[(owner, repo)][branch] = {"sha": sha, "ts": "2026-05-23T00:00:00Z"}
return httpx.Response(201, json={"commit": {"sha": sha}})
# PUT /repos/{owner}/{repo}/contents/{path} — update_file
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/contents/(.+)", path)
if method == "PUT" and m:
owner, repo, fpath = m.groups()
branch = payload["branch"]
content = base64.b64decode(payload["content"]).decode()
sha = self._next_sha()
self.files[(owner, repo, branch, fpath)] = {"content": content, "sha": sha}
self.branches[(owner, repo)][branch] = {"sha": sha, "ts": "2026-05-23T00:00:00Z"}
return httpx.Response(200, json={"commit": {"sha": sha}, "content": {"sha": sha}})
# GET /repos/{owner}/{repo}/pulls?state=...
m = re.fullmatch(r"/repos/([^/]+)/([^/]+)/pulls", path)
if method == "GET" and m:
backend/tests/test_rfc_view_vertical.py
+564
View File
@@ -0,0 +1,564 @@
"""End-to-end integration tests for the Slice 2 vertical (§8 in full).
Reuses FakeGitea + the session-cookie forging helpers from
`test_propose_vertical.py`, extends FakeGitea with the per-RFC repo
routes Slice 2 needs (PUT contents, POST orgs/{org}/repos, seeded
RFC.md), and walks the §8 vertical end-to-end against an in-process
fake Gitea:
* Seed an active RFC with a per-RFC repo holding RFC.md.
* GET /api/rfcs/<slug>/main and /branches/<branch> — three-column
feed against the cache + live branch read.
* POST promote-to-branch — cut a new branch from main.
* Materialize an AI-style change directly in the database (the LLM
is mocked out where possible; one separate test exercises the
chat streaming path with a fake provider injected).
* POST accept — runs the bot's commit and updates `changes` row.
* POST decline — non-commit path; row persists as evidence.
* POST manual-flush — bot commit, system message lands in branch chat.
* POST threads — create a flag, surface it on subsequent reads.
* POST visibility — flip read_public and contribute_mode.
* POST chat — fake provider returns a known <change> block; the
response materializes a `changes` row.
"""
from __future__ import annotations
import json
import pytest
# Reuse the harness already proven by Slice 1. We import via the
# top-level module name (no leading dot) because pytest discovers
# `tests/` as a flat directory of test modules without an __init__.py.
from test_propose_vertical import ( # noqa: F401 — fixtures land via import
FakeGitea,
app_with_fake_gitea,
provision_user_row,
sign_in_as,
tmp_env,
)
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def seed_active_rfc(fake: FakeGitea, *, slug: str, title: str, body: str) -> str:
"""Seed an active RFC end-to-end: create the meta-repo entry, the
per-RFC repo with RFC.md on main, and the cached_rfcs row. The
real graduation flow lands in Slice 5; until it exists, this is
the test seam for "the RFC view's preconditions are met."
"""
from app import db
import yaml
repo_full = f"wiggleverse/rfc-0001-{slug}"
owner, repo = repo_full.split("/", 1)
fake.seed_rfc_repo(owner, repo, rfc_md_body=body)
# Meta-repo entry — what the cache would mirror after graduation.
fm = {
"slug": slug,
"title": title,
"state": "active",
"id": "RFC-0001",
"repo": repo_full,
"proposed_by": "alice",
"proposed_at": "2026-05-01",
"graduated_at": "2026-05-22",
"graduated_by": "ben",
"owners": ["alice"],
"arbiters": ["ben"],
"tags": ["identity"],
}
entry_text = "---\n" + yaml.safe_dump(fm, sort_keys=False).rstrip() + "\n---\n"
sha = fake._next_sha()
fake.files[("wiggleverse", "meta", "main", f"rfcs/{slug}.md")] = {"content": entry_text, "sha": sha}
# Write cached_rfcs row directly — the reconciler would also write
# this on its next sweep, but the test seam avoids the extra hop.
db.conn().execute(
"""
INSERT OR REPLACE INTO cached_rfcs
(slug, title, state, rfc_id, repo, proposed_by, proposed_at,
graduated_at, graduated_by, owners_json, arbiters_json, tags_json,
body, body_sha, last_main_commit_at, last_entry_commit_at)
VALUES (?, ?, 'active', ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, datetime('now'), datetime('now'))
""",
(
slug,
title,
"RFC-0001",
repo_full,
"alice",
"2026-05-01",
"2026-05-22",
"ben",
json.dumps(["alice"]),
json.dumps(["ben"]),
json.dumps(["identity"]),
body,
sha,
),
)
# Seed cached_branches for main, since the reconciler hasn't necessarily
# run yet inside the test client's lifespan. The webhook+reconciler
# path is what writes this in production; we shortcut it here.
db.conn().execute(
"""
INSERT OR IGNORE INTO cached_branches (rfc_slug, branch_name, head_sha, state, last_commit_at)
VALUES (?, 'main', ?, 'open', datetime('now'))
""",
(slug, sha),
)
return repo_full
# ---------------------------------------------------------------------------
# Tests
# ---------------------------------------------------------------------------
SEED_BODY = """# Open Human Model
Open Human Model is a framework for representing humans.
It defines consent, trait, and agency in compatible terms.
"""
def test_rfc_main_view_renders_against_per_rfc_repo(app_with_fake_gitea):
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="open-human-model", title="Open Human Model", body=SEED_BODY)
sign_in_as(client, user_id=1, gitea_login="ben", display_name="Ben", role="owner")
r = client.get("/api/rfcs/open-human-model/main")
assert r.status_code == 200, r.text
d = r.json()
assert d["slug"] == "open-human-model"
assert "Open Human Model" in d["body"]
# main is in the branches list (cached).
assert any(b["name"] == "main" for b in d["branches"])
def test_promote_to_branch_creates_branch_and_navigates(app_with_fake_gitea):
from fastapi.testclient import TestClient
from app import db
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)
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={})
assert r.status_code == 200, r.text
branch_name = r.json()["branch_name"]
assert branch_name.startswith("alice-draft-")
# The branch is reachable as its own view.
r = client.get(f"/api/rfcs/ohm/branches/{branch_name}")
assert r.status_code == 200, r.text
view = r.json()
assert view["branch_name"] == branch_name
# The branch starts from main's body — the editor opens on it.
assert "Open Human Model" in view["body"]
# The whole-doc chat thread exists by default.
assert view["main_thread_id"]
# The bot's create_branch action is in the audit log per §6.5.
actions = db.conn().execute(
"SELECT action_kind, on_behalf_of FROM actions WHERE action_kind = 'create_branch'"
).fetchall()
assert any((a["action_kind"], a["on_behalf_of"]) == ("create_branch", "alice") for a in actions)
def test_accept_ai_change_commits_and_updates_row(app_with_fake_gitea):
from fastapi.testclient import TestClient
from app import db
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)
sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
# Cut a branch the contributor owns.
r = client.post("/api/rfcs/ohm/branches/main/promote-to-branch", json={})
branch = r.json()["branch_name"]
# The whole-doc chat thread is created lazily on first branch
# view (§8.12) — GET the branch so it materializes.
view = client.get(f"/api/rfcs/ohm/branches/{branch}").json()
thread_id = view["main_thread_id"]
cur = db.conn().execute(
"""
INSERT INTO changes (rfc_slug, branch_name, thread_id, kind, state,
original, proposed, reason)
VALUES ('ohm', ?, ?, 'ai', 'pending', ?, ?, ?)
""",
(
branch,
thread_id,
"Open Human Model is a framework for representing humans.",
"Open Human Model is a framework for representing humans in software systems.",
"tightens scope",
),
)
change_id = cur.lastrowid
r = client.post(
f"/api/rfcs/ohm/branches/{branch}/changes/{change_id}/accept",
json={
"proposed": "Open Human Model is a framework for representing humans in software systems.",
"was_edited_before_accept": False,
},
)
assert r.status_code == 200, r.text
body = r.json()
assert body["commit_sha"]
# The change row is now accepted with the commit sha bound.
row = db.conn().execute(
"SELECT state, commit_sha, acted_by, was_edited_before_accept FROM changes WHERE id = ?",
(change_id,),
).fetchone()
assert row["state"] == "accepted"
assert row["commit_sha"] == body["commit_sha"]
assert row["acted_by"] == 2
assert not row["was_edited_before_accept"]
# The branch's RFC.md on Gitea now reflects the change.
owner, repo = "wiggleverse", "rfc-0001-ohm"
new_body = fake.files[(owner, repo, branch, "RFC.md")]["content"]
assert "in software systems" in new_body
def test_accept_with_edit_before_accept_records_flag_and_ai_original(app_with_fake_gitea):
from fastapi.testclient import TestClient
from app import db
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)
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"]
cur = db.conn().execute(
"""
INSERT INTO changes (rfc_slug, branch_name, thread_id, kind, state,
original, proposed, reason)
VALUES ('ohm', ?, ?, 'ai', 'pending', ?, ?, ?)
""",
(branch, thread_id,
"It defines consent, trait, and agency in compatible terms.",
"It defines consent, trait, harm, and agency in compatible terms.",
"adds harm"),
)
change_id = cur.lastrowid
edited = "It defines consent, trait, harm, and agency together."
r = client.post(
f"/api/rfcs/ohm/branches/{branch}/changes/{change_id}/accept",
json={"proposed": edited, "was_edited_before_accept": True},
)
assert r.status_code == 200, r.text
row = db.conn().execute(
"SELECT proposed, was_edited_before_accept FROM changes WHERE id = ?",
(change_id,),
).fetchone()
assert row["was_edited_before_accept"] == 1
assert row["proposed"] == edited
# The commit body carries both the AI's original proposed
# text and the contributor's revision per §8.9.
body = fake.files[("wiggleverse", "rfc-0001-ohm", branch, "RFC.md")]["content"]
assert "harm" in body
# The contributor's edited text won, not the AI's.
assert "together." in body
def test_decline_change_persists_as_evidence_no_commit(app_with_fake_gitea):
from fastapi.testclient import TestClient
from app import db
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)
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"]
cur = db.conn().execute(
"""
INSERT INTO changes (rfc_slug, branch_name, thread_id, kind, state,
original, proposed, reason)
VALUES ('ohm', ?, ?, 'ai', 'pending', 'x', 'y', 'why')
""",
(branch, thread_id),
)
change_id = cur.lastrowid
prior_sha = fake.branches[("wiggleverse", "rfc-0001-ohm")][branch]["sha"]
r = client.post(f"/api/rfcs/ohm/branches/{branch}/changes/{change_id}/decline")
assert r.status_code == 200, r.text
# No commit, no body change.
post_sha = fake.branches[("wiggleverse", "rfc-0001-ohm")][branch]["sha"]
assert prior_sha == post_sha
# The card stays as evidence.
row = db.conn().execute(
"SELECT state FROM changes WHERE id = ?", (change_id,)
).fetchone()
assert row["state"] == "declined"
def test_manual_flush_commits_and_drops_system_message(app_with_fake_gitea):
from fastapi.testclient import TestClient
from app import db
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)
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"]
new_body = SEED_BODY + "\n\nA new paragraph.\n"
r = client.post(
f"/api/rfcs/ohm/branches/{branch}/manual-flush",
json={"new_content": new_body, "paragraph_count": 1},
)
assert r.status_code == 200, r.text
assert r.json()["commit_sha"]
# The branch RFC.md was updated.
body = fake.files[("wiggleverse", "rfc-0001-ohm", branch, "RFC.md")]["content"]
assert "A new paragraph" in body
# Per §10.6: a system-author message landed in the branch chat.
rows = db.conn().execute(
"""
SELECT m.role, m.text FROM thread_messages m
JOIN threads t ON t.id = m.thread_id
WHERE t.rfc_slug = 'ohm' AND t.branch_name = ?
""",
(branch,),
).fetchall()
assert any(r["role"] == "system" and "manual edit" in r["text"] for r in rows)
def test_create_flag_thread_surfaces_on_branch_view(app_with_fake_gitea):
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)
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"]
r = client.post(
f"/api/rfcs/ohm/branches/{branch}/threads",
json={
"thread_kind": "flag",
"anchor_kind": "range",
"anchor_payload": {"quote": "consent"},
"label": "needs an example",
},
)
assert r.status_code == 200, r.text
thread_id = r.json()["thread_id"]
r = client.get(f"/api/rfcs/ohm/branches/{branch}")
threads = r.json()["threads"]
assert any(t["id"] == thread_id and t["thread_kind"] == "flag" for t in threads)
def test_visibility_flip_locks_out_non_grantees(app_with_fake_gitea):
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")
provision_user_row(user_id=3, login="bob", role="contributor")
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
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"]
# Flip the branch private.
r = client.post(
f"/api/rfcs/ohm/branches/{branch}/visibility",
json={"read_public": False},
)
assert r.status_code == 200, r.text
# Bob (a different contributor) is now blocked from reading it.
sign_in_as(client, user_id=3, gitea_login="bob", display_name="Bob", role="contributor")
r = client.get(f"/api/rfcs/ohm/branches/{branch}")
assert r.status_code == 403
# Alice (the creator) still can.
sign_in_as(client, user_id=2, gitea_login="alice", display_name="Alice", role="contributor")
r = client.get(f"/api/rfcs/ohm/branches/{branch}")
assert r.status_code == 200
def test_anonymous_can_read_main_but_not_contribute(app_with_fake_gitea):
from fastapi.testclient import TestClient
app, fake = app_with_fake_gitea
with TestClient(app) as client:
seed_active_rfc(fake, slug="ohm", title="OHM", body=SEED_BODY)
# No sign-in.
r = client.get("/api/rfcs/ohm/main")
assert r.status_code == 200
r = client.post("/api/rfcs/ohm/branches/main/promote-to-branch", json={})
assert r.status_code == 401
def test_stale_change_refuses_silent_apply(app_with_fake_gitea):
from fastapi.testclient import TestClient
from app import db
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)
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"]
# Stale by construction: original text not in the document.
cur = db.conn().execute(
"""
INSERT INTO changes (rfc_slug, branch_name, thread_id, kind, state,
original, proposed, reason)
VALUES ('ohm', ?, ?, 'ai', 'pending', ?, ?, ?)
""",
(branch, thread_id, "Text that does not appear", "Replacement.", "test"),
)
change_id = cur.lastrowid
# Refused without force.
r = client.post(
f"/api/rfcs/ohm/branches/{branch}/changes/{change_id}/accept",
json={"proposed": "Replacement.", "was_edited_before_accept": False},
)
assert r.status_code == 409
# The row is marked stale per §8.11.
row = db.conn().execute(
"SELECT state, stale_since FROM changes WHERE id = ?", (change_id,)
).fetchone()
assert row["state"] == "pending"
assert row["stale_since"]
# Force-apply succeeds and appends.
r = client.post(
f"/api/rfcs/ohm/branches/{branch}/changes/{change_id}/accept",
json={"proposed": "Replacement.", "was_edited_before_accept": False, "force_apply_stale": True},
)
assert r.status_code == 200, r.text
# ---------------------------------------------------------------------------
# Chat streaming with a fake provider
# ---------------------------------------------------------------------------
class FakeProvider:
name = "claude"
display_name = "Claude"
def __init__(self, fixed_response: str):
self._response = fixed_response
def send(self, system, history):
return self._response
def send_streaming(self, system, history):
# Single-chunk stream — sufficient for the orchestration test.
yield self._response
def test_chat_turn_materializes_change_from_change_block(app_with_fake_gitea):
from fastapi.testclient import TestClient
from app import db
app, fake = app_with_fake_gitea
fake_response = (
"Here is a tightening:\n\n"
"<change>\n"
"<original>Open Human Model is a framework for representing humans.</original>\n"
"<proposed>Open Human Model is a framework for representing humans across software systems.</proposed>\n"
"<reason>scopes the framework</reason>\n"
"</change>\n\n"
"Let me know if that fits."
)
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)
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"]
# Inject the fake provider — the app's `providers` dict is built
# at startup; we replace it for the test so the chat endpoint
# resolves a deterministic response.
app.state.providers["claude"] = FakeProvider(fake_response)
# The router resolved `providers` at construction time; rebuild
# the slice 2 router with the fake provider in place.
from app import api as api_routes
# Find and replace the existing branches router. Simpler: monkey
# patch the providers dict referenced by the router closure.
# The closure receives the dict by reference, so mutating it
# propagates.
# (Above mutation already does that — nothing more to do.)
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": "Can you tighten the opening?", "model": "claude"},
)
assert r.status_code == 200, r.text
# Drain the stream so the orchestrator finishes its work.
body = r.content.decode()
assert "DONE" in body
# A change row materialized from the <change> block.
rows = db.conn().execute(
"SELECT kind, state, original, proposed, reason FROM changes WHERE rfc_slug = 'ohm' AND branch_name = ?",
(branch,),
).fetchall()
ai_rows = [r for r in rows if r["kind"] == "ai"]
assert len(ai_rows) == 1
assert ai_rows[0]["state"] == "pending"
assert "humans across software systems" in ai_rows[0]["proposed"]
assert "scopes the framework" in ai_rows[0]["reason"]
# The assistant message persisted with the full text.
msgs = db.conn().execute(
"SELECT role, text FROM thread_messages WHERE thread_id = ? ORDER BY id",
(thread_id,),
).fetchall()
assert msgs[-1]["role"] == "assistant"
assert "<change>" in msgs[-1]["text"]