feat(issues): S3 REST routes — systems CRUD + project open-issues

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

routes/systems.py (nested /api/projects/<id>/...): GET/POST systems (list adds
per-system open_issue_count via one grouped query), GET/PATCH/DELETE a system
(GET returns records split into issues/tasks/notes), GET .../systems/<id>/records
(kind/open_only filters), GET .../issues (project's open issues for the project
view + dashboard roll-up). login_required; project access via get_project_for_user;
writes gated by can_write_project (clean 403); system.project_id verified to match
the path. Blueprint registered in app.py.

services/systems.py: + open_issue_counts_by_system (one grouped query) and
list_issues (project issues, open by default).

Tests: structural (blueprint registered + in app, handlers callable, service
contracts take user_id) — matches the house route-test pattern.

Refs plan 825 (S3).

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

src/scribe/app.py

@@ -29,6 +29,7 @@ from scribe.routes.rulebooks import rulebooks_bp
 from scribe.routes.plugin import plugin_bp
 from scribe.routes.trash import trash_bp
 from scribe.routes.dashboard import dashboard_bp
+from scribe.routes.systems import systems_bp
 from scribe.mcp import mount_mcp
 
 STATIC_DIR = Path(__file__).parent / "static"
@@ -91,6 +92,7 @@ def create_app() -> Quart:
     app.register_blueprint(plugin_bp)
     app.register_blueprint(trash_bp)
     app.register_blueprint(dashboard_bp)
+    app.register_blueprint(systems_bp)
 
     @app.before_request
     async def before_request():

src/scribe/routes/systems.py

@@ -0,0 +1,155 @@
+"""System routes nested under /api/projects/<project_id>/systems, plus the
+project's open-issues list. A System is a per-project, reusable subsystem/area
+that records (notes/tasks/issues) associate with."""
+import logging
+
+from quart import Blueprint, jsonify, request
+
+from scribe.auth import login_required, get_current_user_id
+from scribe.routes.utils import not_found
+from scribe.services import systems as systems_svc
+from scribe.services.access import can_write_project
+from scribe.services.projects import get_project_for_user
+
+logger = logging.getLogger(__name__)
+
+systems_bp = Blueprint("systems", __name__, url_prefix="/api/projects")
+
+
+def _truthy(v: str | None) -> bool:
+    return (v or "").lower() in ("1", "true", "yes")
+
+
+def _split_records(records: list) -> tuple[list, list, list]:
+    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)
+    return issues, tasks, notes
+
+
+@systems_bp.route("/<int:project_id>/systems", methods=["GET"])
+@login_required
+async def list_systems_route(project_id: int):
+    uid = get_current_user_id()
+    if await get_project_for_user(uid, project_id) is None:
+        return not_found("Project")
+    systems = await systems_svc.list_systems(
+        uid, project_id, include_archived=_truthy(request.args.get("include_archived")),
+    )
+    counts = await systems_svc.open_issue_counts_by_system(uid, project_id)
+    out = []
+    for s in systems:
+        d = s.to_dict()
+        d["open_issue_count"] = counts.get(s.id, 0)
+        out.append(d)
+    return jsonify({"systems": out})
+
+
+@systems_bp.route("/<int:project_id>/systems", methods=["POST"])
+@login_required
+async def create_system_route(project_id: int):
+    uid = get_current_user_id()
+    if await get_project_for_user(uid, project_id) is None:
+        return not_found("Project")
+    if not await can_write_project(uid, project_id):
+        return jsonify({"error": "Permission denied"}), 403
+    data = await request.get_json() or {}
+    if not (data.get("name") or "").strip():
+        return jsonify({"error": "name is required"}), 400
+    system = await systems_svc.create_system(
+        uid, project_id=project_id, name=data["name"],
+        description=data.get("description"), color=data.get("color"),
+        order_index=data.get("order_index", 0),
+    )
+    if system is None:
+        return jsonify({"error": "Permission denied"}), 403
+    return jsonify(system.to_dict()), 201
+
+
+@systems_bp.route("/<int:project_id>/systems/<int:system_id>", methods=["GET"])
+@login_required
+async def get_system_route(project_id: int, system_id: int):
+    uid = get_current_user_id()
+    if await get_project_for_user(uid, project_id) is None:
+        return not_found("Project")
+    system = await systems_svc.get_system(uid, system_id)
+    if system is None or system.project_id != project_id:
+        return not_found("System")
+    issues, tasks, notes = _split_records(
+        await systems_svc.list_records_for_system(uid, system_id)
+    )
+    data = system.to_dict()
+    data["issues"], data["tasks"], data["notes"] = issues, tasks, notes
+    return jsonify(data)
+
+
+@systems_bp.route("/<int:project_id>/systems/<int:system_id>", methods=["PATCH"])
+@login_required
+async def update_system_route(project_id: int, system_id: int):
+    uid = get_current_user_id()
+    if await get_project_for_user(uid, project_id) is None:
+        return not_found("Project")
+    if not await can_write_project(uid, project_id):
+        return jsonify({"error": "Permission denied"}), 403
+    system = await systems_svc.get_system(uid, system_id)
+    if system is None or system.project_id != project_id:
+        return not_found("System")
+    data = await request.get_json() or {}
+    allowed = {"name", "description", "color", "status", "order_index"}
+    fields = {k: v for k, v in data.items() if k in allowed}
+    if "status" in fields and fields["status"] not in ("active", "archived"):
+        return jsonify({"error": "status must be 'active' or 'archived'"}), 400
+    updated = await systems_svc.update_system(uid, system_id, **fields)
+    if updated is None:
+        return not_found("System")
+    return jsonify(updated.to_dict())
+
+
+@systems_bp.route("/<int:project_id>/systems/<int:system_id>", methods=["DELETE"])
+@login_required
+async def delete_system_route(project_id: int, system_id: int):
+    uid = get_current_user_id()
+    if await get_project_for_user(uid, project_id) is None:
+        return not_found("Project")
+    if not await can_write_project(uid, project_id):
+        return jsonify({"error": "Permission denied"}), 403
+    system = await systems_svc.get_system(uid, system_id)
+    if system is None or system.project_id != project_id:
+        return not_found("System")
+    await systems_svc.delete_system(uid, system_id)
+    return "", 204
+
+
+@systems_bp.route("/<int:project_id>/systems/<int:system_id>/records", methods=["GET"])
+@login_required
+async def system_records_route(project_id: int, system_id: int):
+    uid = get_current_user_id()
+    if await get_project_for_user(uid, project_id) is None:
+        return not_found("Project")
+    system = await systems_svc.get_system(uid, system_id)
+    if system is None or system.project_id != project_id:
+        return not_found("System")
+    records = await systems_svc.list_records_for_system(
+        uid, system_id,
+        kind=request.args.get("kind") or None,
+        open_only=_truthy(request.args.get("open_only")),
+    )
+    return jsonify({"records": [r.to_dict() for r in records]})
+
+
+@systems_bp.route("/<int:project_id>/issues", methods=["GET"])
+@login_required
+async def project_issues_route(project_id: int):
+    """A project's issues (open by default — pass open_only=false for all)."""
+    uid = get_current_user_id()
+    if await get_project_for_user(uid, project_id) is None:
+        return not_found("Project")
+    open_only = request.args.get("open_only", "true").lower() in ("1", "true", "yes")
+    issues = await systems_svc.list_issues(uid, project_id, open_only=open_only)
+    return jsonify({"issues": [n.to_dict() for n in issues]})

