feat(issues): S2 MCP tools — system CRUD + issue/system wiring

Second slice of Issues + Systems (spec #825).

New mcp/tools/systems.py: create_system, list_systems, get_system (records
split into issues/tasks/notes), update_system (incl. archive via status),
list_system_records (kind/open_only filters), delete_system. Registered in
register_all; read tools (get_system, list_systems, list_system_records) added
to the read-only-key allowlist (write tools default-deny).

create_task/update_task: kind now accepts 'issue'; new system_ids (set-semantics
associations) and arose_from_id (provenance, 0=unchanged/-1=clear) args.
create_note/update_note: new system_ids arg (notes associate with systems too).
services/notes.create_note: arose_from_id passthrough (update_note already
handles it via setattr).

Tests: MCP system tools + create_task issue-wiring (kind/provenance/systems),
service layer mocked.

Refs plan 825 (S2).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

src/scribe/mcp/server.py

@@ -185,6 +185,7 @@ _READ_ONLY_TOOLS = frozenset({
     "list_persons", "list_places", "list_projects", "list_rulebooks",
     "list_rules", "list_tags", "list_tasks", "list_topics", "list_trash",
     "list_always_on_rules", "search",
+    "get_system", "list_systems", "list_system_records",
 })
 
 

src/scribe/mcp/tools/__init__.py

@@ -5,7 +5,7 @@ to a FastMCP instance. `register_all(mcp)` is the single entry point called
 from `mcp.server.build_mcp_server`.
 """
 from scribe.mcp.tools import (
-    entities, events, milestones, notes, processes, projects, recent, repos, rulebooks, search, tags, tasks, trash,
+    entities, events, milestones, notes, processes, projects, recent, repos, rulebooks, search, systems, tags, tasks, trash,
 )
 
 
@@ -16,6 +16,7 @@ def register_all(mcp) -> None:
     tasks.register(mcp)
     projects.register(mcp)
     milestones.register(mcp)
+    systems.register(mcp)
     events.register(mcp)
     tags.register(mcp)
     recent.register(mcp)

src/scribe/mcp/tools/notes.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 
 from scribe.mcp._context import current_user_id
 from scribe.services import notes as notes_svc
+from scribe.services import systems as systems_svc
 from scribe.services import trash as trash_svc
 
 
@@ -68,6 +69,7 @@ async def create_note(
     body: str = "",
     tags: list[str] | None = None,
     project_id: int = 0,
+    system_ids: list[int] | None = None,
 ) -> dict:
     """Create a new note in Scribe.
 
@@ -76,6 +78,8 @@ async def create_note(
         body: Markdown content. Supports [[wikilinks]] to other notes by title.
         tags: List of plain-string tags without # prefix, e.g. ["python", "ideas"].
         project_id: Associate with a project (use 0 for no project / orphan note).
+        system_ids: Ids of the project's Systems to associate this note with
+            (e.g. research about a subsystem). See list_systems / create_system.
 
     Returns the created note object including its assigned id.
     """
@@ -87,7 +91,14 @@ async def create_note(
         tags=tags,
         project_id=project_id or None,
     )
-    return note.to_dict()
+    if system_ids:
+        await systems_svc.set_record_systems(uid, note.id, system_ids)
+    data = note.to_dict()
+    if system_ids:
+        data["systems"] = [
+            s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id)
+        ]
+    return data
 
 
 async def update_note(
@@ -96,6 +107,7 @@ async def update_note(
     body: str = "",
     tags: list[str] | None = None,
     project_id: int = 0,
+    system_ids: list[int] | None = None,
 ) -> dict:
     """Update an existing Scribe note. Only explicitly provided fields are changed.
 
@@ -105,6 +117,8 @@ async def update_note(
         body: New markdown body, or omit to leave unchanged.
         tags: Replaces the full tag list. Pass [] to clear all tags. Omit to leave unchanged.
         project_id: New project association. Omit (or pass 0) to leave unchanged.
+        system_ids: Replace this note's System associations with these ids
+            (set-semantics). None = leave unchanged; [] = clear all.
     """
     uid = current_user_id()
     fields: dict = {}
@@ -119,7 +133,14 @@ async def update_note(
     note = await notes_svc.update_note(uid, note_id, **fields)
     if note is None:
         raise ValueError(f"note {note_id} not found")
-    return note.to_dict()
+    if system_ids is not None:
+        await systems_svc.set_record_systems(uid, note_id, system_ids)
+    data = note.to_dict()
+    if system_ids is not None:
+        data["systems"] = [
+            s.to_dict() for s in await systems_svc.list_record_systems(uid, note_id)
+        ]
+    return data
 
 
 async def delete_note(note_id: int) -> dict:

src/scribe/mcp/tools/systems.py

@@ -0,0 +1,150 @@
+"""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 systems as systems_svc
+
+
+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.
+
+    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.
+    """
+    uid = current_user_id()
+    system = await systems_svc.create_system(
+        uid, project_id=project_id, name=name,
+        description=description or None, color=color or None,
+    )
+    if system is None:
+        raise ValueError(f"cannot create system in project {project_id} (no write access)")
+    return system.to_dict()
+
+
+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:
+    """List records associated with a System.
+
+    Args:
+        kind: filter by task_kind — 'issue', 'work', or '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."}
+
+
+def register(mcp) -> None:
+    for fn in (
+        create_system,
+        list_systems,
+        get_system,
+        update_system,
+        list_system_records,
+        delete_system,
+    ):
+        mcp.tool(name=fn.__name__)(fn)

src/scribe/mcp/tools/tasks.py

@@ -22,6 +22,7 @@ from scribe.mcp._context import current_user_id
 from scribe.services import notes as notes_svc
 from scribe.services import planning as planning_svc
 from scribe.services import rulebooks as rulebooks_svc
+from scribe.services import systems as systems_svc
 from scribe.services import task_logs as task_logs_svc
 from scribe.services import trash as trash_svc
 
@@ -41,7 +42,7 @@ async def list_tasks(
             whenever a project is in scope so you list that project's tasks, not
             every project's. 0 = no filter (all projects — use only for a
             deliberate cross-project view).
-        kind: Filter by task kind — 'work' or 'plan'. Omit (empty) for all kinds.
+        kind: Filter by task kind — 'work', 'plan', or 'issue'. Omit (empty) for all kinds.
 
     Results are ordered by last-updated descending.
     """
@@ -101,6 +102,8 @@ async def create_task(
     parent_id: int = 0,
     tags: list[str] | None = None,
     kind: str = "work",
+    system_ids: list[int] | None = None,
+    arose_from_id: int = 0,
 ) -> dict:
     """Create a new task in Scribe.
 
@@ -113,7 +116,13 @@ async def create_task(
         milestone_id: Place within a project milestone (0 = no milestone).
         parent_id: Make this a sub-task of another task (0 = top-level).
         tags: List of plain-string tags without # prefix.
-        kind: 'work' (default) or 'plan'. Prefer the start_planning tool to create plans.
+        kind: 'work' (default), 'plan', or 'issue'. An issue is corrective work —
+            a problem you fixed or are fixing; record symptom → root cause → fix
+            in the body. Prefer start_planning to create plans.
+        system_ids: Ids of the project's Systems (reusable subsystem/area
+            objects; see list_systems / create_system) to associate this task with.
+        arose_from_id: For an issue, the id of the task/feature it arose from
+            (provenance). 0 = none.
     """
     uid = current_user_id()
     note = await notes_svc.create_note(
@@ -127,8 +136,16 @@ async def create_task(
         parent_id=parent_id or None,
         tags=tags,
         task_kind=kind,
+        arose_from_id=arose_from_id or None,
     )
-    return note.to_dict()
+    if system_ids:
+        await systems_svc.set_record_systems(uid, note.id, system_ids)
+    data = note.to_dict()
+    if system_ids:
+        data["systems"] = [
+            s.to_dict() for s in await systems_svc.list_record_systems(uid, note.id)
+        ]
+    return data
 
 
 async def update_task(
@@ -139,6 +156,8 @@ async def update_task(
     priority: str = "",
     project_id: int = 0,
     milestone_id: int = 0,
+    system_ids: list[int] | None = None,
+    arose_from_id: int = 0,
 ) -> dict:
     """Update an existing Scribe task. Only explicitly provided fields are changed.
 
@@ -154,6 +173,10 @@ async def update_task(
             its project; also clears the milestone), positive = set.
         milestone_id: New milestone. 0 = leave unchanged, -1 = clear (remove
             from its milestone), positive = set.
+        system_ids: Replace this task's System associations with these ids
+            (set-semantics). None = leave unchanged; [] = clear all.
+        arose_from_id: Provenance (issue → originating task). 0 = leave unchanged,
+            -1 = clear, positive = set.
     """
     uid = current_user_id()
     fields: dict = {}
@@ -175,10 +198,21 @@ async def update_task(
         fields["milestone_id"] = None
     elif milestone_id:
         fields["milestone_id"] = milestone_id
+    if arose_from_id == -1:
+        fields["arose_from_id"] = None
+    elif arose_from_id:
+        fields["arose_from_id"] = arose_from_id
     note = await notes_svc.update_note(uid, task_id, **fields)
     if note is None:
         raise ValueError(f"task {task_id} not found")
-    return note.to_dict()
+    if system_ids is not None:
+        await systems_svc.set_record_systems(uid, task_id, system_ids)
+    data = note.to_dict()
+    if system_ids is not None:
+        data["systems"] = [
+            s.to_dict() for s in await systems_svc.list_record_systems(uid, task_id)
+        ]
+    return data
 
 
 async def add_task_log(task_id: int, content: str) -> dict:

src/scribe/services/notes.py

@@ -65,6 +65,7 @@ async def create_note(
     note_type: str = "note",
     entity_meta: dict | None = None,
     task_kind: str = "work",
+    arose_from_id: int | None = None,
 ) -> Note:
     # Validate status/priority here so the MCP create_task path (which passes
     # them straight through) can't persist an out-of-enum value that the REST
@@ -108,6 +109,7 @@ async def create_note(
             note_type=note_type,
             entity_meta=entity_meta,
             task_kind=task_kind,
+            arose_from_id=arose_from_id,
         )
         session.add(note)
         await session.commit()

tests/test_mcp_tool_systems.py

@@ -0,0 +1,75 @@
+"""MCP system tools + task/note issue wiring (service layer mocked)."""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+def _fake_system(sid=1, name="Reader", project_id=5):
+    s = MagicMock()
+    s.to_dict.return_value = {"id": sid, "name": name, "project_id": project_id}
+    s.project_id = project_id
+    return s
+
+
+@pytest.mark.asyncio
+async def test_create_system_returns_dict():
+    with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \
+         patch("scribe.mcp.tools.systems.systems_svc") as svc:
+        svc.create_system = AsyncMock(return_value=_fake_system(name="Reader"))
+        from scribe.mcp.tools.systems import create_system
+        result = await create_system(project_id=5, name="Reader", description="pdf reader")
+    assert result["name"] == "Reader"
+
+
+@pytest.mark.asyncio
+async def test_create_system_no_access_raises():
+    with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \
+         patch("scribe.mcp.tools.systems.systems_svc") as svc:
+        svc.create_system = AsyncMock(return_value=None)
+        from scribe.mcp.tools.systems import create_system
+        with pytest.raises(ValueError):
+            await create_system(project_id=5, name="Reader")
+
+
+@pytest.mark.asyncio
+async def test_get_system_splits_records_by_kind():
+    issue = MagicMock(); issue.to_dict.return_value = {"id": 10}; issue.task_kind = "issue"; issue.status = "todo"
+    work = MagicMock(); work.to_dict.return_value = {"id": 11}; work.task_kind = "work"; work.status = "todo"
+    note = MagicMock(); note.to_dict.return_value = {"id": 12}; note.task_kind = "work"; note.status = None
+    with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \
+         patch("scribe.mcp.tools.systems.systems_svc") as svc:
+        svc.get_system = AsyncMock(return_value=_fake_system(sid=3))
+        svc.list_records_for_system = AsyncMock(return_value=[issue, work, note])
+        from scribe.mcp.tools.systems import get_system
+        result = await get_system(system_id=3)
+    assert [r["id"] for r in result["issues"]] == [10]
+    assert [r["id"] for r in result["tasks"]] == [11]
+    assert [r["id"] for r in result["notes"]] == [12]
+
+
+@pytest.mark.asyncio
+async def test_update_system_not_found_raises():
+    with patch("scribe.mcp.tools.systems.current_user_id", return_value=1), \
+         patch("scribe.mcp.tools.systems.systems_svc") as svc:
+        svc.update_system = AsyncMock(return_value=None)
+        from scribe.mcp.tools.systems import update_system
+        with pytest.raises(ValueError):
+            await update_system(system_id=99, name="x")
+
+
+@pytest.mark.asyncio
+async def test_create_task_issue_sets_kind_provenance_and_systems():
+    note = MagicMock(); note.id = 50; note.to_dict.return_value = {"id": 50, "task_kind": "issue"}
+    with patch("scribe.mcp.tools.tasks.current_user_id", return_value=1), \
+         patch("scribe.mcp.tools.tasks.notes_svc") as notes_svc, \
+         patch("scribe.mcp.tools.tasks.systems_svc") as systems_svc:
+        notes_svc.create_note = AsyncMock(return_value=note)
+        systems_svc.set_record_systems = AsyncMock(return_value=[2, 3])
+        systems_svc.list_record_systems = AsyncMock(return_value=[])
+        from scribe.mcp.tools.tasks import create_task
+        result = await create_task(title="bug", kind="issue", system_ids=[2, 3], arose_from_id=9)
+    _, kwargs = notes_svc.create_note.call_args
+    assert kwargs["task_kind"] == "issue"
+    assert kwargs["arose_from_id"] == 9
+    systems_svc.set_record_systems.assert_awaited_once_with(1, 50, [2, 3])
+    assert result["id"] == 50