feat(issues): S1 schema — issue task_kind, System entity, associations

First slice of the Issues + Systems feature (spec #825, plan #819 T2).

Schema (migration 0065):
- task_kind CHECK expands work|plan -> work|plan|issue (same-change, rule 36)
- notes.arose_from_id: optional self-FK for issue->originating-task provenance
  (distinct from parent_id sub-task hierarchy)
- systems: per-project, self-describing (name + description) subsystem/area
- record_systems: M2M join linking any note/task/issue to systems (mutable)

Models: System + RecordSystem; note.py gains arose_from_id (+ index, to_dict).
Service services/systems.py: CRUD, archive, soft-delete, set/list associations,
records-for-system, open-issue count — all gated via services/access.py project
permissions (rule 78, no bare-owner filters). Unit tests lock the ACL gating;
the migration is exercised by CI's integration lane (alembic upgrade head).

is_task stays a derived property (status is not None) — unchanged. T1 (typing-
axis rationalization) intentionally NOT bundled; this only adds the enum value.

Refs plan 825 (S1).

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

alembic/versions/0065_issues_and_systems.py

@@ -0,0 +1,103 @@
+"""issues + systems: task_kind=issue, systems, record_systems, arose_from_id
+
+Revision ID: 0065
+Revises: 0064
+Create Date: 2026-06-14
+
+Adds the corrective-work 'issue' task_kind (same-change CHECK expand per the
+'new CHECK-enum values need a same-change migration' rule), a per-project
+self-describing System entity, a many-to-many record<->system join (any
+note/task/issue), and an issue->originating-task provenance FK.
+"""
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "0065"
+down_revision = "0064"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # 1. task_kind gains 'issue' (corrective work). DROP+ADD the CHECK in the
+    #    same change that introduces the value.
+    op.drop_constraint("notes_task_kind_check", "notes", type_="check")
+    op.create_check_constraint(
+        "notes_task_kind_check", "notes", "task_kind IN ('work','plan','issue')",
+    )
+
+    # 2. Provenance: an issue can point back at the task/feature it arose from.
+    #    Distinct from parent_id (sub-task hierarchy).
+    op.add_column("notes", sa.Column("arose_from_id", sa.Integer(), nullable=True))
+    op.create_foreign_key(
+        "fk_notes_arose_from_id", "notes", "notes",
+        ["arose_from_id"], ["id"], ondelete="SET NULL",
+    )
+    op.create_index("ix_notes_arose_from_id", "notes", ["arose_from_id"])
+
+    # 3. systems: per-project, reusable, self-describing subsystem/area.
+    op.create_table(
+        "systems",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "user_id", sa.Integer(),
+            sa.ForeignKey("users.id", ondelete="CASCADE"), nullable=False,
+        ),
+        sa.Column(
+            "project_id", sa.Integer(),
+            sa.ForeignKey("projects.id", ondelete="CASCADE"), nullable=False,
+        ),
+        sa.Column("name", sa.Text(), nullable=False, server_default=""),
+        sa.Column("description", sa.Text(), nullable=True),
+        sa.Column("color", sa.Text(), nullable=True),
+        sa.Column("status", sa.Text(), nullable=False, server_default="active"),
+        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),
+    )
+    op.create_index("ix_systems_project_id", "systems", ["project_id"])
+    op.create_check_constraint(
+        "systems_status_check", "systems", "status IN ('active','archived')",
+    )
+
+    # 4. record_systems: M2M join — any note/task/issue <-> system.
+    op.create_table(
+        "record_systems",
+        sa.Column("id", sa.Integer(), primary_key=True),
+        sa.Column(
+            "note_id", sa.Integer(),
+            sa.ForeignKey("notes.id", ondelete="CASCADE"), nullable=False,
+        ),
+        sa.Column(
+            "system_id", sa.Integer(),
+            sa.ForeignKey("systems.id", ondelete="CASCADE"), nullable=False,
+        ),
+        sa.Column(
+            "created_at", sa.DateTime(timezone=True), nullable=False,
+            server_default=sa.text("now()"),
+        ),
+        sa.UniqueConstraint("note_id", "system_id", name="uq_record_systems_note_system"),
+    )
+    op.create_index("ix_record_systems_note_id", "record_systems", ["note_id"])
+    op.create_index("ix_record_systems_system_id", "record_systems", ["system_id"])
+
+
+def downgrade() -> None:
+    op.drop_table("record_systems")
+    op.drop_table("systems")
+    op.drop_index("ix_notes_arose_from_id", table_name="notes")
+    op.drop_constraint("fk_notes_arose_from_id", "notes", type_="foreignkey")
+    op.drop_column("notes", "arose_from_id")
+    op.drop_constraint("notes_task_kind_check", "notes", type_="check")
+    op.create_check_constraint(
+        "notes_task_kind_check", "notes", "task_kind IN ('work','plan')",
+    )