src/scribe/services/systems.py

@@ -207,3 +207,43 @@ async def count_open_issues(user_id: int, project_id: int) -> int:
             )
         )
         return int(result.scalar() or 0)
+
+
+async def open_issue_counts_by_system(user_id: int, project_id: int) -> dict[int, int]:
+    """{system_id: open-issue count} for all systems in a project, in one query.
+    {} if no read access."""
+    if not await access.can_read_project(user_id, project_id):
+        return {}
+    async with async_session() as session:
+        rows = await session.execute(
+            select(RecordSystem.system_id, func.count(Note.id))
+            .join(Note, Note.id == RecordSystem.note_id)
+            .join(System, System.id == RecordSystem.system_id)
+            .where(
+                System.project_id == project_id,
+                Note.task_kind == "issue",
+                Note.status.isnot(None),
+                Note.status.not_in(["done", "cancelled"]),
+                Note.deleted_at.is_(None),
+            )
+            .group_by(RecordSystem.system_id)
+        )
+        return {sid: cnt for sid, cnt in rows.fetchall()}
+
+
+async def list_issues(user_id: int, project_id: int, open_only: bool = True) -> list[Note]:
+    """Issues in a project (open by default), newest first. [] if no read access."""
+    if not await access.can_read_project(user_id, project_id):
+        return []
+    async with async_session() as session:
+        query = select(Note).where(
+            Note.project_id == project_id,
+            Note.task_kind == "issue",
+            Note.status.isnot(None),
+            Note.deleted_at.is_(None),
+        )
+        if open_only:
+            query = query.where(Note.status.not_in(["done", "cancelled"]))
+        query = query.order_by(Note.updated_at.desc())
+        result = await session.execute(query)
+        return list(result.scalars().all())

tests/test_routes_systems.py

@@ -0,0 +1,38 @@
+"""Structural tests for the systems blueprint — registration + handler/service
+contracts. Full HTTP integration needs a live DB + auth the unit env lacks."""
+import inspect
+
+
+def test_systems_blueprint_registered():
+    from scribe.routes.systems import systems_bp
+    assert systems_bp.name == "systems"
+    assert systems_bp.url_prefix == "/api/projects"
+
+
+def test_systems_blueprint_registered_in_app():
+    from scribe.app import create_app
+    app = create_app()
+    assert "systems" in app.blueprints
+
+
+def test_system_handlers_callable():
+    from scribe.routes import systems as routes
+    for name in (
+        "list_systems_route", "create_system_route", "get_system_route",
+        "update_system_route", "delete_system_route", "system_records_route",
+        "project_issues_route",
+    ):
+        assert callable(getattr(routes, name))
+
+
+def test_service_functions_take_user_id():
+    """Routes must call systems services with user_id — verify the contract."""
+    from scribe.services import systems as svc
+    for fn_name in (
+        "create_system", "list_systems", "get_system", "update_system",
+        "delete_system", "list_records_for_system", "list_issues",
+        "open_issue_counts_by_system",
+    ):
+        fn = getattr(svc, fn_name)
+        assert callable(fn)
+        assert "user_id" in inspect.signature(fn).parameters