feat(family): platforms, family ideas, the adoption ledger and its decision log (milestone 463 step 1, #4987)

When one project solves something every project on the same platform will
meet, that solution becomes family canon and every other project on the
platform answers it. This is the storage for that.

- platforms: a global catalog in the canonical_systems shape, seeded with
  generic technology names and the file markers step 2's detection reads.
- project_platforms: declared / detected / rejected. A rejected row is kept
  so detection cannot re-add what a person said no to.
- family_ideas: a note's family state. No new record type; any note, snippet
  or lesson becomes an idea. A canon idea must state when it applies (CHECK).
- family_idea_platforms: the only scope source. A linked rule topic takes its
  scope from the idea, so the two cannot disagree.
- family_idea_references: reference implementations, explicit not inferred.
- family_adoptions: one answer per (project, idea). Variant and exempt require
  a reason (CHECK). Recheck is derived from the two canon versions, never
  stored.
- family_decisions: the append-only log, with a required reason and the
  earlier decisions each one followed. The agent decides with no approval
  step, so precedent is what keeps its calls consistent.

Backup v25 carries all seven: platforms by slug, precedent ids remapped
through the decision map. Both column guards cover the new tables, and a
real-Postgres test exercises the CHECKs and the restore remaps.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/test_family_models.py

@@ -0,0 +1,69 @@
+"""Family canon's schema — the parts that need no database (milestone 463 step 1).
+
+The constraints themselves, against real Postgres, are in
+tests/test_integration_family_canon.py. These pin that the migration and the
+model agree on every whitelist (rule 36: a value one side accepts and the
+other refuses only fails at INSERT time, which CI's unit lane never reaches),
+and that the seed ships nothing an install other than this one would find
+foreign (rule 115).
+"""
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+from scribe.models import family
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+def _migration():
+    path = ROOT / "alembic" / "versions" / "0120_family_canon.py"
+    spec = importlib.util.spec_from_file_location("m0120", path)
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_the_migration_checks_and_the_model_agree():
+    m = _migration()
+    assert tuple(m._MEMBERSHIP) == family.MEMBERSHIP_STATES
+    assert tuple(m._IDEA_STATUSES) == family.IDEA_STATUSES
+    assert tuple(m._ADOPTION_STATUSES) == family.ADOPTION_STATUSES
+    assert tuple(m._REASONED) == family.REASONED_STATUSES
+    assert tuple(m._ACTIONS) == family.DECISION_ACTIONS
+    assert tuple(m._DECIDERS) == family.DECIDERS
+
+
+def test_the_departures_that_need_a_reason_are_adoption_states():
+    assert set(family.REASONED_STATUSES) <= set(family.ADOPTION_STATUSES)
+
+
+def test_the_seed_is_generic_and_unique():
+    seed = _migration()._SEED
+    slugs = [slug for _, slug, _, _ in seed]
+    assert len(slugs) == len(set(slugs)), "a duplicate slug breaks the live-unique index"
+    for name, slug, description, markers in seed:
+        assert name and slug and description
+        assert slug == slug.lower() and " " not in slug
+        # A list, possibly empty — never None. Detection reads it directly.
+        assert isinstance(markers, list)
+
+
+def test_the_column_defaults_match_the_first_state_of_each_lifecycle():
+    """A row created with no state given starts where its lifecycle starts.
+    An idea is a candidate until promoted; an answer is unassessed until
+    someone assesses it — never silently `adopted`."""
+    assert family.FamilyIdea.__table__.c.status.server_default.arg == "candidate"
+    assert family.FamilyAdoption.__table__.c.status.server_default.arg == "unassessed"
+    assert family.ProjectPlatform.__table__.c.state.server_default.arg == "declared"
+
+
+def test_a_canon_ideas_scope_lives_on_the_idea_and_nowhere_else():
+    """One scope source. A rule topic is linked FROM the idea and carries no
+    platforms of its own, so the two cannot disagree about who a standard
+    reaches (#4221's two-sources-of-truth shape)."""
+    from scribe.models.rulebook import RulebookTopic
+
+    assert "topic_id" in family.FamilyIdea.__table__.c
+    assert not {c.name for c in RulebookTopic.__table__.c} & {"platform_id", "platform_ids"}

tests/test_integration_family_canon.py

@@ -0,0 +1,304 @@
+"""Real-Postgres checks for family canon's schema (milestone 463 step 1).
+
+Two kinds of claim live here, and both need a database:
+
+1. THE CONSTRAINTS. The agent assesses and promotes with no approval step, so
+   the few things that must always hold are held by the schema rather than by
+   prose a session might skip: a canon idea states when it applies, a variant
+   or an exemption says why, a decision has a reason. Each is shown refusing
+   the bad row, and each beside a good row so the refusal is not a fixture
+   that fails for some other reason.
+
+2. THE RESTORE REMAPS. Two seams can come back plausible and wrong:
+   - platforms travel by SLUG and must land on the destination's own seeded
+     rows, not create duplicates of them;
+   - `precedent_ids` is a list of ids INSIDE JSON. A restore that copied it
+     raw would leave each decision pointing at whatever took the old number —
+     populated, plausible, and about the wrong decision.
+"""
+import pytest
+import pytest_asyncio
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+
+from scribe.models import async_session
+from scribe.models.family import (
+    FamilyAdoption, FamilyDecision, FamilyIdea, FamilyIdeaPlatform,
+    FamilyIdeaReference, Platform, ProjectPlatform,
+)
+from scribe.models.note import Note
+from scribe.models.project import Project
+from scribe.models.user import User
+from scribe.services import backup
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+OWNER_USERNAME = "family_canon_owner"
+RESTORED_USERNAME = "family_canon_restored"
+
+
+async def _purge(username: str) -> None:
+    """user -> project / note is ON DELETE CASCADE, and every family table
+    cascades from a project or a note, so dropping the user's rows clears
+    everything this file made. Platforms are global and never created here
+    beyond the migration's seed, so there is nothing of theirs to clear."""
+    async with async_session() as s:
+        for user in (await s.execute(
+            select(User).where(User.username == username)
+        )).scalars().all():
+            for note in (await s.execute(
+                select(Note).where(Note.user_id == user.id)
+            )).scalars().all():
+                await s.delete(note)
+            for project in (await s.execute(
+                select(Project).where(Project.user_id == user.id)
+            )).scalars().all():
+                await s.delete(project)
+            if username == RESTORED_USERNAME:
+                await s.delete(user)
+        await s.commit()
+
+
+@pytest_asyncio.fixture(autouse=True)
+async def _no_leftovers():
+    """SETUP ONLY — a database call after a `yield` in an autouse fixture
+    orphans a pooled connection (see the backup round-trip siblings)."""
+    await _purge(RESTORED_USERNAME)
+    await _purge(OWNER_USERNAME)
+
+
+async def _platform(slug: str) -> Platform:
+    async with async_session() as s:
+        return (await s.execute(
+            select(Platform).where(Platform.slug == slug, Platform.deleted_at.is_(None))
+        )).scalars().one()
+
+
+async def _owner_project_and_note() -> tuple[int, int, int]:
+    async with async_session() as s:
+        owner = await ensure_user(s, OWNER_USERNAME)
+        project = Project(user_id=owner.id, title="an app on a platform")
+        note = Note(user_id=owner.id, title="an idea", body="the idea")
+        s.add_all([project, note])
+        await s.commit()
+        return owner.id, project.id, note.id
+
+
+async def _refused(*rows) -> bool:
+    async with async_session() as s:
+        s.add_all(list(rows))
+        try:
+            await s.commit()
+        except IntegrityError:
+            await s.rollback()
+            return True
+    return False
+
+
+# --- the seed ---------------------------------------------------------------
+
+async def test_the_migration_seeds_generic_platforms_with_their_markers():
+    android = await _platform("android-app")
+    assert "AndroidManifest.xml" in android.markers
+    # Declare-only platforms carry an empty list, never NULL: detection reads
+    # the list and must not have to special-case a missing one.
+    assert (await _platform("postgresql")).markers == []
+
+
+# --- the constraints ----------------------------------------------------------
+
+async def test_a_canon_idea_must_say_when_it_applies():
+    _, _, note_id = await _owner_project_and_note()
+    assert await _refused(FamilyIdea(note_id=note_id, status="canon", applies_when="  "))
+    # A candidate may be vague — that is what a candidate is.
+    assert not await _refused(FamilyIdea(note_id=note_id, status="candidate"))
+
+
+async def test_a_variant_or_an_exemption_must_say_why():
+    _, project_id, note_id = await _owner_project_and_note()
+    assert not await _refused(FamilyIdea(
+        note_id=note_id, status="canon",
+        applies_when="any app that installs its own updates",
+    ))
+    for status in ("variant", "exempt"):
+        assert await _refused(FamilyAdoption(
+            project_id=project_id, idea_id=note_id, status=status, reason=" ",
+        )), f"{status} with a blank reason was accepted"
+    # Owed and adopted need no reason: neither is a departure.
+    assert not await _refused(FamilyAdoption(
+        project_id=project_id, idea_id=note_id, status="owed", decided_via="agent",
+    ))
+
+
+async def test_an_unknown_state_is_refused():
+    _, project_id, note_id = await _owner_project_and_note()
+    assert not await _refused(FamilyIdea(note_id=note_id))
+    android = await _platform("android-app")
+    assert await _refused(ProjectPlatform(
+        project_id=project_id, platform_id=android.id, state="maybe",
+    ))
+    assert await _refused(FamilyAdoption(
+        project_id=project_id, idea_id=note_id, status="declined",
+    ))
+
+
+async def test_a_decision_must_carry_a_reason():
+    _, _, note_id = await _owner_project_and_note()
+    assert not await _refused(FamilyIdea(note_id=note_id))
+    assert await _refused(FamilyDecision(idea_id=note_id, action="propose", reason="   "))
+    assert not await _refused(FamilyDecision(
+        idea_id=note_id, action="propose", reason="built twice, in two projects",
+    ))
+
+
+# --- the restore ------------------------------------------------------------
+
+@pytest_asyncio.fixture
+async def source():
+    """An idea with everything family canon can hang off it: a platform scope,
+    a reference snippet, a project that declares one platform and rejects
+    another, a variant answer, and two decisions where the second cites the
+    first as its precedent.
+
+    `decoy` exists so the target database has an ADDITIONAL decision whose id
+    could collide with a source precedent id — without it, a raw-copied
+    precedent list could happen to point at nothing and read as dropped,
+    rather than as pointing at the wrong decision.
+    """
+    owner_id, project_id, idea_id = await _owner_project_and_note()
+    android = await _platform("android-app")
+    container = await _platform("container-image")
+    async with async_session() as s:
+        snippet = Note(user_id=owner_id, title="the reference", body="code",
+                       note_type="snippet")
+        s.add(snippet)
+        s.add(FamilyIdea(note_id=idea_id, status="canon", canon_version=2,
+                         applies_when="any app that installs its own updates"))
+        await s.flush()
+        s.add_all([
+            FamilyIdeaPlatform(note_id=idea_id, platform_id=android.id),
+            FamilyIdeaReference(idea_id=idea_id, snippet_id=snippet.id),
+            ProjectPlatform(project_id=project_id, platform_id=android.id, state="declared"),
+            ProjectPlatform(project_id=project_id, platform_id=container.id, state="rejected"),
+            FamilyAdoption(project_id=project_id, idea_id=idea_id, status="variant",
+                           reason="updates arrive through the store, not in-app",
+                           canon_version=1, decided_via="agent"),
+        ])
+        first = FamilyDecision(idea_id=idea_id, action="promote",
+                               reason="proven in CI, stated in platform terms",
+                               after={"status": "canon"})
+        s.add(first)
+        await s.flush()
+        second = FamilyDecision(idea_id=idea_id, project_id=project_id, action="assess",
+                                reason="store-distributed, as in the promotion's terms",
+                                precedent_ids=[first.id], decided_via="agent")
+        s.add(second)
+        await s.commit()
+        snippet_id = snippet.id
+
+    async with async_session() as s:
+        notes = (await s.execute(
+            select(Note).where(Note.id.in_([idea_id, snippet_id])).order_by(Note.id)
+        )).scalars().all()
+        payload = {
+            "version": backup.BACKUP_VERSION,
+            "users": backup._user_rows([await s.get(User, owner_id)]),
+            "projects": backup._project_rows([await s.get(Project, project_id)]),
+            "notes": backup._note_rows(notes),
+            **backup._family_sections(
+                (await s.execute(select(Platform))).scalars().all(),
+                (await s.execute(select(ProjectPlatform).where(
+                    ProjectPlatform.project_id == project_id))).scalars().all(),
+                [await s.get(FamilyIdea, idea_id)],
+                (await s.execute(select(FamilyIdeaPlatform).where(
+                    FamilyIdeaPlatform.note_id == idea_id))).scalars().all(),
+                (await s.execute(select(FamilyIdeaReference).where(
+                    FamilyIdeaReference.idea_id == idea_id))).scalars().all(),
+                (await s.execute(select(FamilyAdoption).where(
+                    FamilyAdoption.idea_id == idea_id))).scalars().all(),
+                (await s.execute(select(FamilyDecision).where(
+                    FamilyDecision.idea_id == idea_id).order_by(FamilyDecision.id)
+                )).scalars().all(),
+            ),
+        }
+    payload["users"][0]["username"] = RESTORED_USERNAME
+    return {"payload": payload}
+
+
+@pytest_asyncio.fixture
+async def restored(source):
+    async with async_session() as s:
+        before = len((await s.execute(select(Platform))).scalars().all())
+    stats = await backup.restore_full_backup(source["payload"])
+    async with async_session() as s:
+        user = (await s.execute(
+            select(User).where(User.username == RESTORED_USERNAME)
+        )).scalars().one()
+        project = (await s.execute(
+            select(Project).where(Project.user_id == user.id)
+        )).scalars().one()
+        notes = (await s.execute(
+            select(Note).where(Note.user_id == user.id)
+        )).scalars().all()
+        note_ids = [n.id for n in notes]
+        return {
+            "stats": stats,
+            "platforms_before": before,
+            "platforms_after": len((await s.execute(select(Platform))).scalars().all()),
+            "project": project,
+            "notes": {n.note_type: n for n in notes},
+            "idea": (await s.execute(
+                select(FamilyIdea).where(FamilyIdea.note_id.in_(note_ids))
+            )).scalars().one(),
+            "memberships": (await s.execute(
+                select(ProjectPlatform).where(ProjectPlatform.project_id == project.id)
+            )).scalars().all(),
+            "adoption": (await s.execute(
+                select(FamilyAdoption).where(FamilyAdoption.project_id == project.id)
+            )).scalars().one(),
+            "decisions": (await s.execute(
+                select(FamilyDecision).where(FamilyDecision.idea_id.in_(note_ids))
+                .order_by(FamilyDecision.id)
+            )).scalars().all(),
+            "references": (await s.execute(
+                select(FamilyIdeaReference).where(FamilyIdeaReference.idea_id.in_(note_ids))
+            )).scalars().all(),
+        }
+
+
+async def test_platforms_match_by_slug_and_create_nothing(restored):
+    assert restored["stats"]["platforms"] == 0
+    assert restored["platforms_after"] == restored["platforms_before"]
+
+
+async def test_a_rejected_platform_survives_so_detection_cannot_re_add_it(restored):
+    states = {m.state for m in restored["memberships"]}
+    assert states == {"declared", "rejected"}
+
+
+async def test_the_idea_and_its_answer_come_back_whole(restored):
+    idea = restored["idea"]
+    assert idea.status == "canon"
+    assert idea.canon_version == 2
+    adoption = restored["adoption"]
+    assert adoption.status == "variant"
+    assert adoption.reason == "updates arrive through the store, not in-app"
+    # Assessed against version 1 of a version-2 idea: the derived recheck
+    # signal must survive the trip, which needs both numbers intact.
+    assert adoption.canon_version < idea.canon_version
+
+
+async def test_the_reference_points_at_the_RESTORED_snippet(restored):
+    [ref] = restored["references"]
+    assert ref.snippet_id == restored["notes"]["snippet"].id
+
+
+async def test_a_precedent_is_remapped_to_the_RESTORED_decision(restored):
+    """The trap: precedent_ids is a list of ids inside JSON. Copied raw, it
+    would name the SOURCE decision's id — which in a shared database is a
+    real row, the wrong one."""
+    first, second = restored["decisions"]
+    assert first.action == "promote" and second.action == "assess"
+    assert second.precedent_ids == [first.id]
+    assert second.project_id == restored["project"].id

tests/test_services_backup.py

@@ -27,7 +27,7 @@ def test_backup_version_is_current():
 
     (Named for the number it asserted until v10, which is exactly the drift a
     name-carrying-a-value invites; it now says what it checks.)"""
-    assert backup.BACKUP_VERSION == 24
+    assert backup.BACKUP_VERSION == 25
 
 
 def _exportable_note(**over):
@@ -125,11 +125,25 @@ def test_a_repo_binding_carries_the_branch_its_ledger_follows():
     assert row["ref"] == "dev"
 
 
+class _AnySlug(dict):
+    """A slug map in which every platform id resolves. The stand-in's
+    platform_id is an arbitrary integer, and the guard is about which columns
+    travel, not about what a missing platform does — that would make the
+    import builder skip a row the guard needs it to build."""
+
+    def get(self, key, default=None):
+        return "platform-x"
+
+
 # The table -> (model, row helper) registry the column guard walks. Kept here
 # rather than in the service because it exists only to be introspected: the
 # product code already knows these pairings by calling them.
 def _column_guard_targets():
     from scribe.models.canonical_system import CanonicalSystem
+    from scribe.models.family import (
+        FamilyAdoption, FamilyDecision, FamilyIdea, FamilyIdeaPlatform,
+        FamilyIdeaReference, Platform, ProjectPlatform,
+    )
     from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
     from scribe.models.lesson_rule_link import LessonNoRule, LessonRuleLink
     from scribe.models.rule_moment_judgment import RuleMomentJudgment
@@ -187,6 +201,18 @@ def _column_guard_targets():
         "code_shapes": (CodeShape, backup._code_shape_rows),
         "code_shape_events": (CodeShapeEvent, backup._code_shape_event_rows),
         "code_shape_uses": (CodeShapeUse, backup._code_shape_use_rows),
+        "platforms": (Platform, backup._platform_rows),
+        "project_platforms": (
+            ProjectPlatform, lambda rows: backup._project_platform_rows(rows, _AnySlug()),
+        ),
+        "family_ideas": (FamilyIdea, backup._family_idea_rows),
+        "family_idea_platforms": (
+            FamilyIdeaPlatform,
+            lambda rows: backup._family_idea_platform_rows(rows, _AnySlug()),
+        ),
+        "family_idea_references": (FamilyIdeaReference, backup._family_idea_reference_rows),
+        "family_adoptions": (FamilyAdoption, backup._family_adoption_rows),
+        "family_decisions": (FamilyDecision, backup._family_decision_rows),
     }
 
 
@@ -306,6 +332,13 @@ def _import_guard_targets():
         "code_shapes": backup._build_code_shape,
         "code_shape_events": backup._build_code_shape_event,
         "code_shape_uses": backup._build_code_shape_use,
+        "platforms": backup._build_platform,
+        "project_platforms": backup._build_project_platform,
+        "family_ideas": backup._build_family_idea,
+        "family_idea_platforms": backup._build_family_idea_platform,
+        "family_idea_references": backup._build_family_idea_reference,
+        "family_adoptions": backup._build_family_adoption,
+        "family_decisions": backup._build_family_decision,
     }
     return {
         table: (model, helper, builders[table])
@@ -324,7 +357,8 @@ def _everything_maps(row: dict) -> "backup._Maps":
     maps = backup._Maps()
     ids = {v for v in row.values() if isinstance(v, int)} | {0, 1}
     for name in ("users", "projects", "milestones", "notes", "rulebooks",
-                 "topics", "rules", "systems", "design_systems", "shapes"):
+                 "topics", "rules", "systems", "design_systems", "shapes",
+                 "decisions"):
         getattr(maps, name).update({i: i + 1000 for i in ids})
     # `canonical_slug` only, never `slug`: a canonical_systems row is built
     # exactly when its slug is NOT already known to the destination, so
@@ -332,6 +366,11 @@ def _everything_maps(row: dict) -> "backup._Maps":
     slug = row.get("canonical_slug")
     if slug:
         maps.canonical_by_slug[slug] = 7
+    # `platform_slug` only, never `slug`, for the same reason: a platforms row
+    # is built exactly when its slug is unknown to the destination.
+    platform_slug = row.get("platform_slug")
+    if platform_slug:
+        maps.platform_by_slug[platform_slug] = 8
     return maps
 
 
@@ -556,7 +595,11 @@ async def test_export_full_backup_contains_every_declared_section():
                 # v20: whether an area's rulings were read once shown.
                 "system_usage_events",
                 # v23: proposals and judgments about a rule's moments.
-                "rule_moment_judgments"):
+                "rule_moment_judgments",
+                # v25: family canon (milestone 463).
+                "platforms", "project_platforms", "family_ideas",
+                "family_idea_platforms", "family_idea_references",
+                "family_adoptions", "family_decisions"):
         assert key in out, f"missing export section: {key}"
         assert out[key] == []