feat(inception): services/inception.decide() + current_defaults(); the standard Systems vocabulary moves to the service and seeds at inception (#2881, milestone 297 step 3)

- inception.decide(user, project, choices=, via=): owner-only; validates the
  choices (pure) and every target (owned rulebook / always-on for an
  exclusion / readable design system) BEFORE any effect; then, each
  idempotent: exclude always-on rulebooks, subscribe rulebooks, point the
  design system (None = explicitly none), seed the standard Systems if asked
  and the project has none; writes projects.inception LAST. Re-deciding is
  additive for exclusions/subscriptions, replaces the design system, never
  re-seeds.
- inception.current_defaults(): what binds if nobody decides — the ask's
  payload (always-on / other rulebooks, standing exclusions + subscriptions,
  design system + the choices, Systems count).
- services/systems.STANDARD_SYSTEMS (name + generic charter) + seed_standard_systems();
  mcp/tools/systems names the same list in the bootstrap ask — one vocabulary.
- Integration tests: effects land and the record says why; bad targets apply
  nothing; outsiders cannot decide.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

src/scribe/mcp/tools/systems.py

@@ -30,10 +30,10 @@ _BOOTSTRAP_TITLES = 6
 # design (rule #115): archetypes any codebase could have, never one
 # install's subsystems. Mint freely beyond the list; the duplicate gate
 # guards sprawl.
-_STANDARD_SYSTEMS = (
+# The standard vocabulary lives with the service (services/systems.
-    "CI & Release", "Auth & Access", "Data Model & Storage", "API Surface",
+# STANDARD_SYSTEMS) since milestone 297 — the inception seed mints it and this
-    "UI & Design", "Import & Export", "Background Jobs", "Observability",
+# ask names it, one list for both.
-)
+_STANDARD_SYSTEMS = tuple(name for name, _charter in systems_svc.STANDARD_SYSTEMS)
 
 
 async def bootstrap_systems_ask(user_id: int, project_id: int) -> str | None:

src/scribe/services/inception.py

@@ -18,13 +18,24 @@ NULL = undecided → enter_project asks. ``legacy`` is the migration's stamp on
 projects that existed before the step did (inherit-all / no design system /
 no seed), so the ask fires only for projects created after this shipped.
 
-Step 1 (this module's first cut) holds the shape and its validator; the
+The shape and its validator are pure; ``decide`` composes the existing
-effects (decide / current_defaults) arrive in step 3 and compose the
+services — always-on exclusions, subscriptions, set_project_design_system,
-existing services — subscriptions, exclusions, set_project_design_system,
+the standard Systems seed — checks every target BEFORE touching anything,
-the Systems starter mint — and write the record LAST.
+applies the effects (each idempotent), and writes the record LAST, so a
+half-applied decision is re-runnable rather than recorded as done.
+``current_defaults`` is what the enter_project ask shows: what binds today
+if nobody decides.
 """
 from __future__ import annotations
 
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+
+from scribe.models import async_session
+from scribe.models.project import Project
+from scribe.models.rulebook import Rulebook
+
 INCEPTION_VIAS = ("mcp", "ui", "legacy")
 CHOICE_KEYS = ("exclude_always_on_rulebooks", "subscribe_rulebooks", "design_system_id", "seed_systems")
 
@@ -82,3 +93,147 @@ def normalize_choices(choices: dict | None) -> dict:
 def is_decided(project) -> bool:
     """A project is decided once its inception record exists (any via)."""
     return bool(getattr(project, "inception", None))
+
+
+async def current_defaults(user_id: int, project_id: int) -> dict:
+    """What the project inherits if nobody decides — the ask's payload.
+
+    {always_on_rulebooks: [{id,title}], other_rulebooks: [{id,title}],
+     excluded_always_on: [...], subscribed_rulebooks: [...],
+     design_system_id, design_systems: [{id,title}], systems: <count>}.
+    Instance-agnostic: an install with no rulebooks / design systems shows
+    empty lists, and the ask says so rather than inventing a default.
+    """
+    from scribe.services import design_systems as design_systems_svc
+    from scribe.services import projects as projects_svc
+    from scribe.services import rulebooks as rulebooks_svc
+    from scribe.services import systems as systems_svc
+
+    project = await projects_svc.get_project(user_id, project_id)
+    if project is None:
+        raise ValueError(f"project {project_id} not found")
+    async with async_session() as session:
+        rows = (
+            await session.execute(
+                select(Rulebook.id, Rulebook.title, Rulebook.always_on)
+                .where(Rulebook.owner_user_id == user_id, Rulebook.deleted_at.is_(None))
+                .order_by(Rulebook.title)
+            )
+        ).all()
+    applicable = await rulebooks_svc.get_applicable_rules(project_id, user_id, limit=1)
+    designs = await design_systems_svc.list_design_systems(user_id)
+    systems = await systems_svc.list_systems(user_id, project_id, include_archived=True)
+    return {
+        "always_on_rulebooks": [{"id": i, "title": t} for i, t, on in rows if on],
+        "other_rulebooks": [{"id": i, "title": t} for i, t, on in rows if not on],
+        "excluded_always_on": applicable.get("excluded_always_on", []),
+        "subscribed_rulebooks": applicable.get("subscribed_rulebooks", []),
+        "design_system_id": project.design_system_id,
+        "design_systems": [{"id": d.id, "title": d.title} for d in designs],
+        "systems": len(systems),
+    }
+
+
+async def _check_targets(user_id: int, choices: dict) -> None:
+    """Every id a decision names must be the caller's (or readable) BEFORE any
+    effect lands — a decision applies whole or errors whole."""
+    from scribe.services import access
+
+    wanted = set(choices["exclude_always_on_rulebooks"]) | set(choices["subscribe_rulebooks"])
+    if wanted:
+        async with async_session() as session:
+            rows = (
+                await session.execute(
+                    select(Rulebook.id, Rulebook.always_on).where(
+                        Rulebook.id.in_(wanted),
+                        Rulebook.owner_user_id == user_id,
+                        Rulebook.deleted_at.is_(None),
+                    )
+                )
+            ).all()
+        found = {rid: on for rid, on in rows}
+        missing = sorted(wanted - set(found))
+        if missing:
+            raise ValueError(f"rulebook(s) {missing} not found (or not yours)")
+        not_always = sorted(r for r in choices["exclude_always_on_rulebooks"] if not found[r])
+        if not_always:
+            raise ValueError(
+                f"rulebook(s) {not_always} are not always-on — only always-on rulebooks "
+                "can be excluded; a subscribed rulebook is simply not subscribed"
+            )
+    ds = choices["design_system_id"]
+    if ds is not None and not await access.can_read_design_system(user_id, ds):
+        raise ValueError(f"design system {ds} not found (or not readable)")
+
+
+async def decide(
+    user_id: int,
+    project_id: int,
+    *,
+    choices: dict | None,
+    via: str,
+) -> dict:
+    """Record a project's inception decision and apply it (milestone 297).
+
+    Owner-only. Validates the choices (pure) and every target (owned /
+    readable) first; then, each idempotent: exclude the named always-on
+    rulebooks, subscribe the named rulebooks, point the project at the design
+    system (None = explicitly none), seed the standard Systems if asked and
+    the project has none; then write ``projects.inception`` LAST. Re-deciding
+    is additive for exclusions/subscriptions (nothing is silently dropped —
+    include/unsubscribe are explicit calls), replaces the design system, and
+    re-seeds nothing a project already has.
+
+    Returns {"inception": <record>, "effects": {excluded, subscribed,
+    design_system_id, systems_seeded}}.
+    """
+    from scribe.services import design_systems as design_systems_svc
+    from scribe.services import projects as projects_svc
+    from scribe.services import rulebooks as rulebooks_svc
+    from scribe.services import systems as systems_svc
+
+    if via not in INCEPTION_VIAS or via == "legacy":
+        raise ValueError("via must be 'mcp' or 'ui' ('legacy' is the migration's stamp)")
+    error = validate_inception(choices or {})
+    if error:
+        raise ValueError(error)
+    choices = normalize_choices(choices)
+    project = await projects_svc.get_project(user_id, project_id)   # owner-scoped
+    if project is None:
+        raise ValueError(f"project {project_id} not found (or not yours)")
+    await _check_targets(user_id, choices)
+
+    for rb in choices["exclude_always_on_rulebooks"]:
+        await rulebooks_svc.exclude_always_on_rulebook_for_project(project_id, rb, user_id)
+    for rb in choices["subscribe_rulebooks"]:
+        await rulebooks_svc.subscribe_project(project_id, rb, user_id)
+    if not await design_systems_svc.set_project_design_system(
+        user_id, project_id, choices["design_system_id"]
+    ):
+        raise ValueError("could not set the design system (no write on the project?)")
+    seeded = (
+        await systems_svc.seed_standard_systems(user_id, project_id)
+        if choices["seed_systems"] else []
+    )
+
+    record = {
+        "decided_at": datetime.now(timezone.utc).isoformat(),
+        "decided_by": user_id,
+        "via": via,
+        "choices": choices,
+    }
+    async with async_session() as session:
+        row = await session.get(Project, project_id)
+        row.inception = record
+        row.updated_at = datetime.now(timezone.utc)
+        await session.commit()
+    return {
+        "inception": record,
+        "effects": {
+            "excluded": choices["exclude_always_on_rulebooks"],
+            "subscribed": choices["subscribe_rulebooks"],
+            "design_system_id": choices["design_system_id"],
+            "systems_seeded": [sy.name for sy in seeded],
+        },
+    }
+

src/scribe/services/systems.py

@@ -18,6 +18,41 @@ from scribe.services import access
 logger = logging.getLogger(__name__)
 
 
+# The standard cross-project vocabulary (#2798): names that mean the same
+# thing in every project, so a starter set reads the same everywhere. The
+# bootstrap ask (mcp/tools/systems) names them; the inception seed
+# (services/inception, milestone 297) mints them. Charters are deliberately
+# generic — a project refines them as its own records accrue.
+STANDARD_SYSTEMS: tuple[tuple[str, str], ...] = (
+    ("CI & Release", "How the project is verified and shipped: pipelines, runners, image/artifact builds, release tagging and rollback."),
+    ("Auth & Access", "Who may do what: identity, sessions/tokens, permissions and the scoping of every read and write to the right users."),
+    ("Data Model & Storage", "What is stored and how it is shaped: the schema, migrations, serialisation and the services that own a table's lifecycle."),
+    ("API Surface", "The doors into the capability: HTTP routes, tool/RPC surfaces, request parsing, error envelopes and their contracts."),
+    ("UI & Design", "What people see and touch: views, components, client state, and the design tokens/recipes they are built from."),
+    ("Import & Export", "Data crossing the boundary: backups, exports, imports, sync with other systems, file formats."),
+    ("Background Jobs", "Work that runs without a request: schedulers, queues, periodic ticks, retention and maintenance."),
+    ("Observability", "How the system reports on itself: logging, metrics, audit trails, health and diagnostics."),
+)
+
+
+async def seed_standard_systems(user_id: int, project_id: int) -> list[System]:
+    """Mint the standard starter set for a project that has NO Systems yet
+    (milestone 297). Idempotent: a project with any System — the vocabulary
+    already started, standard or not — gets nothing; the duplicate gate and
+    the project's own judgment take it from there. [] without write access."""
+    if await list_systems(user_id, project_id, include_archived=True):
+        return []
+    out: list[System] = []
+    for index, (name, charter) in enumerate(STANDARD_SYSTEMS):
+        system = await create_system(
+            user_id, project_id, name, description=charter, order_index=index,
+        )
+        if system is None:
+            break
+        out.append(system)
+    return out
+
+
 async def create_system(
     user_id: int,
     project_id: int,

tests/test_inception.py

@@ -55,3 +55,11 @@ def test_normalize_choices_is_canonical_and_complete():
                    "design_system_id": None, "seed_systems": False}
     assert normalize_choices(None) == {"exclude_always_on_rulebooks": [], "subscribe_rulebooks": [],
                                        "design_system_id": None, "seed_systems": False}
+
+
+def test_standard_systems_vocabulary_is_one_list_for_ask_and_seed():
+    from scribe.mcp.tools.systems import _STANDARD_SYSTEMS
+    from scribe.services.systems import STANDARD_SYSTEMS
+    assert _STANDARD_SYSTEMS == tuple(n for n, _ in STANDARD_SYSTEMS)
+    assert len(STANDARD_SYSTEMS) == 8 and all(charter for _, charter in STANDARD_SYSTEMS)
+

tests/test_integration_inception.py

@@ -0,0 +1,111 @@
+"""Real-Postgres integration tests for project inception (milestone 297).
+
+What mocks can't prove: a decision's effects land through the real services
+(exclusions filter the always-on set, subscriptions bind, the design system
+points, the standard Systems seed once), the record is written last, a bad
+target applies nothing.
+"""
+import pytest
+import pytest_asyncio
+
+from scribe.models import async_session
+from scribe.models.project import Project
+from scribe.models.rulebook import Rulebook
+from scribe.services import inception as inception_svc
+from scribe.services import rulebooks as rulebooks_svc
+from scribe.services import systems as systems_svc
+from tests.helpers import ensure_user
+
+pytestmark = [pytest.mark.integration, pytest.mark.usefixtures("_dispose_engine")]
+
+
+@pytest_asyncio.fixture
+async def seeded():
+    """Owner, a fresh project, one always-on rulebook (with a rule) and one
+    ordinary rulebook (with a rule)."""
+    async with async_session() as s:
+        owner = await ensure_user(s, "inception_owner")
+        project = Project(user_id=owner.id, title="Inception target")
+        s.add(project)
+        await s.flush()
+        ids = {"owner": owner.id, "pid": project.id}
+        await s.commit()
+    always = await rulebooks_svc.create_rulebook(ids["owner"], "Family standards")
+    other = await rulebooks_svc.create_rulebook(ids["owner"], "Optional practices")
+    async with async_session() as s:
+        rb = await s.get(Rulebook, always.id)
+        rb.always_on = True
+        await s.commit()
+    t1 = await rulebooks_svc.create_topic(always.id, ids["owner"], "git")
+    await rulebooks_svc.create_rule(t1.id, ids["owner"], "dev is home", "Work on dev.")
+    t2 = await rulebooks_svc.create_topic(other.id, ids["owner"], "docs")
+    await rulebooks_svc.create_rule(t2.id, ids["owner"], "Write the why", "Record reasons.")
+    ids.update({"always": always.id, "other": other.id})
+    return ids
+
+
+@pytest.mark.integration
+async def test_decide_applies_every_effect_and_records_last(seeded):
+    owner, pid = seeded["owner"], seeded["pid"]
+    # Undecided: the always-on rulebook binds, nothing subscribed, no Systems.
+    assert [r.title for r in await rulebooks_svc.list_always_on_rules(owner, project_id=pid)] == ["dev is home"]
+    defaults = await inception_svc.current_defaults(owner, pid)
+    assert [r["id"] for r in defaults["always_on_rulebooks"]] == [seeded["always"]]
+    assert [r["id"] for r in defaults["other_rulebooks"]] == [seeded["other"]]
+    assert defaults["systems"] == 0 and defaults["design_system_id"] is None
+
+    out = await inception_svc.decide(owner, pid, via="mcp", choices={
+        "exclude_always_on_rulebooks": [seeded["always"]],
+        "subscribe_rulebooks": [seeded["other"]],
+        "design_system_id": None,
+        "seed_systems": True,
+    })
+    assert out["effects"]["excluded"] == [seeded["always"]]
+    assert out["effects"]["subscribed"] == [seeded["other"]]
+    assert len(out["effects"]["systems_seeded"]) == len(systems_svc.STANDARD_SYSTEMS)
+
+    # The exclusion is total: the project's always-on set is empty, the
+    # departure is named, the subscription binds.
+    assert await rulebooks_svc.list_always_on_rules(owner, project_id=pid) == []
+    assert len(await rulebooks_svc.list_always_on_rules(owner)) == 1   # user-wide unchanged
+    applicable = await rulebooks_svc.get_applicable_rules(pid, owner)
+    assert [r["title"] for r in applicable["rules"]] == ["Write the why"]
+    assert [e["id"] for e in applicable["excluded_always_on"]] == [seeded["always"]]
+    assert [s["id"] for s in applicable["subscribed_rulebooks"]] == [seeded["other"]]
+    # The record, written last, says why.
+    async with async_session() as s:
+        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"]["exclude_always_on_rulebooks"] == [seeded["always"]]
+    # Re-deciding with seed again mints nothing twice; include reverses the exclusion.
+    again = await inception_svc.decide(owner, pid, via="ui", choices={"seed_systems": True})
+    assert again["effects"]["systems_seeded"] == []
+    assert len(await systems_svc.list_systems(owner, pid)) == len(systems_svc.STANDARD_SYSTEMS)
+    await rulebooks_svc.include_always_on_rulebook_for_project(pid, seeded["always"], owner)
+    assert [r.title for r in await rulebooks_svc.list_always_on_rules(owner, project_id=pid)] == ["dev is home"]
+
+
+@pytest.mark.integration
+async def test_a_bad_decision_applies_nothing(seeded):
+    owner, pid = seeded["owner"], seeded["pid"]
+    # Excluding a rulebook that is not always-on is refused BEFORE any effect.
+    with pytest.raises(ValueError, match="not always-on"):
+        await inception_svc.decide(owner, pid, via="mcp", choices={
+            "exclude_always_on_rulebooks": [seeded["other"]], "seed_systems": True,
+        })
+    assert await systems_svc.list_systems(owner, pid) == []
+    with pytest.raises(ValueError, match="not found"):
+        await inception_svc.decide(owner, pid, via="mcp", choices={"subscribe_rulebooks": [999999]})
+    with pytest.raises(ValueError, match="legacy"):
+        await inception_svc.decide(owner, pid, via="legacy", choices={})
+    async with async_session() as s:
+        project = await s.get(Project, pid)
+    assert not inception_svc.is_decided(project)
+    # An outsider cannot decide someone else's project.
+    async with async_session() as s:
+        other = await ensure_user(s, "inception_other")
+        other_id = other.id
+        await s.commit()
+    with pytest.raises(ValueError, match="not found"):
+        await inception_svc.decide(other_id, pid, via="mcp", choices={})