feat(family): platforms declared at inception and detected from bound repos (milestone 463 step 2, #4988)
CI & Build / Python lint (push) Successful in 4s
CI & Build / Plugin hooks (push) Successful in 21s
CI & Build / TypeScript typecheck (push) Successful in 1m10s
CI & Build / integration (push) Successful in 1m48s
CI & Build / Python tests (push) Successful in 2m40s
CI & Build / Build & push image (push) Failing after 44s

A project's platforms decide which family ideas reach it. This step makes membership answerable from every door:

- services/platforms.py: the global catalog (writes are admin-only and duplicate-gated by slug); pure marker detection; and membership reads and writes. Detection only ADDS, and only where nobody has answered. It never overrides a declared or rejected row and never removes one.
- coverage: the archive scan now carries every path, and the refresh runs detection fail-open.
- inception: a platforms choice (slugs, or null for unanswered). The list is the whole answer: members left out of it become rejected.
- MCP: list_platforms and set_project_platforms; enter_project and get_project carry the project's platforms.
- REST: /api/platforms (admin writes) and /api/projects/<id>/platforms.
- UI: a platforms checklist on the inception card, a Family tab on ProjectView, and a Platforms admin tab in Settings.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
2026-10-06 09:51:28 -04:00
co-authored by Claude Opus 5.5
parent e68ccc5884
commit 07d2542479
27 changed files with 1638 additions and 51 deletions
+15
View File
@@ -512,3 +512,18 @@ def skill_text(name: str) -> str:
folder = pathlib.Path(__file__).resolve().parents[1] / "plugin" / "skills" / name
refs = sorted(p for p in folder.glob("*.md") if p.name != "SKILL.md")
return "\n\n".join(p.read_text() for p in [folder / "SKILL.md", *refs])
def forge_tarball(files: dict[str, bytes], top: str = "widget") -> bytes:
"""A gzipped tar shaped like a forge archive: every file under one
top-level directory (repo-ref/), which the scan strips."""
import io
import tarfile
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
for path, data in files.items():
info = tarfile.TarInfo(f"{top}/{path}")
info.size = len(data)
tar.addfile(info, io.BytesIO(data))
return buf.getvalue()
+21 -3
View File
@@ -27,7 +27,7 @@ def test_project_carries_an_inception_record_and_to_dict_shows_it():
def test_validate_inception_pins_the_choice_vocabulary():
assert CHOICE_KEYS == ("design_system_id", "seed_systems")
assert CHOICE_KEYS == ("design_system_id", "seed_systems", "platforms")
assert validate_inception({}) is None
assert validate_inception({"design_system_id": 3, "seed_systems": True}) is None
assert validate_inception({"design_system_id": None}) is None
@@ -46,10 +46,28 @@ def test_validate_inception_pins_the_choice_vocabulary():
assert "true or false" in validate_inception({"seed_systems": "yes"})
def test_validate_inception_takes_platforms_as_slugs_or_unstated():
"""Slugs, not ids: an inception record outlives a restore, and a slug is
what both sides of a restore agree on. None is "not answered", which is
different from [] — "none of them"."""
assert validate_inception({"platforms": ["android-app", "go"]}) is None
assert validate_inception({"platforms": []}) is None
assert validate_inception({"platforms": None}) is None
assert "platform slugs" in validate_inception({"platforms": "android-app"})
assert "platform slugs" in validate_inception({"platforms": [3]})
assert "platform slugs" in validate_inception({"platforms": [" "]})
def test_normalize_choices_is_canonical_and_complete():
out = normalize_choices({"design_system_id": 4})
assert out == {"design_system_id": 4, "seed_systems": False}
assert normalize_choices(None) == {"design_system_id": None, "seed_systems": False}
assert out == {"design_system_id": 4, "seed_systems": False, "platforms": None}
assert normalize_choices(None) == {
"design_system_id": None, "seed_systems": False, "platforms": None,
}
# Sorted and de-duplicated, so two equal answers record identically.
assert normalize_choices({"platforms": ["go", " android-app", "go"]})["platforms"] == [
"android-app", "go",
]
def test_standard_systems_vocabulary_reads_the_catalog_not_a_constant():
+7 -3
View File
@@ -37,14 +37,16 @@ async def seeded():
async def test_decide_applies_every_effect_and_records_last(seeded):
owner, pid = seeded["owner"], seeded["pid"]
defaults = await inception_svc.current_defaults(owner, pid)
assert set(defaults) == {"design_system_id", "design_systems", "systems"}
assert set(defaults) == {
"design_system_id", "design_systems", "systems", "platforms", "project_platforms",
}
assert defaults["systems"] == 0 and defaults["design_system_id"] is None
out = await inception_svc.decide(owner, pid, via="mcp", choices={
"design_system_id": None,
"seed_systems": True,
})
assert set(out["effects"]) == {"design_system_id", "systems_seeded"}
assert set(out["effects"]) == {"design_system_id", "systems_seeded", "platforms"}
catalog = await canonical_svc.list_canonical_systems()
assert len(out["effects"]["systems_seeded"]) == len(catalog)
# Seeded Systems come out mapped, not needing a later reconciliation.
@@ -56,7 +58,9 @@ async def test_decide_applies_every_effect_and_records_last(seeded):
project = await s.get(Project, pid)
assert inception_svc.is_decided(project)
assert project.inception["via"] == "mcp" and project.inception["decided_by"] == owner
assert project.inception["choices"] == {"design_system_id": None, "seed_systems": True}
assert project.inception["choices"] == {
"design_system_id": None, "seed_systems": True, "platforms": None,
}
# Re-deciding with seed again mints nothing twice.
again = await inception_svc.decide(owner, pid, via="ui", choices={"seed_systems": True})
assert again["effects"]["systems_seeded"] == []
+134
View File
@@ -0,0 +1,134 @@
"""Real-Postgres checks for platform membership (milestone 463 step 2).
The one invariant: detection only ever ADDS, and only where nobody has
answered. A refresh that overwrote a "no" would put a project back into a
family it was taken out of; one that overwrote a "yes" would erase who said
it. Each is shown here against the real writer, beside the person-facing
writes it must defer to.
"""
import pytest
import pytest_asyncio
from sqlalchemy import select
from scribe.models import async_session
from scribe.models.family import Platform
from scribe.models.project import Project
from scribe.services import inception as inception_svc
from scribe.services import platforms as platforms_svc
from tests.helpers import ensure_user
pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
OWNER = "platforms_owner"
async def _id(slug: str) -> int:
async with async_session() as s:
return await s.scalar(
select(Platform.id).where(Platform.slug == slug, Platform.deleted_at.is_(None))
)
@pytest_asyncio.fixture
async def seeded():
"""An owner, an outsider and a fresh project. Each test gets its own
project, so answers never leak between tests through a shared row."""
async with async_session() as s:
owner = await ensure_user(s, OWNER)
outsider = await ensure_user(s, "platforms_outsider")
project = Project(user_id=owner.id, title="an app with a server")
s.add(project)
await s.flush()
ids = {"owner": owner.id, "outsider": outsider.id, "pid": project.id}
await s.commit()
return ids
def _states(rows: list[dict]) -> dict[str, str]:
return {r["slug"]: r["state"] for r in rows}
async def test_detection_adds_only_where_nobody_has_answered(seeded):
owner, pid = seeded["owner"], seeded["pid"]
await platforms_svc.set_project_platforms(owner, pid, {
"go": "declared", "rust": "rejected",
})
added = await platforms_svc.detect_for_project(pid, [
"server/go.mod", "core/Cargo.toml", "app/src/main/AndroidManifest.xml",
])
# Only android was unanswered; go and rust keep the person's answers.
assert added == [await _id("android-app")]
states = _states(await platforms_svc.project_platforms(owner, pid))
assert states == {"go": "declared", "rust": "rejected", "android-app": "detected"}
async def test_detection_never_removes_a_platform_whose_marker_is_gone(seeded):
owner, pid = seeded["owner"], seeded["pid"]
await platforms_svc.detect_for_project(pid, ["server/go.mod"])
await platforms_svc.detect_for_project(pid, ["README.md"])
assert _states(await platforms_svc.project_platforms(owner, pid)) == {"go": "detected"}
async def test_a_withdrawn_answer_lets_detection_decide_again(seeded):
owner, pid = seeded["owner"], seeded["pid"]
await platforms_svc.set_project_platforms(owner, pid, {"go": "rejected"})
assert await platforms_svc.detect_for_project(pid, ["go.mod"]) == []
await platforms_svc.set_project_platforms(owner, pid, {"go": None})
assert await platforms_svc.detect_for_project(pid, ["go.mod"]) == [await _id("go")]
async def test_a_bad_update_writes_nothing(seeded):
owner, pid = seeded["owner"], seeded["pid"]
with pytest.raises(ValueError, match="unknown platform"):
await platforms_svc.set_project_platforms(owner, pid, {
"go": "declared", "not-a-platform": "declared",
})
assert await platforms_svc.project_platforms(owner, pid) == []
with pytest.raises(ValueError, match="no write access"):
await platforms_svc.set_project_platforms(seeded["outsider"], pid, {"go": "declared"})
assert await platforms_svc.project_platforms(seeded["outsider"], pid) is None
async def test_inception_records_slugs_and_rejects_what_it_leaves_out(seeded):
"""The inception list is the WHOLE answer. Detection found Android and
Go; the person says the project is Go and Python — so Android becomes a
recorded "no", and the next refresh cannot bring it back."""
owner, pid = seeded["owner"], seeded["pid"]
await platforms_svc.detect_for_project(pid, ["app/AndroidManifest.xml", "go.mod"])
defaults = await inception_svc.current_defaults(owner, pid)
assert {"android-app", "go"} <= {p["slug"] for p in defaults["project_platforms"]}
out = await inception_svc.decide(owner, pid, via="ui", choices={
"seed_systems": False, "platforms": ["python", "go"],
})
assert _states(out["effects"]["platforms"]) == {
"android-app": "rejected", "go": "declared", "python": "declared",
}
async with async_session() as s:
project = await s.get(Project, pid)
assert project.inception["choices"]["platforms"] == ["go", "python"]
assert await platforms_svc.detect_for_project(pid, ["app/AndroidManifest.xml"]) == []
async def test_an_unknown_inception_platform_applies_nothing(seeded):
owner, pid = seeded["owner"], seeded["pid"]
with pytest.raises(ValueError, match="unknown platform"):
await inception_svc.decide(owner, pid, via="mcp", choices={
"seed_systems": True, "platforms": ["go", "cobol-mainframe"],
})
assert await platforms_svc.project_platforms(owner, pid) == []
async def test_the_catalog_is_admin_written_and_duplicate_gated():
async with async_session() as s:
admin = await ensure_user(s, "platforms_admin", role="admin")
user = await ensure_user(s, OWNER)
await s.commit()
admin_id, user_id = admin.id, user.id
assert await platforms_svc.create_platform(user_id, "Kotlin Multiplatform") is None
# "Android App" reduces to the seeded android-app: the existing one comes
# back rather than a second spelling of it.
dup = await platforms_svc.create_platform(admin_id, "Android App")
assert dup["duplicate"] and dup["existing_id"] == await _id("android-app")
with pytest.raises(ValueError, match="repo-relative"):
await platforms_svc.create_platform(admin_id, "Odd", markers=["/abs"])
+12
View File
@@ -25,6 +25,18 @@ def _no_systems():
yield
@pytest.fixture(autouse=True)
def _no_platforms():
"""enter_project and get_project carry the project's platforms (milestone
463). No database in this lane, so the lookup is stubbed to the common
case — a project with none yet. The populated shape is asserted in its
own test.
"""
with patch("scribe.mcp.tools.projects.platforms_svc.project_platforms",
AsyncMock(return_value=[])):
yield
@pytest.fixture(autouse=True)
def _no_coverage():
"""enter_project also reads the pattern-coverage cache (#2692) — same
+13
View File
@@ -24,6 +24,19 @@ PLAN = "A plan paragraph long enough to matter. " * 125 # ~5k chars
GOAL = "What the project is for. " * 50 # ~1.2k chars
@pytest.fixture(autouse=True)
def _no_platforms():
"""enter_project and get_project carry the project's platforms (milestone
463). No database in this lane, so the lookup is stubbed to the common
case — a project with none yet. The populated shape is asserted in its
own test.
"""
with patch("scribe.mcp.tools.projects.platforms_svc.project_platforms",
AsyncMock(return_value=[])):
yield
def _milestone(mid: int, status: str, touched_day: int) -> dict:
"""A summary row as get_project_milestone_summary returns it."""
touched = f"2026-08-{touched_day:02d}T00:00:00+00:00"
+2 -11
View File
@@ -8,11 +8,9 @@ deliberately reuse the definitions test_write_path_trigger stages for the
hook; extending one detector means extending both, and this comment is the
tripwire.
"""
import io
import json
import shutil
import subprocess
import tarfile
from pathlib import Path
import pytest
@@ -29,7 +27,7 @@ from scribe.services.coverage import (
shapes_from_archive,
)
from scribe.services.shape_ledger import location_covers
from tests.helpers import ensure_user
from tests.helpers import ensure_user, forge_tarball
PLUGIN = Path(__file__).resolve().parents[1] / "plugin"
@@ -173,14 +171,7 @@ def test_scannable_gates_prose_vendored_and_sourcemaps():
# --- unit: reading shapes out of a forge tarball -----------------------------
def _tarball(files: dict[str, bytes], top: str = "widget") -> bytes:
buf = io.BytesIO()
with tarfile.open(fileobj=buf, mode="w:gz") as tar:
for path, data in files.items():
info = tarfile.TarInfo(f"{top}/{path}")
info.size = len(data)
tar.addfile(info, io.BytesIO(data))
return buf.getvalue()
_tarball = forge_tarball # shared with tests/test_platforms.py
TREE = {
+145
View File
@@ -0,0 +1,145 @@
"""Platforms without a database (milestone 463 step 2).
Detection is pure — a catalog and a list of repo paths in, platform ids out —
so it is pinned here against fixture trees. What it writes, and what it must
never overwrite, is in tests/test_integration_platforms.py.
"""
from __future__ import annotations
from types import SimpleNamespace
from scribe.mcp.server import _READ_ONLY_TOOLS, _WRITE_TOOLS
from scribe.mcp.tools.projects import _inception_choices
from scribe.services.coverage import scan_archive
from tests.helpers import forge_tarball
from scribe.services.platforms import (
MEMBER_STATES, SETTABLE_STATES, detect, marker_matches, members,
validate_markers, validate_updates,
)
def _platform(pid: int, *markers: str) -> SimpleNamespace:
return SimpleNamespace(id=pid, markers=list(markers))
# --- a single marker ----------------------------------------------------------
def test_a_bare_marker_matches_the_basename_anywhere_in_the_tree():
assert marker_matches("go.mod", "go.mod")
assert marker_matches("go.mod", "services/api/go.mod")
assert marker_matches("AndroidManifest.xml", "app/src/main/AndroidManifest.xml")
assert marker_matches("vite.config.*", "frontend/vite.config.ts")
assert not marker_matches("go.mod", "go.mod.bak")
assert not marker_matches("go.mod", "docs/go.mod.md")
def test_a_marker_with_a_slash_matches_the_whole_path_only():
"""A directory-shaped marker cannot be satisfied by a same-named file
somewhere else — `.github/workflows/*` is about the repo root's CI, not
a vendored copy of some other project's."""
assert marker_matches(".github/workflows/*", ".github/workflows/ci.yml")
assert not marker_matches(".github/workflows/*", "vendor/x/.github/workflows/ci.yml")
assert not marker_matches(".github/workflows/*", "ci.yml")
def test_a_blank_marker_matches_nothing():
assert not marker_matches("", "anything")
assert not marker_matches(" ", "anything")
# --- detection over a tree -------------------------------------------------------
ANDROID_AND_GO = [
"app/build.gradle.kts",
"app/src/main/AndroidManifest.xml",
"server/go.mod",
"server/main.go",
"README.md",
]
def test_detect_finds_every_platform_a_tree_carries_and_nothing_else():
catalog = [
_platform(1, "AndroidManifest.xml"),
_platform(2, "go.mod"),
_platform(3, "Cargo.toml"),
_platform(4, ".github/workflows/*"),
]
assert detect(catalog, ANDROID_AND_GO) == [1, 2]
def test_a_platform_with_no_markers_is_declare_only():
"""An empty marker list is how a platform says "a person must tell you" —
it must not match everything, and it must not match nothing by error."""
catalog = [_platform(1), _platform(2, "go.mod")]
assert detect(catalog, ANDROID_AND_GO) == [2]
assert detect([SimpleNamespace(id=9, markers=None)], ANDROID_AND_GO) == []
def test_detect_on_an_empty_tree_finds_nothing():
assert detect([_platform(1, "go.mod")], []) == []
def test_the_archive_scan_carries_every_path_not_only_the_scannable_ones():
"""The markers that say what a project IS are mostly files the shape scan
skips — a manifest, a go.mod, a gradle script. Detection reads the scan's
paths, so a scan that listed only code files would detect nothing."""
scan = scan_archive(forge_tarball({
"app/src/main/AndroidManifest.xml": b"<manifest/>",
"server/go.mod": b"module x\n",
"server/main.py": b"def main():\n pass\n",
}))
assert set(scan.paths) == {
"app/src/main/AndroidManifest.xml", "server/go.mod", "server/main.py",
}
# The forge's wrapping directory is stripped, as it is for shapes.
assert not any(p.startswith("widget/") for p in scan.paths)
# --- the answers a person may give -------------------------------------------------
def test_detected_is_the_refreshs_to_write_never_a_persons():
assert "detected" in MEMBER_STATES
assert "detected" not in SETTABLE_STATES
assert "rejected" not in MEMBER_STATES
def test_validate_updates():
assert validate_updates({"go": "declared", "rust": "rejected", "python": None}) is None
assert validate_updates({}) is None
assert "slug: state" in validate_updates(["go"])
assert "not a state a person sets" in validate_updates({"go": "detected"})
assert "not a state a person sets" in validate_updates({"go": "maybe"})
assert "named by its slug" in validate_updates({"": "declared"})
def test_validate_markers():
assert validate_markers(None) is None
assert validate_markers(["go.mod", ".github/workflows/*"]) is None
assert "list of glob patterns" in validate_markers("go.mod")
assert "list of glob patterns" in validate_markers([1])
assert "repo-relative" in validate_markers(["/etc/passwd"])
def test_members_drops_the_rejected_answers():
rows = [
{"id": 1, "slug": "go", "name": "Go", "state": "declared"},
{"id": 2, "slug": "rust", "name": "Rust", "state": "rejected"},
{"id": 3, "slug": "python", "name": "Python", "state": "detected"},
]
assert [m["slug"] for m in members(rows)] == ["go", "python"]
# --- the doors ------------------------------------------------------------------------
def test_the_platform_tools_are_classified():
assert "list_platforms" in _READ_ONLY_TOOLS
assert "set_project_platforms" in _WRITE_TOOLS
def test_inception_choices_carry_platforms_only_when_given():
"""Left out, platforms stays UNSTATED (None after normalising) — which is
not the same answer as an empty list, "none of them"."""
assert "platforms" not in _inception_choices(0, True)
assert _inception_choices(0, True, ["go"])["platforms"] == ["go"]
assert _inception_choices(0, True, [])["platforms"] == []
+28
View File
@@ -0,0 +1,28 @@
"""Structural tests for the platforms blueprint (milestone 463 step 2) —
registration and the route-to-service contract. What the writes do is in
tests/test_integration_platforms.py."""
import inspect
def test_platforms_blueprint_registered_in_app():
from scribe.app import create_app
app = create_app()
assert "platforms" in app.blueprints
rules = {(r.rule, m) for r in app.url_map.iter_rules() for m in r.methods}
for rule, method in (
("/api/platforms", "GET"),
("/api/platforms", "POST"),
("/api/platforms/<int:platform_id>", "PATCH"),
("/api/projects/<int:project_id>/platforms", "GET"),
("/api/projects/<int:project_id>/platforms", "PUT"),
):
assert (rule, method) in rules, f"{method} {rule} is not routed"
def test_writes_to_the_catalog_and_to_a_project_take_the_caller():
"""The catalog's admin gate and the project's write gate both live in the
service, so every write must be handed the caller to check."""
from scribe.services import platforms as svc
for name in ("create_platform", "update_platform", "set_project_platforms",
"project_platforms"):
assert "user_id" in inspect.signature(getattr(svc, name)).parameters