feat(issues): S4b backend — REST task issue fields + dashboard open-issues

Closes the REST gap S4b's UI needs (S2 only extended MCP tools):
- routes/tasks.py: create/update accept system_ids (set-semantics) + arose_from_id;
  GET/create/update return the task's associated systems. kind=issue already
  flowed via task_kind. Associations set via services/systems (ACL-checked;
  can_write_note already gated).
- services/dashboard.py: _open_issues section (owner-scoped, ranked like other
  task lists, capped) added to build_dashboard. Dashboard test updated for the
  new key.

Refs plan 825 (S4b, backend half).

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

src/scribe/routes/tasks.py

@@ -7,6 +7,7 @@ from scribe.auth import login_required, get_current_user_id
 from scribe.models.note import TaskPriority, TaskStatus
 from scribe.routes.utils import not_found, parse_iso_date, parse_pagination
 from scribe.services.access import can_write_note
+from scribe.services import systems as systems_svc
 from scribe.services.embeddings import upsert_note_embedding
 from scribe.services.notes import (
     create_note,
@@ -136,11 +137,16 @@ async def create_task_route():
         parent_id=data.get("parent_id"),
         recurrence_rule=recurrence_rule if recurrence_rule is not _UNSET else None,
         task_kind=data.get("kind", "work"),
+        arose_from_id=data.get("arose_from_id"),
     )
+    if data.get("system_ids") is not None:
+        await systems_svc.set_record_systems(uid, task.id, data["system_ids"])
     text = f"{task.title}\n{task.body}".strip() if task.body else (task.title or "")
     if text:
         asyncio.create_task(upsert_note_embedding(task.id, uid, text))
-    return jsonify(task.to_dict()), 201
+    out = task.to_dict()
+    out["systems"] = [s.to_dict() for s in await systems_svc.list_record_systems(uid, task.id)]
+    return jsonify(out), 201
 
 
 @tasks_bp.route("/planning", methods=["POST"])
@@ -174,6 +180,7 @@ async def get_task_route(task_id: int):
     if task.parent_id:
         parent = await get_note(uid, task.parent_id)
         data["parent_title"] = parent.title if parent else None
+    data["systems"] = [s.to_dict() for s in await systems_svc.list_record_systems(uid, task_id)]
     return jsonify(data)
 
 
@@ -223,7 +230,7 @@ async def update_task_route(task_id: int):
     if "tags" in data:
         fields["tags"] = data["tags"]
 
-    for key in ("project_id", "milestone_id", "parent_id"):
+    for key in ("project_id", "milestone_id", "parent_id", "arose_from_id"):
         if key in data:
             fields[key] = data[key]
 
@@ -236,10 +243,14 @@ async def update_task_route(task_id: int):
     task = await update_note(task_note.user_id, task_id, **fields)
     if task is None:
         return not_found("Task")
+    if data.get("system_ids") is not None:
+        await systems_svc.set_record_systems(uid, task_id, data["system_ids"])
     text = f"{task.title}\n{task.body}".strip() if task.body else (task.title or "")
     if text:
         asyncio.create_task(upsert_note_embedding(task.id, task_note.user_id, text))
-    return jsonify(task.to_dict())
+    out = task.to_dict()
+    out["systems"] = [s.to_dict() for s in await systems_svc.list_record_systems(uid, task_id)]
+    return jsonify(out)
 
 
 @tasks_bp.route("/<int:task_id>/status", methods=["PATCH"])

src/scribe/services/dashboard.py

@@ -22,6 +22,7 @@ logger = logging.getLogger(__name__)
 N_PROJECTS = 3        # most-recently-active projects shown
 TASKS_PER_GROUP = 5   # open-task cap per milestone / no-milestone group
 RECENT_DONE_LIMIT = 8 # recently-completed tasks shown
+OPEN_ISSUES_LIMIT = 10 # open issues shown on the dashboard
 WINDOW_DAYS = 7       # look-back (done) / look-ahead (events) window
 _OPEN = ["todo", "in_progress"]
 
@@ -49,11 +50,42 @@ async def _safe(coro, empty):
         return empty
 
 
+async def _open_issues(user_id: int) -> list[dict]:
+    """Open issues (task_kind='issue', not done/cancelled) across the owner's
+    projects, ranked like the other task lists. Owner-scoped, matching the rest
+    of the dashboard."""
+    async with async_session() as session:
+        rows = await session.execute(
+            select(
+                Note.id, Note.title, Note.status, Note.priority,
+                Note.project_id, Project.title,
+            )
+            .join(Project, Project.id == Note.project_id, isouter=True)
+            .where(
+                Note.user_id == user_id,
+                Note.task_kind == "issue",
+                Note.status.in_(_OPEN),
+                Note.deleted_at.is_(None),
+            )
+            .order_by(*_open_order())
+            .limit(OPEN_ISSUES_LIMIT)
+        )
+        return [
+            {
+                "id": iid, "title": title, "status": status,
+                "priority": priority or "none",
+                "project_id": pid, "project_title": pname,
+            }
+            for iid, title, status, priority, pid, pname in rows.fetchall()
+        ]
+
+
 async def build_dashboard(user_id: int) -> dict:
     return {
         "active_projects": await _safe(_active_projects(user_id), []),
         "recently_completed": await _safe(_recently_completed(user_id), []),
         "upcoming_events": await _safe(_upcoming_events(user_id), []),
+        "open_issues": await _safe(_open_issues(user_id), []),
         "week_stats": await _safe(_week_stats(user_id), {}),
     }
 

tests/test_services_dashboard.py

@@ -43,12 +43,14 @@ async def test_build_dashboard_composes_sections():
     with patch.object(dash, "_active_projects", AsyncMock(return_value=["P"])), \
          patch.object(dash, "_recently_completed", AsyncMock(return_value=["done"])), \
          patch.object(dash, "_upcoming_events", AsyncMock(return_value=["evt"])), \
+         patch.object(dash, "_open_issues", AsyncMock(return_value=["iss"])), \
          patch.object(dash, "_week_stats", AsyncMock(return_value={"open_total": 4})):
         out = await dash.build_dashboard(user_id=1)
     assert out == {
         "active_projects": ["P"],
         "recently_completed": ["done"],
         "upcoming_events": ["evt"],
+        "open_issues": ["iss"],
         "week_stats": {"open_total": 4},
     }
 
@@ -59,6 +61,7 @@ async def test_build_dashboard_isolates_failing_section():
     with patch.object(dash, "_active_projects", AsyncMock(side_effect=RuntimeError("db down"))), \
          patch.object(dash, "_recently_completed", AsyncMock(return_value=["done"])), \
          patch.object(dash, "_upcoming_events", AsyncMock(return_value=[])), \
+         patch.object(dash, "_open_issues", AsyncMock(return_value=[])), \
          patch.object(dash, "_week_stats", AsyncMock(return_value={})):
         out = await dash.build_dashboard(user_id=1)
     # failing section degrades to its empty default; others still populate