src/scribe/models/__init__.py

@@ -41,3 +41,4 @@ from scribe.models.rulebook import (  # noqa: E402, F401
     Rulebook, RulebookTopic, Rule, project_rulebook_subscriptions,
 )
 from scribe.models.repo_binding import RepoBinding  # noqa: E402, F401
+from scribe.models.system import System, RecordSystem  # noqa: E402, F401

src/scribe/models/note.py

@@ -40,6 +40,12 @@ class Note(Base, TimestampMixin, SoftDeleteMixin):
     parent_id: Mapped[int | None] = mapped_column(
         Integer, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
     )
+    # Provenance: the task/feature an issue arose from. Distinct from parent_id
+    # (sub-task hierarchy) — this is "what spawned this". Only meaningful for
+    # issues; nullable for every record.
+    arose_from_id: Mapped[int | None] = mapped_column(
+        Integer, ForeignKey("notes.id", ondelete="SET NULL"), nullable=True
+    )
     project_id: Mapped[int | None] = mapped_column(
         Integer, ForeignKey("projects.id", ondelete="SET NULL"), nullable=True
     )
@@ -60,9 +66,10 @@ class Note(Base, TimestampMixin, SoftDeleteMixin):
     # Structured metadata for entity types (person/place/list)
     # Named 'entity_meta' to avoid collision with SQLAlchemy's reserved 'metadata' attribute
     entity_meta: Mapped[dict | None] = mapped_column("metadata", JSONB, nullable=True)
-    # Task sub-kind — 'work' (default) or 'plan'. Only meaningful when the note
+    # Task sub-kind — 'work' (default), 'plan', or 'issue' (corrective work).
-    # is a task (status is not None); ordinary notes keep the 'work' default and
+    # Only meaningful when the note is a task (status is not None); ordinary
-    # ignore it. Orthogonal to note_type (which is the note/entity axis).
+    # notes keep the 'work' default and ignore it. Orthogonal to note_type
+    # (which is the note/entity axis).
     task_kind: Mapped[str] = mapped_column(Text, default="work", server_default="work")
 
     __table_args__ = (
@@ -73,6 +80,7 @@ class Note(Base, TimestampMixin, SoftDeleteMixin):
         Index("ix_notes_project_id", "project_id"),
         Index("ix_notes_milestone_id", "milestone_id"),
         Index("ix_notes_note_type", "note_type"),
+        Index("ix_notes_arose_from_id", "arose_from_id"),
     )
 
     @property
@@ -95,6 +103,7 @@ class Note(Base, TimestampMixin, SoftDeleteMixin):
             ),
             "tags": self.tags or [],
             "parent_id": self.parent_id,
+            "arose_from_id": self.arose_from_id,
             "project_id": self.project_id,
             "milestone_id": self.milestone_id,
             "status": self.status,

src/scribe/models/system.py

@@ -0,0 +1,73 @@
+from sqlalchemy import ForeignKey, Index, Integer, Text, UniqueConstraint
+from sqlalchemy.orm import Mapped, mapped_column
+
+from scribe.models import Base
+from scribe.models.base import CreatedAtMixin, TimestampMixin, SoftDeleteMixin
+
+
+class System(Base, TimestampMixin, SoftDeleteMixin):
+    """A per-project, reusable, self-describing subsystem/area.
+
+    Any record (note, task, or issue) can be associated with one or more
+    systems via record_systems, so research notes, build-tasks, and corrective
+    work line up under the same area — and recurring problem-areas become
+    filterable. Self-describing (name + description) because a bare name is
+    never enough to convey what a system is or how it's used.
+    """
+    __tablename__ = "systems"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    user_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("users.id", ondelete="CASCADE")
+    )
+    project_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("projects.id", ondelete="CASCADE")
+    )
+    name: Mapped[str] = mapped_column(Text, default="", server_default="")
+    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.
+    status: Mapped[str] = mapped_column(Text, default="active", server_default="active")
+    order_index: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+
+    __table_args__ = (
+        Index("ix_systems_project_id", "project_id"),
+    )
+
+    def to_dict(self) -> dict:
+        return {
+            "id": self.id,
+            "user_id": self.user_id,
+            "project_id": self.project_id,
+            "name": self.name,
+            "description": self.description,
+            "color": self.color,
+            "status": self.status,
+            "order_index": self.order_index,
+            "created_at": self.created_at.isoformat(),
+            "updated_at": self.updated_at.isoformat(),
+        }
+
+
+class RecordSystem(Base, CreatedAtMixin):
+    """M2M association: a record (notes.id — any note/task/issue) <-> a System.
+
+    Mutable over time; uniqueness keeps a record from linking the same system
+    twice. Cascades on both sides, so deleting a record or a system clears its
+    associations.
+    """
+    __tablename__ = "record_systems"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    note_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("notes.id", ondelete="CASCADE")
+    )
+    system_id: Mapped[int] = mapped_column(
+        Integer, ForeignKey("systems.id", ondelete="CASCADE")
+    )
+
+    __table_args__ = (
+        UniqueConstraint("note_id", "system_id", name="uq_record_systems_note_system"),
+        Index("ix_record_systems_note_id", "note_id"),
+        Index("ix_record_systems_system_id", "system_id"),
+    )

