"""System CRUD + record-association MCP tools — wrappers over services/systems.py. A System is a per-project, reusable, self-describing subsystem/area that any record (note, task, or issue) can be associated with — so research, build-work, and corrective work line up under the same area and recurring problem-spots are visible. The service enforces the multi-user ACL (project permission); these tools are thin wrappers. Sentinels (match the milestone/task tool conventions): - name="" / description="" / color="" / status="" → "leave unchanged" on update - order_index=-1 → "leave unchanged" on update (0 is a valid order_index) """ 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 # Below this, a project is young enough that the mild "which area is this # about?" question stays proportionate; at or above it, a zero-Systems project # has demonstrated that the question never converts (#2683 — Minstrel reached # 282 records without a single System) and the ask escalates. _BOOTSTRAP_MIN_RECORDS = 20 _BOOTSTRAP_TITLES = 6 # The standard vocabulary (#2798): area names that recur across software # projects, offered so "CI & Release" means the same thing in every project # on the instance. Consistency comes from the shared names — NOT from asking # the operator to approve each System; agents mint directly. Generic by # 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 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: """The escalated vocabulary-bootstrap ask for a mature zero-Systems project. The generic zero-state question habituates: identical on every record, maximal in scope ("invent the taxonomy"), asked at wrap-up time — so organic sessions skip it forever and only audit-shaped sessions ever mint (#2683). What separates the nudges that convert from the prose that doesn't (the duplicate gate, the prior-art "already defined in 2 files") is the project's OWN evidence in the ask — so this one carries the record count and the recent titles, and asks for a concrete deliverable: create a starter set directly, preferring the standard cross-project names. Deliberately NOT an approval flow (#2798): the operator is not a permission gate for vocabulary; the standard names carry consistency. Self-retiring by construction: callers only reach for it while the project has zero Systems, so the first create_system ends it everywhere. Returns None below the record threshold or on any failure (fail-open — a hint must never break the call it rides on). """ try: recent, total = await notes_svc.list_notes( user_id, project_id=project_id, sort="updated_at", limit=_BOOTSTRAP_TITLES, ) except Exception: return None if total < _BOOTSTRAP_MIN_RECORDS: return 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 " "invisible. Bootstrap the vocabulary now, in this session, without " "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 " 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." ) async def untagged_systems_hint(user_id: int, project_id: int) -> str | None: """The Systems question, for an untagged project record. Not a tool — attach_systems() rides this on tool responses so the question arrives in-band at the exact moment a record is touched. It is ONE question regardless of vocabulary state ("which area is this about, and is it modelled?"); only the vocabulary listing varies, because an empty vocabulary is not an exemption — it is the question at its most urgent (#2562, #2569). Instruction prose alone demonstrably doesn't fire at write time; in-band behavior (the duplicate gate) does. In a MATURE zero-Systems project the question escalates to the bootstrap ask instead (#2683): the mild form demonstrably never converts there, and an evidence-carrying, deliverable-shaped ask is the form that does. """ # Fail-open like the dedup gate: a hint must never break the call. try: systems = await systems_svc.list_systems(user_id, project_id) except Exception: return None if systems: vocab = "Existing Systems: " + ", ".join( f"#{s.id} {s.name}" for s in systems ) + "." else: ask = await bootstrap_systems_ask(user_id, project_id) if ask: return ask vocab = "This project has no Systems yet." return ( "This record is untagged — which area(s) of the project is it about? " f"{vocab} Tag it (system_ids=[...]) — cross-cutting records like " "audits take several; create_system any area it concerns that isn't " "modelled yet (a sweep that names areas is the moment to mint them, " "and the duplicate gate guards against sprawl). Leave it untagged " "only if it is genuinely about no particular area." ) async def attach_systems( caller_id: int, owner_id: int, data: dict, note_id: int, project_id: int | None, ) -> None: """Attach a record's Systems to its payload — or the Systems question. ONE seam for every tool that returns a project record, read or write: a tagged record shows its areas (the touching-a-System reflex needs the affiliation visible on read, not just settable on write), an untagged project record carries the question instead. Neither field is ever attached empty (same reasoning as supersession_svc.attach_relations / #2483 — a field that always says nothing trains readers to skip fields). The hint goes only to the record's owner: tagging someone else's record in someone else's project is not the caller's call to make. Fail-open — decoration must never break the call it rides on. """ try: systems = await systems_svc.list_record_systems(owner_id, note_id) if systems: data["systems"] = [s.to_dict() for s in systems] elif project_id and caller_id == owner_id: hint = await untagged_systems_hint(owner_id, project_id) if hint: data["systems_hint"] = hint except Exception: pass async def create_system( project_id: int, name: str, description: str = "", color: str = "", ) -> dict: """Create a System (a reusable, self-describing subsystem/area) in a project. Associate records with it via the `system_ids` arg on create/update_task and create/update_note. Create one the moment two records would share an area that has no System yet — the same two-or-more test snippets use. Don't wait to be asked to name an area that plainly exists in the code, and don't route the creation through operator approval — minting vocabulary is the agent's call (#2798); an unmodelled area means every record about it stays untaggable. An audit or sweep that walks the codebase is a DISCOVERY moment: mint the Systems it names as it names them — the duplicate gate below, plus reviewing the existing list, is what guards against sprawl, not holding back. Prefer the standard cross-project names where the area fits one (CI & Release, Auth & Access, Data Model & Storage, API Surface, UI & Design, Import & Export, Background Jobs, Observability) so the same word means the same thing in every project. Give each one a one-paragraph charter, not just a label: the description is what tells a later session whether a record belongs here. Args: project_id: The project this system belongs to (required). name: Short label (required). description: What the system is and how it's used — a name is rarely enough. color: Optional UI accent (hex), or empty. Duplicate-gated like the other creates: if a System with the same normalized name already exists in this project (archived included), the call returns {"duplicate": true, "existing_id": ...} instead of creating — tag records to that one, or update_system it if its charter needs work. Also mapped against the GLOBAL area catalog, so the same word means the same thing in every project (milestone 307). A name that IS a catalog area up to spelling ("CI and Release" vs "CI & Release") is mapped for you and the response says so. A name that merely RESEMBLES one comes back with `canonical_suggestion` — an offer, not a decision: apply it with map_system_to_canonical if it really is that area, ignore it if this is a project-specific area. Either way the System is created; the catalog never blocks a name. """ uid = current_user_id() assessment = await systems_svc.assess_system_name(uid, project_id, name) duplicate = assessment["duplicate"] if duplicate: return { "duplicate": True, "existing_id": duplicate["id"], "message": ( f"System '{duplicate['name']}' (#{duplicate['id']}) already " "covers this area in this project. Tag records to it with " "system_ids, or update_system it if the charter needs " "revising — a second System with the same name would split " "the area's records across two piles." ), } # An exact match is mechanical, so it is applied; an overlap is a judgment # call, so it is only offered (see services/canonical_systems). canonical = assessment["canonical"] applied = canonical["id"] if canonical and canonical["basis"] == "exact" else None system = await systems_svc.create_system( uid, project_id=project_id, name=name, description=description or None, color=color or None, canonical_id=applied, ) if system is None: raise ValueError(f"cannot create system in project {project_id} (no write access)") out = system.to_dict() if applied: out["canonical_note"] = ( f"Mapped to the global area '{canonical['name']}' — the same " "spelling-insensitive name. Your System keeps the name you gave it." ) elif canonical: out["canonical_suggestion"] = { **canonical, "message": ( f"The global catalog has '{canonical['name']}', which may be " f"this same area. If it is, map_system_to_canonical(" f"{system.id}, {canonical['id']}) so records and rules about " "this area line up across projects. If this area is specific " "to this project, ignore it — unmapped is a valid state." ), } return out async def list_systems(project_id: int, include_archived: bool = False) -> dict: """List a project's systems (active by default), ordered by order_index. Pass include_archived=True to include archived systems. """ uid = current_user_id() rows = await systems_svc.list_systems(uid, project_id, include_archived=include_archived) return {"systems": [s.to_dict() for s in rows]} async def get_system(system_id: int) -> dict: """Fetch a System plus the records associated with it. Returns the system, plus its associated records split into `issues`, `tasks` (work/plan), and `notes`. """ uid = current_user_id() system = await systems_svc.get_system(uid, system_id) if system is None: raise ValueError(f"system {system_id} not found") records = await systems_svc.list_records_for_system(uid, system_id) issues, tasks, notes = [], [], [] for r in records: d = r.to_dict() if r.status is None: notes.append(d) elif r.task_kind == "issue": issues.append(d) else: tasks.append(d) data = system.to_dict() data["issues"] = issues data["tasks"] = tasks data["notes"] = notes return data async def update_system( system_id: int, name: str = "", description: str = "", color: str = "", status: str = "", order_index: int = -1, ) -> dict: """Update a System. Only explicitly provided fields change. Args: status: 'active' or 'archived'. Archive a system to retire it without losing history; archived systems hide from default lists. order_index: display position (0-based); -1 = leave unchanged. """ uid = current_user_id() fields: dict = {} if name: fields["name"] = name if description: fields["description"] = description if color: fields["color"] = color if status: fields["status"] = status if order_index >= 0: fields["order_index"] = order_index system = await systems_svc.update_system(uid, system_id, **fields) if system is None: raise ValueError(f"system {system_id} not found or no write access") return system.to_dict() async def list_system_records( system_id: int, kind: str = "", open_only: bool = False ) -> dict: """Everything filed under one System — the way to READ a subsystem. Reach for this when investigating a specific area: it returns the notes, tasks, issues and snippets someone deliberately tagged to it — the subsystem's accumulated record, unranked. Start with its reference note if one exists (titled "«System» — reference"); that is the living state, and the rest is history and open work around it. For a ranked cut of the same slice, search(system_id=...) filters semantic search to this association. Args: kind: filter by task_kind — 'issue', 'work', 'spike' (or the retired 'plan'). Omit for all. open_only: limit to tasks not done/cancelled (e.g. open issues only). """ uid = current_user_id() rows = await systems_svc.list_records_for_system( uid, system_id, kind=kind or None, open_only=open_only, ) return {"records": [r.to_dict() for r in rows]} async def delete_system(system_id: int) -> dict: """Soft-delete a System (recoverable). Its record associations are removed.""" uid = current_user_id() ok = await systems_svc.delete_system(uid, system_id) if not ok: raise ValueError(f"system {system_id} not found or no write access") 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, list_systems, get_system, update_system, list_system_records, delete_system, list_canonical_systems, propose_canonical_mappings, map_system_to_canonical, ): mcp.tool(name=fn.__name__)(fn)