feat(systems): the area vocabulary becomes a global table so a rule can point at one (#3027, milestone 307 step 1)
CI & Build / Plugin hooks (push) Successful in 10s
CI & Build / TypeScript typecheck (push) Successful in 39s
CI & Build / Python lint (push) Successful in 4s
CI & Build / integration (push) Successful in 30s
CI & Build / Python tests (push) Failing after 56s
CI & Build / Build & push image (push) Skipped

The eight standard area names already existed — as STANDARD_SYSTEMS, a tuple in
services/systems.py that milestone 297 seeds at inception. A constant cannot be
a foreign key, so nothing outside a project could reference an area: systems.
project_id is NOT NULL, and a rule that spans projects would have to chain
itself to one project's row. And because the list only ever applied on the
inception-seed path, three spellings of one area reached this instance anyway
(CI & runners / CI and Release / CI & release).

- canonical_systems: global, no user_id — a shared project inherits the
  vocabulary instead of re-earning it. Migration 0087 seeds the same eight.
- systems.canonical_id: nullable, SET NULL. Association only — no System is
  renamed and record_systems is untouched, so no record's tags move.
- canonical_slug folds &/and, case and punctuation, so spelling variants map
  mechanically and a real difference ("CI & runners") becomes a proposal a
  human confirms. propose_mappings reports; set_system_canonical is the only
  writer.
- seed_standard_systems now reads the catalog and maps as it mints, so a
  project born standard never needs a reconciliation pass.
- Catalog writes are admin-only; reads are open — a global list anyone can
  extend stops being shared.
- backup: carried by SLUG, not id (ids are per-install). Restore reuses the
  target's own rows and only creates entries an admin added on the source; an
  unknown slug restores unmapped rather than failing.

Rule 22: STANDARD_SYSTEMS is removed, not deprecated. Rule 115: nothing seeded
names an app, repo or house convention. Design in note 3026.

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-08-26 12:31:18 -04:00
co-authored by Claude Opus 5
parent a8f35e465e
commit a97547fbc6
14 changed files with 800 additions and 38 deletions
+109
View File
@@ -0,0 +1,109 @@
"""canonical_systems — the global area vocabulary, promoted from a constant
to a table (milestone 307 step 1, decision note 3026)
Revision ID: 0087
Revises: 0086
Create Date: 2026-08-26
The eight standard area names already existed as `STANDARD_SYSTEMS`, a tuple in
services/systems.py that milestone 297 seeds into a project at inception. A
constant cannot be referenced: a rule that applies across projects has nothing
to point at, because `systems.project_id` is NOT NULL and a family rule cannot
be chained to one project's row. This makes the vocabulary a table so it can be
a foreign key, and adds the nullable `systems.canonical_id` that maps a
project's local System onto it.
Deliberately no `user_id`: the catalog is GLOBAL so a shared project inherits
the vocabulary rather than re-earning it. `record_systems` is untouched — it
joins note_id/system_id and never sees this table, so no association data
moves, and no System's own `name` is rewritten.
The seed rows are written here verbatim rather than imported from the service:
a migration is a historical record and must keep running unchanged after the
service's list moves on.
"""
import sqlalchemy as sa
from alembic import op
revision = "0087"
down_revision = "0086"
branch_labels = None
depends_on = None
# (name, slug, description) — the milestone-297 vocabulary, with the slug the
# service computes (canonical_slug: lowercase, "&" -> "and", non-alphanumerics
# collapsed to "-"). Charters stay generic on purpose: a project refines its
# own System's description, never this one. Nothing here names an app, a repo,
# a vendor or a house convention — the catalog ships to every install (rule 115).
_SEED = (
("CI & Release", "ci-and-release",
"How the project is verified and shipped: pipelines, runners, image/artifact builds, release tagging and rollback."),
("Auth & Access", "auth-and-access",
"Who may do what: identity, sessions/tokens, permissions and the scoping of every read and write to the right users."),
("Data Model & Storage", "data-model-and-storage",
"What is stored and how it is shaped: the schema, migrations, serialisation and the services that own a table's lifecycle."),
("API Surface", "api-surface",
"The doors into the capability: HTTP routes, tool/RPC surfaces, request parsing, error envelopes and their contracts."),
("UI & Design", "ui-and-design",
"What people see and touch: views, components, client state, and the design tokens/recipes they are built from."),
("Import & Export", "import-and-export",
"Data crossing the boundary: backups, exports, imports, sync with other systems, file formats."),
("Background Jobs", "background-jobs",
"Work that runs without a request: schedulers, queues, periodic ticks, retention and maintenance."),
("Observability", "observability",
"How the system reports on itself: logging, metrics, audit trails, health and diagnostics."),
)
def upgrade() -> None:
canonical_systems = op.create_table(
"canonical_systems",
sa.Column("id", sa.Integer(), primary_key=True),
sa.Column("name", sa.Text(), nullable=False),
sa.Column("slug", sa.Text(), nullable=False),
sa.Column("description", sa.Text(), nullable=True),
sa.Column("order_index", sa.Integer(), nullable=False, server_default="0"),
sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.text("now()")),
sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
sa.Column("deleted_batch_id", sa.Text(), nullable=True),
)
# Unique among LIVE rows only, so a soft-deleted entry doesn't block
# recreating or restoring the same area (the rules/topics convention).
op.create_index(
"uq_canonical_systems_slug", "canonical_systems", ["slug"],
unique=True, postgresql_where=sa.text("deleted_at IS NULL"),
)
op.bulk_insert(
canonical_systems,
[
{"name": name, "slug": slug, "description": description, "order_index": index}
for index, (name, slug, description) in enumerate(_SEED)
],
)
op.add_column(
"systems",
sa.Column("canonical_id", sa.Integer(), nullable=True),
)
# SET NULL, not CASCADE: retiring a catalog entry must never delete a
# project's System along with it.
op.create_foreign_key(
"fk_systems_canonical_id", "systems", "canonical_systems",
["canonical_id"], ["id"], ondelete="SET NULL",
)
op.create_index("ix_systems_canonical_id", "systems", ["canonical_id"])
# Existing Systems are left UNMAPPED on purpose. An exact-slug match would
# be safe, but a near miss ("CI & runners" vs "CI & Release") is a judgment
# call — those go through the propose/confirm path so a human approves each
# one, rather than being decided by a migration nobody reviews.
def downgrade() -> None:
op.drop_index("ix_systems_canonical_id", table_name="systems")
op.drop_constraint("fk_systems_canonical_id", "systems", type_="foreignkey")
op.drop_column("systems", "canonical_id")
op.drop_index("uq_canonical_systems_slug", table_name="canonical_systems")
op.drop_table("canonical_systems")
+2
View File
@@ -30,6 +30,7 @@ from scribe.routes.design_systems import design_systems_bp
from scribe.routes.trash import trash_bp
from scribe.routes.dashboard import dashboard_bp
from scribe.routes.systems import systems_bp
from scribe.routes.canonical_systems import canonical_systems_bp
from scribe.routes.snippets import snippets_bp
from scribe.routes.webhooks import webhooks_bp
from scribe.mcp import mount_mcp
@@ -95,6 +96,7 @@ def create_app() -> Quart:
app.register_blueprint(trash_bp)
app.register_blueprint(dashboard_bp)
app.register_blueprint(systems_bp)
app.register_blueprint(canonical_systems_bp)
app.register_blueprint(snippets_bp)
app.register_blueprint(webhooks_bp)
+3
View File
@@ -91,6 +91,9 @@ _READ_ONLY_TOOLS = frozenset({
"list_rules", "list_tags", "list_tasks", "list_topics", "list_trash",
"list_always_on_rules", "search",
"get_system", "list_systems", "list_system_records",
# The global area catalog and its mapping REPORT — propose writes nothing;
# map_system_to_canonical is the separate, explicitly-called write.
"list_canonical_systems", "propose_canonical_mappings",
# Reports on the corpus. Reads only — the merge or supersession each
# suggests is a separate, explicitly-called write.
"find_duplicate_snippets", "find_duplicate_records",
+78 -8
View File
@@ -13,6 +13,7 @@ Sentinels (match the milestone/task tool conventions):
from __future__ import annotations
from scribe.mcp._context import current_user_id
from scribe.services import canonical_systems as canonical_systems_svc
from scribe.services import notes as notes_svc
from scribe.services import systems as systems_svc
@@ -30,10 +31,9 @@ _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.
# The standard vocabulary lives with the service (services/systems.
# STANDARD_SYSTEMS) since milestone 297 — the inception seed mints it and this
# ask names it, one list for both.
_STANDARD_SYSTEMS = tuple(name for name, _charter in systems_svc.STANDARD_SYSTEMS)
# The standard vocabulary lives in the GLOBAL canonical catalog since
# milestone 307 — the inception seed mints it and this ask names it, one list
# for both, now a table so a rule can reference an area by id (note 3026).
async def bootstrap_systems_ask(user_id: int, project_id: int) -> str | None:
@@ -67,6 +67,17 @@ async def bootstrap_systems_ask(user_id: int, project_id: int) -> str | None:
titles = "; ".join(
'"' + " ".join((n.title or "").split())[:70] + '"' for n in recent
)
try:
standard = ", ".join(name for name, _charter in await systems_svc.standard_systems())
except Exception:
standard = ""
# An install whose catalog is empty still gets a usable ask — the standard
# names are an aid, not the point of the question.
standard_line = (
"Where an area fits a standard name, use it verbatim so it means the "
f"same thing in every project: {standard}. Mint freely beyond that "
"list — the duplicate gate guards sprawl. "
) if standard else "The duplicate gate guards sprawl. "
return (
f"This project has {total} records and NO Systems modelled — none of "
"them can be tagged to an area, so recurring problem-spots stay "
@@ -74,10 +85,8 @@ async def bootstrap_systems_ask(user_id: int, project_id: int) -> str | None:
"asking permission — creating Systems is your call, not an approval "
f"flow. From the areas the records themselves name (recent: {titles}), "
"create_system 3-6 Systems, each with a one-paragraph charter, then "
"tag this record (system_ids=[...]). Where an area fits a standard "
f"name, use it verbatim so it means the same thing in every project: "
f"{', '.join(_STANDARD_SYSTEMS)}. Mint freely beyond that list — the "
"duplicate gate guards sprawl. This ask repeats until the first "
f"tag this record (system_ids=[...]). {standard_line}"
"This ask repeats until the first "
"System exists; answering it once retires it for every future record."
)
@@ -322,6 +331,64 @@ async def delete_system(system_id: int) -> dict:
return {"message": f"System {system_id} deleted."}
async def list_canonical_systems() -> dict:
"""The GLOBAL vocabulary of area names, shared by every project.
These are the standard names to prefer when creating a System, so the same
word means the same thing in every project on the instance — and, from
milestone 307, the ids a cross-project record can point at. A project's own
System keeps whatever name the project calls the area; mapping it here is
an association, never a rename.
Reach for it before create_system when the area is an ordinary one (CI,
auth, storage, the API, the UI), and pass the matching `canonical_id`.
"""
entries = await canonical_systems_svc.list_canonical_systems()
return {"canonical_systems": [e.to_dict() for e in entries]}
async def propose_canonical_mappings(project_id: int) -> dict:
"""Suggest a global area for each of this project's UNMAPPED Systems.
Returns PROPOSALS ONLY — nothing is written. Confirm the ones that are
right with map_system_to_canonical(system_id, canonical_id); ignore the
rest. Each carries a `basis`:
- `exact` — the names reduce to the same match key ("CI and Release" vs
"CI & Release"). Safe to confirm without much thought.
- `overlap` — they share a meaningful word ("CI & runners" vs "CI &
Release"). A judgment call: confirm only if they really are the same
area, since a wrong mapping surfaces cross-project records in the wrong
place.
A System with no proposal is not a problem — unmapped is a valid resting
state, and a genuinely project-specific area should stay that way.
"""
uid = current_user_id()
return {"proposals": await canonical_systems_svc.propose_mappings(uid, project_id)}
async def map_system_to_canonical(system_id: int, canonical_id: int = 0) -> dict:
"""Map one of a project's Systems onto a global area (or clear it).
Sets `canonical_id` and NOTHING else — the System's name, charter and every
record tagged to it are untouched. Pass canonical_id=0 to unmap.
Args:
canonical_id: id from list_canonical_systems; 0 clears the mapping.
"""
uid = current_user_id()
system = await canonical_systems_svc.set_system_canonical(
uid, system_id, canonical_id or None,
)
if system is None:
raise ValueError(
f"system {system_id} not found, no write access, "
f"or canonical_id {canonical_id} is not a live catalog entry"
)
return system.to_dict()
def register(mcp) -> None:
for fn in (
create_system,
@@ -330,5 +397,8 @@ def register(mcp) -> None:
update_system,
list_system_records,
delete_system,
list_canonical_systems,
propose_canonical_mappings,
map_system_to_canonical,
):
mcp.tool(name=fn.__name__)(fn)
+1
View File
@@ -46,4 +46,5 @@ from scribe.models.repo_binding import RepoBinding # noqa: E402, F401
from scribe.models.forge_connection import ForgeConnection # noqa: E402, F401
from scribe.models.code_shape import CodeShape, CodeShapeConsumer, CodeShapeEvent, CodeShapeUse # noqa: E402, F401
from scribe.models.system import System, RecordSystem # noqa: E402, F401
from scribe.models.canonical_system import CanonicalSystem # noqa: E402, F401
from scribe.models.design_system import DesignSystem, DesignToken # noqa: E402, F401
+58
View File
@@ -0,0 +1,58 @@
from sqlalchemy import Index, Integer, Text, text
from sqlalchemy.orm import Mapped, mapped_column
from scribe.models import Base
from scribe.models.base import SoftDeleteMixin, TimestampMixin, iso
class CanonicalSystem(Base, TimestampMixin, SoftDeleteMixin):
"""A GLOBAL area name — the shared vocabulary every project's Systems
can point at (milestone 307, decision note 3026).
A `System` is per-project and NOT NULL on project_id, so nothing outside a
project can reference one: a rule that applies across projects has no way
to say "this is about CI" without chaining itself to one project's row.
This table is that join key. It carries no `user_id` on purpose — a shared
project must INHERIT the vocabulary rather than re-earn it, so the catalog
is global and the same word means the same thing in every install.
It is a convergence aid, never a gate: `systems.canonical_id` is nullable,
an unmapped System stays fully usable, and `record_systems` never sees this
table at all — the local name is a legitimate local label and is never
rewritten to match.
`slug` is the match key, not a display value. It folds the spelling
differences that produced four names for one area on the author's own
instance ("CI & Release" / "CI and Release" / "CI & release"): an exact slug
hit maps automatically, and anything short of that becomes a proposal for a
human to confirm. See services/canonical_systems.canonical_slug.
"""
__tablename__ = "canonical_systems"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column(Text, nullable=False)
# Normalized match key — unique among LIVE rows, so a soft-deleted entry
# doesn't block recreating the same area (the partial-unique convention
# rules/topics already use).
slug: Mapped[str] = mapped_column(Text, nullable=False)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
order_index: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
__table_args__ = (
Index(
"uq_canonical_systems_slug", "slug",
unique=True, postgresql_where=text("deleted_at IS NULL"),
),
)
def to_dict(self) -> dict:
return {
"id": self.id,
"name": self.name,
"slug": self.slug,
"description": self.description,
"order_index": self.order_index,
"created_at": iso(self.created_at),
"updated_at": iso(self.updated_at),
}
+10
View File
@@ -24,6 +24,15 @@ class System(Base, TimestampMixin, SoftDeleteMixin):
Integer, ForeignKey("projects.id", ondelete="CASCADE")
)
name: Mapped[str] = mapped_column(Text, default="", server_default="")
# The GLOBAL area this local System is an instance of (milestone 307).
# Nullable and SET NULL on purpose: the catalog is a convergence aid, not a
# gate — an unmapped System is fully usable, and retiring a canonical entry
# must never take a project's System with it. The local `name` is NEVER
# rewritten to match the canonical one; this column is the join key, and
# the name stays whatever the project calls the area.
canonical_id: Mapped[int | None] = mapped_column(
Integer, ForeignKey("canonical_systems.id", ondelete="SET NULL"), nullable=True
)
description: Mapped[str | None] = mapped_column(Text, nullable=True)
color: Mapped[str | None] = mapped_column(Text, nullable=True)
# active | archived — systems accumulate; archive rather than delete.
@@ -40,6 +49,7 @@ class System(Base, TimestampMixin, SoftDeleteMixin):
"user_id": self.user_id,
"project_id": self.project_id,
"name": self.name,
"canonical_id": self.canonical_id,
"description": self.description,
"color": self.color,
"status": self.status,
+93
View File
@@ -0,0 +1,93 @@
"""Canonical-system routes — the GLOBAL area vocabulary, and the mapping of a
project's Systems onto it (milestone 307, decision note 3026).
Two shapes live here because they are two halves of one idea:
- `/api/canonical-systems` — the catalog itself. Readable by any signed-in
user (it is shared vocabulary, not user data); writable only by an admin,
since a global list anyone can extend stops being a shared list.
- the mapping endpoints — authorised by the PROJECT, because mapping writes a
project's own System row. The service enforces both; these are thin wrappers.
"""
import logging
from quart import Blueprint, jsonify, request
from scribe.auth import admin_required, get_current_user_id, login_required
from scribe.routes.utils import not_found
from scribe.services import canonical_systems as canonical_svc
from scribe.services.projects import get_project_for_user
logger = logging.getLogger(__name__)
canonical_systems_bp = Blueprint("canonical_systems", __name__, url_prefix="/api")
@canonical_systems_bp.route("/canonical-systems", methods=["GET"])
@login_required
async def list_canonical_systems_route():
entries = await canonical_svc.list_canonical_systems()
return jsonify({"canonical_systems": [e.to_dict() for e in entries]})
@canonical_systems_bp.route("/canonical-systems", methods=["POST"])
@admin_required
async def create_canonical_system_route():
uid = get_current_user_id()
data = await request.get_json() or {}
if not (data.get("name") or "").strip():
return jsonify({"error": "name is required"}), 400
entry = await canonical_svc.create_canonical_system(
uid, data["name"], description=data.get("description"),
)
if entry is None:
return jsonify({"error": "Permission denied"}), 403
# The slug duplicate gate answers with the entry that already covers the
# area rather than minting a second spelling of it — 409, not a silent
# second row (the whole point of the table).
if isinstance(entry, dict):
return jsonify(entry), 409
return jsonify(entry.to_dict()), 201
@canonical_systems_bp.route("/canonical-systems/<int:canonical_id>", methods=["PATCH"])
@admin_required
async def update_canonical_system_route(canonical_id: int):
uid = get_current_user_id()
data = await request.get_json() or {}
fields = {k: v for k, v in data.items() if k in ("name", "description", "order_index")}
entry = await canonical_svc.update_canonical_system(uid, canonical_id, **fields)
if entry is None:
return not_found("Canonical system")
return jsonify(entry.to_dict())
@canonical_systems_bp.route(
"/projects/<int:project_id>/canonical-proposals", methods=["GET"]
)
@login_required
async def propose_canonical_mappings_route(project_id: int):
"""Proposals only — this endpoint writes nothing. The PUT below applies one."""
uid = get_current_user_id()
if await get_project_for_user(uid, project_id) is None:
return not_found("Project")
return jsonify({"proposals": await canonical_svc.propose_mappings(uid, project_id)})
@canonical_systems_bp.route("/systems/<int:system_id>/canonical", methods=["PUT"])
@login_required
async def map_system_to_canonical_route(system_id: int):
"""Map or unmap one System. Body: {"canonical_id": <id>|null}.
Sets that column and nothing else — no rename, no change to which records
are tagged to the System.
"""
uid = get_current_user_id()
data = await request.get_json() or {}
canonical_id = data.get("canonical_id")
if canonical_id is not None and not isinstance(canonical_id, int):
return jsonify({"error": "canonical_id must be an integer or null"}), 400
system = await canonical_svc.set_system_canonical(uid, system_id, canonical_id)
if system is None:
return not_found("System or canonical system")
return jsonify(system.to_dict())
+70 -4
View File
@@ -11,6 +11,7 @@ from scribe.models.note_supersession import NoteSupersession
from scribe.models.note_version import NoteVersion
from scribe.models.design_system import DesignSystem, DesignToken
from scribe.models.note_usage import NoteUsageEvent
from scribe.models.canonical_system import CanonicalSystem
from scribe.models.code_shape import CodeShape, CodeShapeEvent, CodeShapeUse
from scribe.models.project import Project
from scribe.models.repo_binding import RepoBinding
@@ -69,6 +70,10 @@ _BACKED_UP = [
"note_usage_events", "repo_bindings", "note_supersessions",
# v7 (2026-08): the shape ledger (#2787); v8: its history (#2793).
"code_shapes", "code_shape_events", "code_shape_uses",
# v9 (2026-08): the global area catalog (milestone 307). Global, not
# user-scoped, so it rides in EVERY export — including a single-user
# one, whose Systems would otherwise restore unmapped.
"canonical_systems",
]
# Tables intentionally NOT in the backup, surfaced in the payload so the gap is
@@ -127,12 +132,29 @@ def _rulebook_exclusion_rows(rows) -> list[dict]:
# same reason: CI has no database, so a serialiser that is a plain function is
# one that can actually be tested.
def _system_rows(rows) -> list[dict]:
def _canonical_system_rows(rows) -> list[dict]:
"""The global area catalog. Carried WITHOUT ids: a restore matches on slug,
so a target install that already seeded the standard vocabulary reuses its
own rows and only gains the entries an admin added here."""
return [
{
"name": r.name, "slug": r.slug, "description": r.description,
"order_index": r.order_index,
}
for r in rows
]
def _system_rows(rows, canonical_slugs: dict[int, str]) -> list[dict]:
"""A project's Systems. The canonical mapping travels as a SLUG, not an id
— the catalog is global and its ids are per-install, so an id would restore
pointing at whatever area happened to land on that number."""
return [
{
"id": r.id, "user_id": r.user_id, "project_id": r.project_id,
"name": r.name, "description": r.description, "color": r.color,
"status": r.status, "order_index": r.order_index,
"canonical_slug": canonical_slugs.get(r.canonical_id or 0),
}
for r in rows
]
@@ -363,6 +385,10 @@ async def export_full_backup() -> dict:
)).scalars().all()
settings = (await session.execute(select(Setting))).scalars().all()
systems = (await session.execute(select(System))).scalars().all()
canonical_systems = (await session.execute(
select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
.order_by(CanonicalSystem.order_index)
)).scalars().all()
record_systems = (await session.execute(select(RecordSystem))).scalars().all()
supersessions = (
await session.execute(select(NoteSupersession))
@@ -424,7 +450,10 @@ async def export_full_backup() -> dict:
"rule_suppressions": _rule_suppression_rows(rule_suppressions),
"topic_suppressions": _topic_suppression_rows(topic_suppressions),
"rulebook_exclusions": _rulebook_exclusion_rows(rulebook_exclusions),
"systems": _system_rows(systems),
"canonical_systems": _canonical_system_rows(canonical_systems),
"systems": _system_rows(
systems, {c.id: c.slug for c in canonical_systems}
),
"record_systems": _record_system_rows(record_systems),
"design_systems": _design_system_rows(design_systems),
"design_tokens": _design_token_rows(design_tokens),
@@ -467,6 +496,12 @@ async def export_user_backup(user_id: int) -> dict:
systems = (await session.execute(
select(System).where(System.user_id == user_id)
)).scalars().all()
# Global: taken whole even in a per-user export, because the Systems
# above reference it and a partial catalog restores partial mappings.
canonical_systems = (await session.execute(
select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
.order_by(CanonicalSystem.order_index)
)).scalars().all()
system_ids = [sy.id for sy in systems]
note_ids = [n.id for n in notes]
# Scoped by the user's SYSTEMS, not their notes: a shared note carrying
@@ -583,7 +618,10 @@ async def export_user_backup(user_id: int) -> dict:
"rule_suppressions": _rule_suppression_rows(rule_suppressions),
"topic_suppressions": _topic_suppression_rows(topic_suppressions),
"rulebook_exclusions": _rulebook_exclusion_rows(rulebook_exclusions),
"systems": _system_rows(systems),
"canonical_systems": _canonical_system_rows(canonical_systems),
"systems": _system_rows(
systems, {c.id: c.slug for c in canonical_systems}
),
"record_systems": _record_system_rows(record_systems),
"design_systems": _design_system_rows(design_systems),
"design_tokens": _design_token_rows(design_tokens),
@@ -697,7 +735,7 @@ async def _restore_v2(data: dict) -> dict:
"systems": 0, "record_systems": 0, "design_systems": 0,
"design_tokens": 0, "note_usage_events": 0, "repo_bindings": 0,
"note_supersessions": 0, "code_shapes": 0, "code_shape_events": 0,
"code_shape_uses": 0,
"code_shape_uses": 0, "canonical_systems": 0,
}
async with async_session() as session:
@@ -972,6 +1010,31 @@ async def _restore_v2(data: dict) -> dict:
# 15. Systems
system_id_map: dict[int, int] = {}
# 14b. The global area catalog, matched on SLUG. This install already
# has the standard vocabulary from its migrations, so the common case
# adds nothing and simply learns which local id each slug is; only an
# entry an admin added on the source instance is created here. Runs
# BEFORE systems, which resolve their mapping through this map.
canonical_id_by_slug: dict[str, int] = {}
existing_canonical = (await session.execute(
select(CanonicalSystem).where(CanonicalSystem.deleted_at.is_(None))
)).scalars().all()
for entry in existing_canonical:
canonical_id_by_slug[entry.slug] = entry.id
for cs_data in data.get("canonical_systems", []):
slug = cs_data.get("slug") or ""
if not slug or slug in canonical_id_by_slug:
continue
entry = CanonicalSystem(
name=cs_data.get("name", ""), slug=slug,
description=cs_data.get("description"),
order_index=cs_data.get("order_index", 0),
)
session.add(entry)
await session.flush()
canonical_id_by_slug[slug] = entry.id
stats["canonical_systems"] += 1
for sy_data in data.get("systems", []):
mapped_uid = user_id_map.get(sy_data.get("user_id", 0))
mapped_pid = project_id_map.get(sy_data.get("project_id", 0))
@@ -984,6 +1047,9 @@ async def _restore_v2(data: dict) -> dict:
color=sy_data.get("color"),
status=sy_data.get("status", "active"),
order_index=sy_data.get("order_index", 0),
# An unknown slug restores UNMAPPED rather than failing: the
# System and its records are the payload, the mapping is an aid.
canonical_id=canonical_id_by_slug.get(sy_data.get("canonical_slug") or ""),
)
session.add(system)
await session.flush()
+256
View File
@@ -0,0 +1,256 @@
"""The global canonical area vocabulary, and the mapping from a project's
Systems onto it (milestone 307 step 1, decision note 3026).
A `System` is per-project. Nothing outside a project can reference one, so a
rule that spans projects has no way to say "this is about CI" without chaining
itself to one project's row. `CanonicalSystem` is that join key, and it is
GLOBAL — no `user_id`, so a shared project inherits the vocabulary instead of
re-earning it.
Two rules govern everything here:
- **Associate, never rewrite.** Mapping a System sets `systems.canonical_id`
and nothing else. The local name stays whatever the project calls the area,
and `record_systems` is never touched — no record's tags move.
- **Propose, never decide.** An exact slug hit is mechanical and maps on
request; anything short of that is a PROPOSAL a human confirms. "CI &
Release" vs "CI & runners" is a judgment call, and the cost of guessing it
wrong silently is a rule surfacing in the wrong project.
Reads are open to any authenticated caller (the catalog is shared vocabulary,
not user data). Writes to the catalog itself are admin-only: a global table
that anyone can extend is how a shared vocabulary stops being shared.
"""
import logging
import re
from datetime import datetime, timezone
from sqlalchemy import select
from scribe.models import async_session
from scribe.models.canonical_system import CanonicalSystem
from scribe.models.system import System
from scribe.models.user import User
from scribe.services import access
logger = logging.getLogger(__name__)
# Tokens that carry no meaning for matching — "&" becomes "and" before the
# split, so it would otherwise dominate the overlap score of every pair.
_NOISE_TOKENS = frozenset({"and", "the", "a", "of"})
_NON_ALNUM = re.compile(r"[^a-z0-9]+")
def canonical_slug(name: str) -> str:
"""The match key for an area name — NOT a display value.
Folds exactly the spelling differences that produced three names for one
area on the author's instance: `CI & Release`, `CI and Release` and
`CI & release` all slug to `ci-and-release`, so they map mechanically.
A real difference survives: `CI & runners` slugs to `ci-and-runners` and
goes through the proposal path where a human decides.
"""
lowered = name.strip().lower().replace("&", " and ")
return "-".join(_NON_ALNUM.sub(" ", lowered).split())
def _tokens(slug: str) -> frozenset[str]:
return frozenset(slug.split("-")) - _NOISE_TOKENS
async def _is_admin(user_id: int) -> bool:
async with async_session() as session:
role = await session.scalar(select(User.role).where(User.id == user_id))
return role == "admin"
async def list_canonical_systems() -> list[CanonicalSystem]:
"""The whole catalog, in display order. Global — no ownership filter."""
async with async_session() as session:
result = await session.execute(
select(CanonicalSystem)
.where(CanonicalSystem.deleted_at.is_(None))
.order_by(CanonicalSystem.order_index.asc(), CanonicalSystem.name.asc())
)
return list(result.scalars().all())
async def get_canonical_system(canonical_id: int) -> CanonicalSystem | None:
async with async_session() as session:
entry = await session.get(CanonicalSystem, canonical_id)
return entry if entry is not None and entry.deleted_at is None else None
async def find_by_name(name: str) -> CanonicalSystem | None:
"""The exact-slug lookup — the mechanical half of matching."""
slug = canonical_slug(name)
if not slug:
return None
async with async_session() as session:
return await session.scalar(
select(CanonicalSystem).where(
CanonicalSystem.slug == slug,
CanonicalSystem.deleted_at.is_(None),
)
)
async def create_canonical_system(
user_id: int, name: str, description: str | None = None,
) -> CanonicalSystem | dict | None:
"""Add an area to the global catalog. Admin only.
Duplicate-gated on the SLUG, not the raw name, so "CI and Release" cannot
be added alongside "CI & Release" — that is the drift this table exists to
end. Returns the existing entry's id instead of creating a second one.
"""
if not await _is_admin(user_id):
return None
slug = canonical_slug(name)
if not slug:
return None
existing = await find_by_name(name)
if existing is not None:
return {
"duplicate": True,
"existing_id": existing.id,
"message": (
f"'{existing.name}' (#{existing.id}) already covers this area — "
f"both names reduce to '{slug}'. Map Systems to it, or "
"update_canonical_system if the charter needs revising."
),
}
async with async_session() as session:
highest = await session.scalar(
select(CanonicalSystem.order_index)
.order_by(CanonicalSystem.order_index.desc())
.limit(1)
)
entry = CanonicalSystem(
name=" ".join(name.split()),
slug=slug,
description=description,
order_index=(highest or 0) + 1,
)
session.add(entry)
await session.commit()
await session.refresh(entry)
return entry
async def update_canonical_system(
user_id: int, canonical_id: int, **fields: object,
) -> CanonicalSystem | None:
"""Rename or re-charter a catalog entry. Admin only.
A rename recomputes the slug — the display name and the match key must not
be allowed to disagree, or the exact-match path silently stops finding it.
"""
if not await _is_admin(user_id):
return None
allowed = {"name", "description", "order_index"}
async with async_session() as session:
entry = await session.get(CanonicalSystem, canonical_id)
if entry is None or entry.deleted_at is not None:
return None
for key, value in fields.items():
if key in allowed and value is not None:
setattr(entry, key, value)
if "name" in fields and fields["name"]:
entry.name = " ".join(str(fields["name"]).split())
entry.slug = canonical_slug(entry.name)
entry.updated_at = datetime.now(timezone.utc)
await session.commit()
await session.refresh(entry)
return entry
async def set_system_canonical(
user_id: int, system_id: int, canonical_id: int | None,
) -> System | None:
"""Map (or unmap) one project System onto a catalog entry.
Authorised by the PROJECT, not the catalog: mapping changes the project's
row, so project write access is the right gate (rule 78 — never a bare
owner filter). Passing None clears the mapping.
Touches `canonical_id` and nothing else — the System's own name, charter
and record associations are left exactly as they are.
"""
if canonical_id is not None and await get_canonical_system(canonical_id) is None:
return None
async with async_session() as session:
system = await session.get(System, system_id)
if system is None or system.deleted_at is not None:
return None
if not await access.can_write_project(user_id, system.project_id):
return None
system.canonical_id = canonical_id
system.updated_at = datetime.now(timezone.utc)
await session.commit()
await session.refresh(system)
return system
async def propose_mappings(user_id: int, project_id: int) -> list[dict]:
"""Suggest a catalog entry for each of a project's UNMAPPED Systems.
Returns proposals, never applied changes — `set_system_canonical` is the
only thing that writes. Each carries a `basis` so the reviewer knows what
they are approving:
- `exact` — the two names reduce to the same slug. Mechanical.
- `overlap` — they share a meaningful word ("CI & runners" / "CI &
Release"). A judgment call, and the reason this is a proposal at all.
A System with no plausible match simply gets no proposal: unmapped is a
perfectly good resting state, so silence here is an answer, not a gap.
"""
if not await access.can_read_project(user_id, project_id):
return []
catalog = await list_canonical_systems()
if not catalog:
return []
async with async_session() as session:
result = await session.execute(
select(System).where(
System.project_id == project_id,
System.canonical_id.is_(None),
System.deleted_at.is_(None),
).order_by(System.order_index.asc(), System.created_at.asc())
)
systems = list(result.scalars().all())
by_slug = {entry.slug: entry for entry in catalog}
proposals: list[dict] = []
for system in systems:
slug = canonical_slug(system.name)
if not slug:
continue
exact = by_slug.get(slug)
if exact is not None:
match, basis, score = exact, "exact", 1.0
else:
local = _tokens(slug)
scored = [
(len(local & _tokens(entry.slug)) / max(len(local | _tokens(entry.slug)), 1), entry)
for entry in catalog
]
# Any shared meaningful word is enough to ASK. The threshold is
# deliberately generous because a wrong proposal costs one click
# and a missing one costs a mapping nobody thinks to make again.
best_score, best = max(scored, key=lambda pair: pair[0])
if best_score <= 0:
continue
match, basis, score = best, "overlap", round(best_score, 3)
proposals.append({
"system_id": system.id,
"system_name": system.name,
"canonical_id": match.id,
"canonical_name": match.name,
"basis": basis,
"score": score,
})
proposals.sort(key=lambda p: (-p["score"], p["system_name"]))
return proposals
+31 -19
View File
@@ -14,38 +14,39 @@ from scribe.models import async_session
from scribe.models.note import Note
from scribe.models.system import RecordSystem, System
from scribe.services import access
from scribe.services import canonical_systems as canonical_systems_svc
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 standard_systems() -> list[tuple[str, str]]:
"""The standard cross-project vocabulary (#2798) as (name, charter) pairs.
Reads the GLOBAL canonical catalog (milestone 307). This was a tuple
constant in this module until the catalog became a table: a constant
cannot be a foreign key, so nothing outside a project could reference an
area, and the list only ever applied on the inception-seed path — which is
how three spellings of "CI & Release" reached one instance anyway.
"""
return [(entry.name, entry.description or "") for entry in
await canonical_systems_svc.list_canonical_systems()]
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."""
the project's own judgment take it from there. [] without write access.
Seeded Systems are mapped to their catalog entry as they are created, so a
project born this way needs no reconciliation pass later."""
if await list_systems(user_id, project_id, include_archived=True):
return []
out: list[System] = []
for index, (name, charter) in enumerate(STANDARD_SYSTEMS):
for index, entry in enumerate(await canonical_systems_svc.list_canonical_systems()):
system = await create_system(
user_id, project_id, name, description=charter, order_index=index,
user_id, project_id, entry.name, description=entry.description,
order_index=index, canonical_id=entry.id,
)
if system is None:
break
@@ -60,8 +61,14 @@ async def create_system(
description: str | None = None,
color: str | None = None,
order_index: int = 0,
canonical_id: int | None = None,
) -> System | None:
"""Create a System. None if the user can't write the project."""
"""Create a System. None if the user can't write the project.
`canonical_id` maps the new System onto the global catalog; leaving it None
is fine — an unmapped System is fully usable, and the mapping can be
proposed later (services/canonical_systems.propose_mappings).
"""
if not await access.can_write_project(user_id, project_id):
return None
async with async_session() as session:
@@ -72,6 +79,7 @@ async def create_system(
description=description,
color=color,
order_index=order_index,
canonical_id=canonical_id,
)
session.add(system)
await session.commit()
@@ -110,6 +118,10 @@ async def list_systems(
async def update_system(user_id: int, system_id: int, **fields: object) -> System | None:
"""Update a System if the user can write its project."""
# canonical_id is deliberately NOT settable here: canonical_systems.
# set_system_canonical is its single writer, because it also validates the
# catalog entry is live. Two entry points onto one column is the drift this
# table exists to end.
allowed = {"name", "description", "color", "status", "order_index"}
async with async_session() as session:
system = await session.get(System, system_id)
+18 -5
View File
@@ -57,9 +57,22 @@ def test_normalize_choices_is_canonical_and_complete():
"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)
def test_standard_systems_vocabulary_reads_the_catalog_not_a_constant():
"""The vocabulary moved from a module constant to the global catalog table
(milestone 307): a constant cannot be a foreign key, so nothing outside a
project could reference an area. The seed and the bootstrap ask must both
read the table, or the list they show and the list they mint diverge."""
import inspect
from scribe.services import systems as systems_svc
assert not hasattr(systems_svc, "STANDARD_SYSTEMS"), (
"the constant is gone — the catalog table is the single source"
)
source = inspect.getsource(systems_svc.seed_standard_systems)
assert "list_canonical_systems" in source
assert "canonical_id=entry.id" in source, (
"a seeded System must be mapped as it is created, or a project born "
"from the standard set still needs a reconciliation pass"
)
+7 -2
View File
@@ -13,6 +13,7 @@ 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 canonical_systems as canonical_svc
from scribe.services import systems as systems_svc
from tests.helpers import ensure_user
@@ -62,7 +63,11 @@ async def test_decide_applies_every_effect_and_records_last(seeded):
})
assert out["effects"]["excluded"] == [seeded["always"]]
assert out["effects"]["subscribed"] == [seeded["other"]]
assert len(out["effects"]["systems_seeded"]) == len(systems_svc.STANDARD_SYSTEMS)
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.
seeded_systems = await systems_svc.list_systems(owner, pid)
assert all(s.canonical_id is not None for s in seeded_systems)
# The exclusion is total: the project's always-on set is empty, the
# departure is named, the subscription binds.
@@ -81,7 +86,7 @@ async def test_decide_applies_every_effect_and_records_last(seeded):
# 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)
assert len(await systems_svc.list_systems(owner, pid)) == len(catalog)
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"]
+64
View File
@@ -0,0 +1,64 @@
"""The canonical-slug matcher — the line between what maps mechanically and
what a human is asked to confirm (milestone 307, note 3026).
These cases are the real drift found across the author's own instance: one area
carrying three spellings, and two areas that LOOK alike and are not the same.
Getting the boundary wrong in either direction is a silent failure — a missed
mapping nobody thinks to make again, or a cross-project record surfacing in the
wrong project.
"""
import pytest
from scribe.services.canonical_systems import canonical_slug
@pytest.mark.parametrize("name", ["CI & Release", "CI and Release", "CI & release",
" ci and release "])
def test_spelling_variants_of_one_area_collapse_to_one_key(name):
"""Case, spacing, punctuation and "&" vs "and" are not real differences.
All three of the first spellings were live in different projects at once."""
assert canonical_slug(name) == "ci-and-release"
@pytest.mark.parametrize("name,expected", [
("Auth & Access", "auth-and-access"),
("Data Model & Storage", "data-model-and-storage"),
("UI & Design", "ui-and-design"),
("Background Jobs", "background-jobs"),
("Observability", "observability"),
])
def test_the_seeded_vocabulary_slugs_match_the_migration(name, expected):
"""Migration 0087 writes these slugs literally. If the function and the
migration disagree, every seeded entry becomes unreachable by exact match
and every mapping silently degrades to a proposal."""
assert canonical_slug(name) == expected
@pytest.mark.parametrize("a,b", [
("CI & Release", "CI & runners"),
("Auth & Access", "Auth & Accounts"),
("UI & Design", "Frontend (Vue app)"),
("UI & Design", "Web Shell and Theme"),
])
def test_genuinely_different_names_do_not_collapse(a, b):
"""These pairs may or may not be the same area — that is a judgment call,
so they must NOT map automatically. They reach the operator as proposals."""
assert canonical_slug(a) != canonical_slug(b)
def test_a_nameless_system_yields_no_key():
"""An empty slug is the one value callers must special-case: two unnameable
Systems must not map onto each other. Both propose_mappings and find_by_name
bail on a falsy slug for this reason."""
assert canonical_slug("") == ""
assert canonical_slug(" ") == ""
assert canonical_slug("---") == ""
def test_a_punctuation_separator_is_not_read_as_the_word_and():
""""CI/Release" is a real spelling and it does NOT collapse onto
"CI & Release" — only "&" carries that meaning. It still reaches the
operator through the overlap path, where both words match; what it must not
do is map itself silently."""
assert canonical_slug("CI/Release") == "ci-release"
assert canonical_slug("CI/Release") != canonical_slug("CI & Release")