src/scribe/services/systems.py

@@ -0,0 +1,209 @@
+"""System management + record<->system associations.
+
+A System is a per-project, reusable, self-describing subsystem/area. Access is
+governed by the project's permission via services/access.py (multi-user ACL
+rule) — never a bare owner filter. Records (notes/tasks/issues) link to systems
+many-to-many through record_systems, mutable over time.
+"""
+import logging
+from datetime import datetime, timezone
+
+from sqlalchemy import delete, func, select
+
+from scribe.models import async_session
+from scribe.models.note import Note
+from scribe.models.system import RecordSystem, System
+from scribe.services import access
+
+logger = logging.getLogger(__name__)
+
+
+async def create_system(
+    user_id: int,
+    project_id: int,
+    name: str,
+    description: str | None = None,
+    color: str | None = None,
+    order_index: int = 0,
+) -> System | None:
+    """Create a System. None if the user can't write the project."""
+    if not await access.can_write_project(user_id, project_id):
+        return None
+    async with async_session() as session:
+        system = System(
+            user_id=user_id,
+            project_id=project_id,
+            name=name.strip(),
+            description=description,
+            color=color,
+            order_index=order_index,
+        )
+        session.add(system)
+        await session.commit()
+        await session.refresh(system)
+        return system
+
+
+async def get_system(user_id: int, system_id: int) -> System | None:
+    """Fetch a System if the user can read its project."""
+    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_read_project(user_id, system.project_id):
+        return None
+    return system
+
+
+async def list_systems(
+    user_id: int, project_id: int, include_archived: bool = False
+) -> list[System]:
+    """A project's systems (active by default). [] if no read access."""
+    if not await access.can_read_project(user_id, project_id):
+        return []
+    async with async_session() as session:
+        query = select(System).where(
+            System.project_id == project_id,
+            System.deleted_at.is_(None),
+        )
+        if not include_archived:
+            query = query.where(System.status == "active")
+        query = query.order_by(System.order_index.asc(), System.created_at.asc())
+        result = await session.execute(query)
+        return list(result.scalars().all())
+
+
+async def update_system(user_id: int, system_id: int, **fields: object) -> System | None:
+    """Update a System if the user can write its project."""
+    allowed = {"name", "description", "color", "status", "order_index"}
+    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
+        for key, value in fields.items():
+            if key in allowed and value is not None:
+                setattr(system, key, value)
+        system.updated_at = datetime.now(timezone.utc)
+        await session.commit()
+        await session.refresh(system)
+        return system
+
+
+async def archive_system(user_id: int, system_id: int) -> System | None:
+    return await update_system(user_id, system_id, status="archived")
+
+
+async def delete_system(user_id: int, system_id: int) -> bool:
+    """Soft-delete a System (recoverable). Requires project write access."""
+    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 False
+        if not await access.can_write_project(user_id, system.project_id):
+            return False
+        system.deleted_at = datetime.now(timezone.utc)
+        await session.commit()
+        return True
+
+
+# --- record <-> system associations (many-to-many, mutable) ---
+
+async def set_record_systems(
+    user_id: int, note_id: int, system_ids: list[int]
+) -> list[int] | None:
+    """Replace a record's system associations with `system_ids` (set semantics).
+
+    Returns the resulting associated system ids, or None if the user can't write
+    the record. Silently drops ids that don't exist or the user can't read —
+    associations only link accessible systems.
+    """
+    if not await access.can_write_note(user_id, note_id):
+        return None
+    async with async_session() as session:
+        wanted: list[int] = []
+        for sid in dict.fromkeys(system_ids):  # de-dup, preserve order
+            system = await session.get(System, sid)
+            if system is None or system.deleted_at is not None:
+                continue
+            if not await access.can_read_project(user_id, system.project_id):
+                continue
+            wanted.append(sid)
+
+        existing = set(
+            (
+                await session.execute(
+                    select(RecordSystem.system_id).where(RecordSystem.note_id == note_id)
+                )
+            ).scalars().all()
+        )
+        wanted_set = set(wanted)
+
+        to_remove = existing - wanted_set
+        if to_remove:
+            await session.execute(
+                delete(RecordSystem).where(
+                    RecordSystem.note_id == note_id,
+                    RecordSystem.system_id.in_(to_remove),
+                )
+            )
+        for sid in wanted:
+            if sid not in existing:
+                session.add(RecordSystem(note_id=note_id, system_id=sid))
+        await session.commit()
+        return wanted
+
+
+async def list_record_systems(user_id: int, note_id: int) -> list[System]:
+    """Systems associated with a record (if the user can read it)."""
+    if not await access.can_read_note(user_id, note_id):
+        return []
+    async with async_session() as session:
+        result = await session.execute(
+            select(System)
+            .join(RecordSystem, RecordSystem.system_id == System.id)
+            .where(RecordSystem.note_id == note_id, System.deleted_at.is_(None))
+            .order_by(System.order_index.asc(), System.name.asc())
+        )
+        return list(result.scalars().all())
+
+
+async def list_records_for_system(
+    user_id: int, system_id: int, kind: str | None = None, open_only: bool = False
+) -> list[Note]:
+    """Records associated with a System. `kind` filters task_kind (e.g. 'issue');
+    `open_only` limits to tasks not done/cancelled. [] if no read access."""
+    system = await get_system(user_id, system_id)
+    if system is None:
+        return []
+    async with async_session() as session:
+        query = (
+            select(Note)
+            .join(RecordSystem, RecordSystem.note_id == Note.id)
+            .where(RecordSystem.system_id == system_id, Note.deleted_at.is_(None))
+        )
+        if kind:
+            query = query.where(Note.task_kind == kind)
+        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())
+
+
+async def count_open_issues(user_id: int, project_id: int) -> int:
+    """Count open (not done/cancelled) issues in a project. 0 if no read access."""
+    if not await access.can_read_project(user_id, project_id):
+        return 0
+    async with async_session() as session:
+        result = await session.execute(
+            select(func.count(Note.id)).where(
+                Note.project_id == project_id,
+                Note.task_kind == "issue",
+                Note.status.isnot(None),
+                Note.status.not_in(["done", "cancelled"]),
+                Note.deleted_at.is_(None),
+            )
+        )
+        return int(result.scalar() or 0)

tests/test_services_systems.py

@@ -0,0 +1,70 @@
+"""ACL gating + field handling for services/systems.py (unit, mocked session)."""
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+
+def _make_mock_session():
+    s = AsyncMock()
+    s.__aenter__ = AsyncMock(return_value=s)
+    s.__aexit__ = AsyncMock(return_value=False)
+    s.add = MagicMock()
+    s.commit = AsyncMock()
+    s.refresh = AsyncMock()
+    return s
+
+
+@pytest.mark.asyncio
+async def test_create_system_denied_without_project_write():
+    with patch("scribe.services.systems.access") as acc:
+        acc.can_write_project = AsyncMock(return_value=False)
+        from scribe.services.systems import create_system
+        result = await create_system(user_id=1, project_id=5, name="Reader")
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_create_system_sets_fields_when_authorized():
+    mock_session = _make_mock_session()
+    captured = {}
+
+    def _capture_add(obj):
+        captured["name"] = getattr(obj, "name", "MISSING")
+        captured["project_id"] = getattr(obj, "project_id", "MISSING")
+
+    mock_session.add = MagicMock(side_effect=_capture_add)
+    with patch("scribe.services.systems.async_session") as mock_cls, \
+         patch("scribe.services.systems.access") as acc:
+        acc.can_write_project = AsyncMock(return_value=True)
+        mock_cls.return_value = mock_session
+        from scribe.services.systems import create_system
+        await create_system(user_id=1, project_id=5, name="  Reader  ")
+    assert captured["name"] == "Reader"  # stripped
+    assert captured["project_id"] == 5
+
+
+@pytest.mark.asyncio
+async def test_set_record_systems_denied_without_note_write():
+    with patch("scribe.services.systems.access") as acc:
+        acc.can_write_note = AsyncMock(return_value=False)
+        from scribe.services.systems import set_record_systems
+        result = await set_record_systems(user_id=1, note_id=9, system_ids=[1, 2])
+    assert result is None
+
+
+@pytest.mark.asyncio
+async def test_list_systems_denied_returns_empty():
+    with patch("scribe.services.systems.access") as acc:
+        acc.can_read_project = AsyncMock(return_value=False)
+        from scribe.services.systems import list_systems
+        result = await list_systems(user_id=1, project_id=5)
+    assert result == []
+
+
+@pytest.mark.asyncio
+async def test_count_open_issues_denied_returns_zero():
+    with patch("scribe.services.systems.access") as acc:
+        acc.can_read_project = AsyncMock(return_value=False)
+        from scribe.services.systems import count_open_issues
+        result = await count_open_issues(user_id=1, project_id=5)
+    assert result == 0