feat(ticket): 工单协同后端——升级不搬数据、受理到验收、站内通知

- item.kind='ticket' + ticket_ext 1:1，迁移 0003 同时重写 item 状态约束为任务∪工单并集
- 工单状态机与动作权限：受理/开工/提交限办理人，验收限需求人（办理人不能自验）
- 升级时原任务转 blocked，验收通过后自动回流 todo，事件流两侧都留痕
- item_link（父子/相关）与 notification（只发当事人），三张表均开 RLS
- 抽出 serializers.py 复用任务序列化，修复工单响应缺 tags 导致前端白屏
- 评论与 PATCH 放开给工单；conftest 清库前终止残留事务，避免 TRUNCATE 死等锁
- tests/test_tickets.py 11 个用例，全量 57 passed

backend/alembic/versions/0003_tickets.py

@@ -0,0 +1,181 @@
+"""M2 工单域：ticket_ext / item_link / notification + 扩宽 item.status 取值。
+
+三个需要注意的点：
+
+1. **item 的 status CHECK 必须重写**。0002 里只允许任务的六个状态，
+   工单的 `open/accepted/in_progress/pending_verify` 会直接被约束拦下。
+   这里改成任务 ∪ 工单的并集，而不是按 kind 做条件约束——PostgreSQL 的 CHECK 不能引用其它行，
+   而"任务不该有工单状态"这件事由状态机（`app/domain/*.py`）保证，不靠数据库。
+2. **新表同样开 RLS**。通知是"谁的"、工单附加信息是"哪个租户的"，漏一张就等于开了一个洞。
+3. **downgrade 会删掉所有工单数据**（kind='ticket' 的 item 及其扩展）。这是有意的：
+   旧约束容纳不下工单状态，不回删就无法回到 0002。已在函数注释里写明。
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+from sqlalchemy.dialects import postgresql
+
+revision: str = "0003"
+down_revision: str | None = "0002"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+APP_ROLE = "dadao_app"
+
+TENANT_TABLES_0003: tuple[str, ...] = ("ticket_ext", "item_link", "notification")
+
+TASK_STATUSES = ("inbox", "todo", "doing", "done", "blocked", "canceled")
+TICKET_STATUSES = ("open", "accepted", "in_progress", "pending_verify", "done", "canceled")
+
+
+def _ws_fk() -> sa.Column:
+    return sa.Column(
+        "workspace_id",
+        postgresql.UUID(as_uuid=True),
+        sa.ForeignKey("workspace.id", ondelete="CASCADE"),
+        nullable=False,
+    )
+
+
+def _user_fk(name: str, *, ondelete: str, nullable: bool = False) -> sa.Column:
+    return sa.Column(
+        name,
+        postgresql.UUID(as_uuid=True),
+        sa.ForeignKey("app_user.id", ondelete=ondelete),
+        nullable=nullable,
+    )
+
+
+def _item_fk(name: str, *, ondelete: str, nullable: bool = False) -> sa.Column:
+    return sa.Column(
+        name,
+        postgresql.UUID(as_uuid=True),
+        sa.ForeignKey("item.id", ondelete=ondelete),
+        nullable=nullable,
+    )
+
+
+def upgrade() -> None:
+    # 1) 放宽 item.status：任务 ∪ 工单
+    op.execute("ALTER TABLE item DROP CONSTRAINT IF EXISTS ck_item_status")
+    statuses = ",".join(f"'{s}'" for s in (*TASK_STATUSES, *TICKET_STATUSES))
+    op.execute(
+        f"ALTER TABLE item ADD CONSTRAINT ck_item_status CHECK (status IN ({statuses}))"
+    )
+
+    # 2) 工单扩展：item_id 既是主键也是外键，天然 1:1
+    op.create_table(
+        "ticket_ext",
+        sa.Column(
+            "item_id",
+            postgresql.UUID(as_uuid=True),
+            sa.ForeignKey("item.id", ondelete="CASCADE"),
+            primary_key=True,
+        ),
+        _ws_fk(),
+        _user_fk("requester_id", ondelete="RESTRICT"),
+        _user_fk("assignee_id", ondelete="RESTRICT"),
+        _item_fk("parent_item_id", ondelete="SET NULL", nullable=True),
+        sa.Column("accepted_at", sa.DateTime(timezone=True)),
+        sa.Column("submitted_at", sa.DateTime(timezone=True)),
+        sa.Column("verified_at", sa.DateTime(timezone=True)),
+        _user_fk("verified_by", ondelete="SET NULL", nullable=True),
+        sa.Column("reject_count", sa.Integer, nullable=False, server_default="0"),
+        sa.Column("resolution_md", sa.Text, nullable=False, server_default=""),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("version", sa.Integer, nullable=False, server_default="1"),
+        sa.CheckConstraint("reject_count >= 0", name="ck_ticket_reject_count"),
+    )
+    op.create_index("ix_ticket_ws_assignee", "ticket_ext", ["workspace_id", "assignee_id"])
+    op.create_index("ix_ticket_ws_requester", "ticket_ext", ["workspace_id", "requester_id"])
+
+    # 3) 事项关系（子任务 / 相关）
+    op.create_table(
+        "item_link",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        _ws_fk(),
+        _item_fk("from_item_id", ondelete="CASCADE"),
+        _item_fk("to_item_id", ondelete="CASCADE"),
+        sa.Column("link_type", sa.String(16), nullable=False, server_default="related"),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("version", sa.Integer, nullable=False, server_default="1"),
+        sa.UniqueConstraint("from_item_id", "to_item_id", "link_type", name="uq_item_link"),
+        sa.CheckConstraint("from_item_id <> to_item_id", name="ck_item_link_not_self"),
+        sa.CheckConstraint("link_type IN ('parent','related')", name="ck_item_link_type"),
+    )
+    op.create_index("ix_item_link_ws_from", "item_link", ["workspace_id", "from_item_id"])
+    op.create_index("ix_item_link_ws_to", "item_link", ["workspace_id", "to_item_id"])
+
+    # 4) 站内通知
+    op.create_table(
+        "notification",
+        sa.Column("id", postgresql.UUID(as_uuid=True), primary_key=True),
+        _ws_fk(),
+        _user_fk("user_id", ondelete="CASCADE"),
+        sa.Column("kind", sa.String(32), nullable=False),
+        _item_fk("item_id", ondelete="CASCADE", nullable=True),
+        sa.Column("title", sa.String(120), nullable=False),
+        sa.Column("body", sa.String(300), nullable=False, server_default=""),
+        sa.Column("read_at", sa.DateTime(timezone=True)),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False, server_default=sa.func.now()),
+        sa.Column("version", sa.Integer, nullable=False, server_default="1"),
+    )
+    op.create_index("ix_notification_ws_user_read", "notification", ["workspace_id", "user_id", "read_at"])
+    op.create_index("ix_notification_ws_created", "notification", ["workspace_id", "created_at"])
+
+    for statement in _POLICY_STATEMENTS:
+        op.execute(statement)
+    for statement in _GRANT_STATEMENTS:
+        op.execute(statement)
+
+
+def downgrade() -> None:
+    # 旧约束容纳不下工单状态，回退必须先清掉工单数据（含它们的 ticket_ext）。
+    # 这是破坏性操作，只在回滚迁移时发生；生产环境回滚前请先导出 kanban/ticket 数据。
+    op.execute("DELETE FROM item WHERE kind = 'ticket'")
+
+    for table in TENANT_TABLES_0003:
+        op.execute(f"DROP POLICY IF EXISTS {table}_isolation ON {table}")
+
+    op.drop_index("ix_notification_ws_created", table_name="notification")
+    op.drop_index("ix_notification_ws_user_read", table_name="notification")
+    op.drop_table("notification")
+
+    op.drop_index("ix_item_link_ws_to", table_name="item_link")
+    op.drop_index("ix_item_link_ws_from", table_name="item_link")
+    op.drop_table("item_link")
+
+    op.drop_index("ix_ticket_ws_requester", table_name="ticket_ext")
+    op.drop_index("ix_ticket_ws_assignee", table_name="ticket_ext")
+    op.drop_table("ticket_ext")
+
+    statuses = ",".join(f"'{s}'" for s in TASK_STATUSES)
+    op.execute("ALTER TABLE item DROP CONSTRAINT IF EXISTS ck_item_status")
+    op.execute(
+        f"ALTER TABLE item ADD CONSTRAINT ck_item_status CHECK (status IN ({statuses}))"
+    )
+
+
+_POLICY_STATEMENTS: tuple[str, ...] = tuple(
+    statement
+    for table in TENANT_TABLES_0003
+    for statement in (
+        f"ALTER TABLE {table} ENABLE ROW LEVEL SECURITY;",
+        f"""
+        CREATE POLICY {table}_isolation ON {table}
+          USING (workspace_id = app_ws_id())
+          WITH CHECK (workspace_id = app_ws_id());
+        """,
+    )
+)
+
+_GRANT_STATEMENTS: tuple[str, ...] = (
+    f"GRANT SELECT, INSERT, UPDATE, DELETE ON {', '.join(TENANT_TABLES_0003)} TO {APP_ROLE};",
+)

backend/app/api/items.py

@@ -28,7 +28,9 @@ from ..domain.task import (
 )
 from ..errors import AppError, ErrorCode
 from ..models_task import Activity, Comment, Item, ItemTag, Tag
+from ..models_ticket import TicketExt
 from ..schemas import Pagination, encode_cursor, ok
+from ..serializers import item_out, tag_out, tags_of
 
 router = APIRouter(prefix="/workspaces/{ws_id}/items", tags=["任务"])
 tag_router = APIRouter(prefix="/workspaces/{ws_id}/tags", tags=["标签"])
@@ -76,36 +78,20 @@ class TagIn(BaseModel):
     color: str = Field(default="", max_length=16)
 
 
-# ── 序列化 ──────────────────────────────────────────────────────────────
+# ── 序列化（与工单接口共用一份，两份拼装迟早会不一致）─────────────────────
-def _tag_out(tag: Tag) -> dict[str, Any]:
+_tag_out = tag_out
-    return {"id": str(tag.id), "name": tag.name, "color": tag.color}
+_item_out = item_out
-
-
-def _item_out(item: Item, tags: list[Tag] | None = None) -> dict[str, Any]:
-    return {
-        "id": str(item.id),
-        "kind": item.kind,
-        "title": item.title,
-        "body_md": item.body_md,
-        "status": item.status,
-        "priority": item.priority,
-        "creator_id": str(item.creator_id),
-        "assignee_id": str(item.assignee_id) if item.assignee_id else None,
-        "due_at": item.due_at,
-        "completed_at": item.completed_at,
-        "sort_order": item.sort_order,
-        "deleted_at": item.deleted_at,
-        "tags": [_tag_out(t) for t in (tags or [])],
-        "created_at": item.created_at,
-        "updated_at": item.updated_at,
-        "version": item.version,
-    }
 
 
 # ── 内部工具 ────────────────────────────────────────────────────────────
-async def _load_item(session: AsyncSession, ws_id: uuid.UUID, item_id: uuid.UUID) -> Item:
+async def _load_item(
+    session: AsyncSession,
+    ws_id: uuid.UUID,
+    item_id: uuid.UUID,
+    kinds: tuple[str, ...] = ("task",),
+) -> Item:
     item = await session.get(Item, item_id)
-    if item is None or item.workspace_id != ws_id or item.kind != "task":
+    if item is None or item.workspace_id != ws_id or item.kind not in kinds:
         # 跨租户 / 不存在 / 不是任务，对外一律"不存在"，不泄漏别人有没有这条数据
         raise AppError(ErrorCode.ITEM_NOT_FOUND, "任务不存在")
     return item
@@ -114,20 +100,7 @@ async def _load_item(session: AsyncSession, ws_id: uuid.UUID, item_id: uuid.UUID
 async def _tags_of(
     session: AsyncSession, ws_id: uuid.UUID, item_ids: list[uuid.UUID]
 ) -> dict[uuid.UUID, list[Tag]]:
-    if not item_ids:
+    return await tags_of(session, ws_id, item_ids)
-        return {}
-    rows = (
-        await session.execute(
-            select(ItemTag.item_id, Tag)
-            .join(Tag, Tag.id == ItemTag.tag_id)
-            .where(ItemTag.workspace_id == ws_id, ItemTag.item_id.in_(item_ids))
-            .order_by(Tag.name)
-        )
-    ).all()
-    grouped: dict[uuid.UUID, list[Tag]] = {}
-    for item_id, tag in rows:
-        grouped.setdefault(item_id, []).append(tag)
-    return grouped
 
 
 def _normalize_tag_names(names: list[str]) -> list[str]:
@@ -415,9 +388,9 @@ async def board(
     )
 
 
-@router.get("/{item_id}", summary="任务详情（含最近 50 条事件）")
+@router.get("/{item_id}", summary="事项详情（任务或工单，含最近 50 条事件）")
 async def get_item(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
-    item = await _load_item(session, ctx.ws_id, item_id)
+    item = await _load_item(session, ctx.ws_id, item_id, ("task", "ticket"))
     tags = await _tags_of(session, ctx.ws_id, [item.id])
     activities = (
         (
@@ -442,6 +415,20 @@ async def get_item(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> d
         }
         for a in activities
     ]
+    if item.kind == "ticket":
+        ext = await session.get(TicketExt, item.id)
+        if ext is not None:
+            data["ticket"] = {
+                "requester_id": str(ext.requester_id),
+                "assignee_id": str(ext.assignee_id),
+                "parent_item_id": str(ext.parent_item_id) if ext.parent_item_id else None,
+                "accepted_at": ext.accepted_at,
+                "submitted_at": ext.submitted_at,
+                "verified_at": ext.verified_at,
+                "verified_by": str(ext.verified_by) if ext.verified_by else None,
+                "reject_count": ext.reject_count,
+                "resolution_md": ext.resolution_md,
+            }
     return ok(data)
 
 
@@ -556,11 +543,12 @@ async def batch_status(body: BatchStatusIn, ctx: MemberDep, session: SessionDep)
     )
 
 
-@router.patch("/{item_id}", summary="修改任务字段（乐观锁）")
+@router.patch("/{item_id}", summary="修改事项字段（乐观锁）")
 async def update_item(
     item_id: uuid.UUID, body: UpdateItemIn, ctx: MemberDep, session: SessionDep
 ) -> dict:
-    item = await _load_item(session, ctx.ws_id, item_id)
+    # 工单也能改标题/优先级/截止/标签；状态必须走 /ticket/* 的动作接口
+    item = await _load_item(session, ctx.ws_id, item_id, ("task", "ticket"))
 
     values: dict[str, Any] = {}
     changes: dict[str, Any] = {}
@@ -662,7 +650,7 @@ async def delete_item(
 async def add_comment(
     item_id: uuid.UUID, body: CommentIn, ctx: MemberDep, session: SessionDep
 ) -> dict:
-    item = await _load_item(session, ctx.ws_id, item_id)
+    item = await _load_item(session, ctx.ws_id, item_id, ("task", "ticket"))
     comment = Comment(
         workspace_id=ctx.ws_id,
         item_id=item.id,
@@ -686,7 +674,7 @@ async def add_comment(
 
 @router.get("/{item_id}/comments", summary="评论列表")
 async def list_comments(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
-    item = await _load_item(session, ctx.ws_id, item_id)
+    item = await _load_item(session, ctx.ws_id, item_id, ("task", "ticket"))
     rows = (
         (
             await session.execute(

backend/app/api/tickets.py

@@ -0,0 +1,763 @@
+"""工单域接口（M2）：升级、指派、办理、验收、回流、通知、事项关系。
+
+一段流程长什么样：
+
+    任务卡住了
+      -> POST /items/{task_id}/promote {assignee_id}   任务变 blocked，生成一条工单（open）
+      -> 被指派人 POST /items/{ticket_id}/ticket/accept   accepted
+      -> POST /items/{ticket_id}/ticket/start             in_progress
+      -> POST /items/{ticket_id}/ticket/submit            pending_verify（提需求的人收到通知）
+      -> POST /items/{ticket_id}/ticket/verify {approve}  done，原任务自动 blocked -> todo
+
+驳回走同一个入口：verify {approve: false, note: "..."} -> 回到 in_progress，reject_count + 1。
+每一步都写 activity；通知只是提醒，状态永远以 item 为准。
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime, timezone
+from typing import Annotated, Any
+
+from fastapi import APIRouter, Query
+from pydantic import BaseModel, Field
+from sqlalchemy import Select, and_, or_, select, update
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from ..audit import record as audit_record
+from ..deps import MemberDep, SessionDep, TenantCtx
+from ..domain.ticket import (
+    CANCELABLE,
+    NOW_FIELD,
+    ensure_can_act,
+    returned_task_status,
+    target_status,
+)
+from ..errors import AppError, ErrorCode
+from ..models import Member
+from ..models_task import Activity, Item
+from ..models_ticket import ItemLink, Notification, TicketExt
+from ..schemas import Pagination, encode_cursor, ok
+from ..serializers import item_out, tags_of
+
+router = APIRouter(prefix="/workspaces/{ws_id}", tags=["工单"])
+notify_router = APIRouter(prefix="/workspaces/{ws_id}/notifications", tags=["通知"])
+
+MAX_REJECT_NOTE = 500
+
+
+# -- 入参 ---------------------------------------------------------------
+class PromoteIn(BaseModel):
+    assignee_id: uuid.UUID
+    title: str | None = Field(default=None, max_length=200)
+    body_md: str | None = Field(default=None, max_length=20000)
+    due_at: datetime | None = None
+    priority: int | None = Field(default=None, ge=0, le=3)
+    version: int | None = Field(default=None, description="原任务的版本，给了就做乐观锁")
+
+
+class ReassignIn(BaseModel):
+    assignee_id: uuid.UUID
+
+
+class VerifyIn(BaseModel):
+    approve: bool
+    note: str = Field(default="", max_length=MAX_REJECT_NOTE)
+    resolution_md: str = Field(
+        default="", max_length=20000, description="验收通过时写的结论，M3 用它长知识"
+    )
+
+
+class ActionIn(BaseModel):
+    note: str = Field(default="", max_length=MAX_REJECT_NOTE)
+    version: int | None = None
+
+
+class LinkIn(BaseModel):
+    to_item_id: uuid.UUID
+    link_type: str = "related"
+
+
+# -- 内部工具 -----------------------------------------------------------
+async def _load_item(
+    session: AsyncSession, ws_id: uuid.UUID, item_id: uuid.UUID, kinds: tuple[str, ...]
+) -> Item:
+    item = await session.get(Item, item_id)
+    if item is None or item.workspace_id != ws_id or item.kind not in kinds:
+        # 跨租户 / 不存在 / 类型不符：对外一律"不存在"
+        raise AppError(ErrorCode.ITEM_NOT_FOUND, "事项不存在")
+    if item.deleted_at is not None:
+        raise AppError(ErrorCode.ITEM_NOT_FOUND, "事项不存在")
+    return item
+
+
+async def _load_ext(session: AsyncSession, item_id: uuid.UUID) -> TicketExt:
+    ext = await session.get(TicketExt, item_id)
+    if ext is None:
+        raise AppError(ErrorCode.TICKET_NOT_FOUND, "这不是一条工单")
+    return ext
+
+
+async def _active_member_id(session: AsyncSession, ws_id: uuid.UUID, user_id: uuid.UUID) -> None:
+    exists = (
+        await session.execute(
+            select(Member.id).where(
+                Member.workspace_id == ws_id,
+                Member.user_id == user_id,
+                Member.status == "active",
+            )
+        )
+    ).first()
+    if exists is None:
+        raise AppError(ErrorCode.BAD_REQUEST, "被指派的人不在这个工作区里")
+
+
+async def _log(
+    session: AsyncSession,
+    item: Item,
+    actor_id: uuid.UUID | None,
+    action: str,
+    payload: dict | None = None,
+) -> None:
+    session.add(
+        Activity(
+            workspace_id=item.workspace_id,
+            item_id=item.id,
+            actor_id=actor_id,
+            action=action,
+            payload=payload or {},
+        )
+    )
+
+
+async def _notify(
+    session: AsyncSession,
+    *,
+    ws_id: uuid.UUID,
+    user_id: uuid.UUID,
+    kind: str,
+    item_id: uuid.UUID | None,
+    title: str,
+    body: str = "",
+) -> None:
+    session.add(
+        Notification(
+            workspace_id=ws_id,
+            user_id=user_id,
+            kind=kind,
+            item_id=item_id,
+            title=title[:120],
+            body=body[:300],
+        )
+    )
+
+
+def _ticket_ext_out(ext: TicketExt) -> dict[str, Any]:
+    return {
+        "requester_id": str(ext.requester_id),
+        "assignee_id": str(ext.assignee_id),
+        "parent_item_id": str(ext.parent_item_id) if ext.parent_item_id else None,
+        "accepted_at": ext.accepted_at,
+        "submitted_at": ext.submitted_at,
+        "verified_at": ext.verified_at,
+        "verified_by": str(ext.verified_by) if ext.verified_by else None,
+        "reject_count": ext.reject_count,
+        "resolution_md": ext.resolution_md,
+    }
+
+
+async def _item_out(session: AsyncSession, ws_id: uuid.UUID, item: Item) -> dict[str, Any]:
+    """单条序列化。标签必须一起带上——前端把 `item.tags` 当数组用，
+    少这个字段整页会白屏（M2 首版就是这么炸的）。"""
+    grouped = await tags_of(session, ws_id, [item.id])
+    return item_out(item, grouped.get(item.id))
+
+
+async def _items_out(
+    session: AsyncSession, ws_id: uuid.UUID, items: list[Item]
+) -> list[dict[str, Any]]:
+    grouped = await tags_of(session, ws_id, [i.id for i in items])
+    return [item_out(i, grouped.get(i.id)) for i in items]
+
+
+async def _ticket_out(
+    session: AsyncSession, ws_id: uuid.UUID, ticket: Item, ext: TicketExt
+) -> dict:
+    return ok({**await _item_out(session, ws_id, ticket), "ticket": _ticket_ext_out(ext)})
+
+
+async def _patch_item(
+    session: AsyncSession,
+    item: Item,
+    values: dict[str, Any],
+    *,
+    expect_version: int | None,
+) -> None:
+    """带版本的 UPDATE：给了 version 就必须匹配，否则 409 让客户端刷新。"""
+    where = [Item.id == item.id, Item.workspace_id == item.workspace_id]
+    if expect_version is not None:
+        where.append(Item.version == expect_version)
+    result = await session.execute(
+        update(Item)
+        .where(*where)
+        .values(**values, version=Item.version + 1, updated_at=datetime.now(timezone.utc))
+    )
+    if result.rowcount == 0:
+        await session.refresh(item)
+        raise AppError(
+            ErrorCode.VERSION_CONFLICT,
+            "这条工单刚被改过，请刷新后重试",
+            details={"current_version": item.version},
+        )
+    await session.refresh(item)
+
+
+# -- 升级 ---------------------------------------------------------------
+@router.post("/items/{item_id}/promote", summary="把任务升级成工单")
+async def promote(item_id: uuid.UUID, body: PromoteIn, ctx: MemberDep, session: SessionDep) -> dict:
+    task = await _load_item(session, ctx.ws_id, item_id, ("task",))
+    if task.status in ("done", "canceled"):
+        raise AppError(
+            ErrorCode.ITEM_STATUS_INVALID,
+            f"「{task.status}」的任务不能再升级成工单",
+            details={"status": task.status},
+        )
+
+    existing = (
+        await session.execute(
+            select(TicketExt.item_id).where(
+                TicketExt.workspace_id == ctx.ws_id, TicketExt.parent_item_id == task.id
+            )
+        )
+    ).first()
+    if existing is not None:
+        raise AppError(ErrorCode.TICKET_ALREADY_EXISTS, "这条任务已经升级过工单了")
+
+    await _active_member_id(session, ctx.ws_id, body.assignee_id)
+
+    ticket = Item(
+        workspace_id=ctx.ws_id,
+        kind="ticket",
+        title=(body.title or task.title)[:200],
+        body_md=body.body_md if body.body_md is not None else task.body_md,
+        status="open",
+        priority=body.priority if body.priority is not None else task.priority,
+        creator_id=ctx.user.id,
+        assignee_id=body.assignee_id,
+        due_at=body.due_at or task.due_at,
+    )
+    session.add(ticket)
+    await session.flush()
+
+    ext = TicketExt(
+        item_id=ticket.id,
+        workspace_id=ctx.ws_id,
+        requester_id=ctx.user.id,
+        assignee_id=body.assignee_id,
+        parent_item_id=task.id,
+    )
+    session.add(ext)
+    await session.flush()
+
+    # 原任务确实卡住了，进 blocked；不能静默把用户的待办改成别的状态
+    await _patch_item(session, task, {"status": "blocked"}, expect_version=body.version)
+    await _log(session, task, ctx.user.id, "promoted_to_ticket", {"ticket_id": str(ticket.id)})
+    await _log(session, ticket, ctx.user.id, "ticket_created", {"parent_item_id": str(task.id)})
+    await _notify(
+        session,
+        ws_id=ctx.ws_id,
+        user_id=body.assignee_id,
+        kind="ticket_assigned",
+        item_id=ticket.id,
+        title=f"你被指派了一条工单：{ticket.title}",
+        body=f"{ctx.user.nickname or '同事'} 请你处理，完成后提交验收",
+    )
+    await audit_record(
+        session,
+        workspace_id=ctx.ws_id,
+        actor_user_id=ctx.user.id,
+        action="ticket.promote",
+        target_type="item",
+        target_id=str(ticket.id),
+        payload={"parent_item_id": str(task.id)},
+    )
+    await session.flush()
+
+    task_out, ticket_out = await _items_out(session, ctx.ws_id, [task, ticket])
+    return ok({"task": task_out, "ticket": {**ticket_out, "ticket": _ticket_ext_out(ext)}})
+
+
+# -- 办理/验收 ----------------------------------------------------------
+async def _advance(
+    ctx: TenantCtx,
+    session: AsyncSession,
+    item_id: uuid.UUID,
+    *,
+    action: str,
+    note: str = "",
+    approve: bool | None = None,
+    resolution_md: str = "",
+    version: int | None = None,
+) -> dict:
+    ticket = await _load_item(session, ctx.ws_id, item_id, ("ticket",))
+    ext = await _load_ext(session, ticket.id)
+
+    target = target_status(action, ticket.status)
+    ensure_can_act(
+        action=action,
+        actor_id=ctx.user.id,
+        role=ctx.role,
+        assignee_id=ext.assignee_id,
+        requester_id=ext.requester_id,
+    )
+
+    now = datetime.now(timezone.utc)
+    previous = ticket.status
+    values: dict[str, Any] = {"status": target}
+    if target == "done":
+        values["completed_at"] = now
+        ext.verified_at = now
+        ext.verified_by = ctx.user.id
+        if resolution_md:
+            ext.resolution_md = resolution_md
+    if target == "in_progress" and action == "reject":
+        ext.reject_count = ext.reject_count + 1
+
+    stamp_field = NOW_FIELD.get(target)
+    if stamp_field and action != "reject":
+        setattr(ext, stamp_field, now)
+
+    await _patch_item(session, ticket, values, expect_version=version)
+
+    payload: dict[str, Any] = {"action": action, "from": previous, "to": target}
+    if note:
+        payload["note"] = note
+    await _log(session, ticket, ctx.user.id, f"ticket_{action}", payload)
+
+    # 通知对家：验收找提需求的人，其余找办理人
+    if action == "submit":
+        await _notify(
+            session,
+            ws_id=ctx.ws_id,
+            user_id=ext.requester_id,
+            kind="ticket_submitted",
+            item_id=ticket.id,
+            title=f"工单待你验收：{ticket.title}",
+            body=note or "办理人已提交，请确认结果",
+        )
+    elif action == "approve":
+        await _notify(
+            session,
+            ws_id=ctx.ws_id,
+            user_id=ext.assignee_id,
+            kind="ticket_approved",
+            item_id=ticket.id,
+            title=f"工单已通过验收：{ticket.title}",
+            body=note or "谢谢，这事结了",
+        )
+    elif action == "reject":
+        await _notify(
+            session,
+            ws_id=ctx.ws_id,
+            user_id=ext.assignee_id,
+            kind="ticket_rejected",
+            item_id=ticket.id,
+            title=f"工单被退回：{ticket.title}",
+            body=note or "请按反馈再改一版",
+        )
+
+    # 验收通过 -> 原任务闭环回流
+    returned = returned_task_status(target)
+    if returned and ext.parent_item_id:
+        parent = await session.get(Item, ext.parent_item_id)
+        if parent is not None and parent.workspace_id == ctx.ws_id and parent.status == "blocked":
+            parent.status = returned
+            parent.version += 1
+            parent.updated_at = now
+            await _log(
+                session,
+                parent,
+                ctx.user.id,
+                "ticket_done",
+                {"ticket_id": str(ticket.id), "status": returned},
+            )
+            await _notify(
+                session,
+                ws_id=ctx.ws_id,
+                user_id=parent.creator_id,
+                kind="task_unblocked",
+                item_id=parent.id,
+                title=f"任务可以继续了：{parent.title}",
+                body="它对应的工单已验收通过",
+            )
+
+    await session.flush()
+    return await _ticket_out(session, ctx.ws_id, ticket, ext)
+
+
+@router.post("/items/{item_id}/ticket/accept", summary="接受工单")
+async def accept(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
+    return await _advance(ctx, session, item_id, action="accept")
+
+
+@router.post("/items/{item_id}/ticket/start", summary="开始办理")
+async def start(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
+    return await _advance(ctx, session, item_id, action="start")
+
+
+@router.post("/items/{item_id}/ticket/submit", summary="提交验收")
+async def submit(item_id: uuid.UUID, body: ActionIn, ctx: MemberDep, session: SessionDep) -> dict:
+    return await _advance(
+        ctx, session, item_id, action="submit", note=body.note, version=body.version
+    )
+
+
+@router.post("/items/{item_id}/ticket/verify", summary="验收：通过或驳回")
+async def verify(item_id: uuid.UUID, body: VerifyIn, ctx: MemberDep, session: SessionDep) -> dict:
+    return await _advance(
+        ctx,
+        session,
+        item_id,
+        action="approve" if body.approve else "reject",
+        note=body.note,
+        approve=body.approve,
+        resolution_md=body.resolution_md,
+    )
+
+
+@router.post("/items/{item_id}/ticket/cancel", summary="取消工单")
+async def cancel(item_id: uuid.UUID, body: ActionIn, ctx: MemberDep, session: SessionDep) -> dict:
+    return await _advance(
+        ctx, session, item_id, action="cancel", note=body.note, version=body.version
+    )
+
+
+@router.post("/items/{item_id}/ticket/reassign", summary="改派给同事")
+async def reassign(
+    item_id: uuid.UUID, body: ReassignIn, ctx: MemberDep, session: SessionDep
+) -> dict:
+    ticket = await _load_item(session, ctx.ws_id, item_id, ("ticket",))
+    ext = await _load_ext(session, ticket.id)
+
+    ensure_can_act(
+        action="reassign",
+        actor_id=ctx.user.id,
+        role=ctx.role,
+        assignee_id=ext.assignee_id,
+        requester_id=ext.requester_id,
+    )
+    if ticket.status not in CANCELABLE:
+        raise AppError(
+            ErrorCode.TICKET_STATUS_INVALID,
+            f"「{ticket.status}」的工单不能改派",
+            details={"allowed_from": list(CANCELABLE)},
+        )
+    if body.assignee_id == ext.assignee_id:
+        return await _ticket_out(session, ctx.ws_id, ticket, ext)
+
+    await _active_member_id(session, ctx.ws_id, body.assignee_id)
+
+    await _patch_item(
+        session,
+        ticket,
+        {"status": "open", "assignee_id": body.assignee_id, "completed_at": None},
+        expect_version=None,
+    )
+    ext.assignee_id = body.assignee_id
+    ext.accepted_at = None
+    ext.submitted_at = None
+    await _log(session, ticket, ctx.user.id, "ticket_reassigned", {"to": str(body.assignee_id)})
+    await _notify(
+        session,
+        ws_id=ctx.ws_id,
+        user_id=body.assignee_id,
+        kind="ticket_assigned",
+        item_id=ticket.id,
+        title=f"转给你一条工单：{ticket.title}",
+        body="请先接收，再开始办理",
+    )
+    await session.flush()
+    return await _ticket_out(session, ctx.ws_id, ticket, ext)
+
+
+# -- 工单列表 -----------------------------------------------------------
+@router.get("/tickets", summary="工单列表：我指派的 / 我发起的")
+async def list_tickets(
+    ctx: MemberDep,
+    session: SessionDep,
+    role: Annotated[str, Query(description="assigned|requested|all")] = "assigned",
+    status: Annotated[list[str] | None, Query(description="按状态过滤，可重复")] = None,
+    limit: Annotated[int, Query(ge=1, le=100)] = 20,
+    cursor: Annotated[str | None, Query()] = None,
+) -> dict:
+    page = Pagination(limit=limit, cursor=cursor)
+    limit, parsed_cursor = page.parsed()
+
+    stmt: Select = (
+        select(Item, TicketExt)
+        .join(TicketExt, TicketExt.item_id == Item.id)
+        .where(Item.workspace_id == ctx.ws_id, Item.kind == "ticket", Item.deleted_at.is_(None))
+    )
+    if role == "assigned":
+        stmt = stmt.where(TicketExt.assignee_id == ctx.user.id)
+    elif role == "requested":
+        stmt = stmt.where(TicketExt.requester_id == ctx.user.id)
+    elif role != "all":
+        raise AppError(ErrorCode.BAD_REQUEST, "role 只能是 assigned / requested / all")
+
+    if status:
+        stmt = stmt.where(Item.status.in_(status))
+
+    stmt = stmt.order_by(Item.updated_at.desc(), Item.id.desc()).limit(limit + 1)
+    if parsed_cursor is not None:
+        stmt = stmt.where(
+            or_(
+                Item.updated_at < parsed_cursor.created_at,
+                and_(Item.updated_at == parsed_cursor.created_at, Item.id < parsed_cursor.id),
+            )
+        )
+
+    rows = (await session.execute(stmt)).all()
+    has_more = len(rows) > limit
+    rows = rows[:limit]
+    next_cursor = (
+        encode_cursor(rows[-1][0].updated_at, rows[-1][0].id) if has_more and rows else None
+    )
+    tickets = await _items_out(session, ctx.ws_id, [item for item, _ in rows])
+    return ok(
+        {
+            "tickets": [
+                {**out, "ticket": _ticket_ext_out(ext)}
+                for out, (_, ext) in zip(tickets, rows, strict=True)
+            ],
+            "next_cursor": next_cursor,
+            "has_more": has_more,
+        }
+    )
+
+
+# -- 事项关系（子任务 / 相关）------------------------------------------
+@router.get("/items/{item_id}/links", summary="这条事项的子项与相关项")
+async def list_links(item_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
+    await _load_item(session, ctx.ws_id, item_id, ("task", "ticket"))
+
+    # parent 边的方向：from 是 to 的子项。所以"我的子项"= 指向我、且类型为 parent 的 from。
+    children = (
+        await session.execute(
+            select(ItemLink, Item)
+            .join(Item, Item.id == ItemLink.from_item_id)
+            .where(
+                ItemLink.workspace_id == ctx.ws_id,
+                ItemLink.to_item_id == item_id,
+                ItemLink.link_type == "parent",
+            )
+        )
+    ).all()
+    parents = (
+        await session.execute(
+            select(ItemLink, Item)
+            .join(Item, Item.id == ItemLink.to_item_id)
+            .where(
+                ItemLink.workspace_id == ctx.ws_id,
+                ItemLink.from_item_id == item_id,
+                ItemLink.link_type == "parent",
+            )
+        )
+    ).all()
+    related = (
+        await session.execute(
+            select(ItemLink, Item)
+            .join(
+                Item,
+                or_(
+                    and_(ItemLink.from_item_id == item_id, Item.id == ItemLink.to_item_id),
+                    and_(ItemLink.to_item_id == item_id, Item.id == ItemLink.from_item_id),
+                ),
+            )
+            .where(
+                ItemLink.workspace_id == ctx.ws_id,
+                ItemLink.link_type == "related",
+                or_(ItemLink.from_item_id == item_id, ItemLink.to_item_id == item_id),
+            )
+        )
+    ).all()
+
+    async def _pack(rows: list[Any]) -> list[dict[str, Any]]:
+        packed = await _items_out(session, ctx.ws_id, [item for _, item in rows])
+        return [
+            {"link_id": str(link.id), "link_type": link.link_type, "item": out}
+            for (link, _), out in zip(rows, packed, strict=True)
+        ]
+
+    return ok(
+        {
+            "children": await _pack(children),
+            "parents": await _pack(parents),
+            "related": await _pack(related),
+        }
+    )
+
+
+@router.post("/items/{item_id}/links", summary="把另一条事项挂成子项/相关项")
+async def create_link(
+    item_id: uuid.UUID, body: LinkIn, ctx: MemberDep, session: SessionDep
+) -> dict:
+    if body.link_type not in ("parent", "related"):
+        raise AppError(ErrorCode.BAD_REQUEST, "link_type 只能是 parent / related")
+    if body.to_item_id == item_id:
+        raise AppError(ErrorCode.LINK_CONFLICT, "不能把自己挂到自己下面")
+
+    await _load_item(session, ctx.ws_id, body.to_item_id, ("task", "ticket"))
+    await _load_item(session, ctx.ws_id, item_id, ("task", "ticket"))
+
+    existing = (
+        await session.execute(
+            select(ItemLink).where(
+                ItemLink.from_item_id == body.to_item_id,
+                ItemLink.to_item_id == item_id,
+                ItemLink.link_type == body.link_type,
+            )
+        )
+    ).scalar_one_or_none()
+    if existing is not None:
+        raise AppError(ErrorCode.LINK_CONFLICT, "这条关系已经存在")
+
+    link = ItemLink(
+        workspace_id=ctx.ws_id,
+        from_item_id=body.to_item_id,
+        to_item_id=item_id,
+        link_type=body.link_type,
+    )
+    session.add(link)
+    await session.flush()
+    return ok({"id": str(link.id), "link_type": link.link_type})
+
+
+@router.delete("/items/{item_id}/links/{link_id}", summary="解除关系")
+async def delete_link(
+    item_id: uuid.UUID, link_id: uuid.UUID, ctx: MemberDep, session: SessionDep
+) -> dict:
+    link = await session.get(ItemLink, link_id)
+    if (
+        link is None
+        or link.workspace_id != ctx.ws_id
+        or item_id
+        not in (
+            link.from_item_id,
+            link.to_item_id,
+        )
+    ):
+        raise AppError(ErrorCode.LINK_NOT_FOUND, "关系不存在")
+    await session.delete(link)
+    return ok({"deleted": True})
+
+
+# -- 通知 ---------------------------------------------------------------
+def _notification_out(n: Notification, item_title: str | None = None) -> dict[str, Any]:
+    return {
+        "id": str(n.id),
+        "kind": n.kind,
+        "item_id": str(n.item_id) if n.item_id else None,
+        "item_title": item_title,
+        "title": n.title,
+        "body": n.body,
+        "read_at": n.read_at,
+        "created_at": n.created_at,
+    }
+
+
+@notify_router.get("", summary="我的通知（默认只看未读）")
+async def list_notifications(
+    ctx: MemberDep,
+    session: SessionDep,
+    unread_only: bool = True,
+    limit: Annotated[int, Query(ge=1, le=100)] = 30,
+    cursor: Annotated[str | None, Query()] = None,
+) -> dict:
+    page = Pagination(limit=limit, cursor=cursor)
+    limit, parsed_cursor = page.parsed()
+
+    stmt = select(Notification).where(
+        Notification.workspace_id == ctx.ws_id, Notification.user_id == ctx.user.id
+    )
+    if unread_only:
+        stmt = stmt.where(Notification.read_at.is_(None))
+    stmt = stmt.order_by(Notification.created_at.desc(), Notification.id.desc()).limit(limit + 1)
+    if parsed_cursor is not None:
+        stmt = stmt.where(
+            or_(
+                Notification.created_at < parsed_cursor.created_at,
+                and_(
+                    Notification.created_at == parsed_cursor.created_at,
+                    Notification.id < parsed_cursor.id,
+                ),
+            )
+        )
+
+    rows = (await session.execute(stmt)).scalars().all()
+    has_more = len(rows) > limit
+    rows = list(rows[:limit])
+
+    titles: dict[uuid.UUID, str] = {}
+    item_ids = [n.item_id for n in rows if n.item_id]
+    if item_ids:
+        for item_id, title in (
+            await session.execute(select(Item.id, Item.title).where(Item.id.in_(item_ids)))
+        ).all():
+            titles[item_id] = title
+
+    next_cursor = encode_cursor(rows[-1].created_at, rows[-1].id) if has_more and rows else None
+    return ok(
+        {
+            "notifications": [
+                _notification_out(n, titles.get(n.item_id) if n.item_id else None) for n in rows
+            ],
+            "next_cursor": next_cursor,
+            "has_more": has_more,
+        }
+    )
+
+
+@notify_router.get("/unread-count", summary="未读数（导航角标用）")
+async def unread_count(ctx: MemberDep, session: SessionDep) -> dict:
+    from sqlalchemy import func as sa_func
+
+    total = (
+        await session.execute(
+            select(sa_func.count(Notification.id)).where(
+                Notification.workspace_id == ctx.ws_id,
+                Notification.user_id == ctx.user.id,
+                Notification.read_at.is_(None),
+            )
+        )
+    ).scalar_one()
+    return ok({"unread": int(total)})
+
+
+@notify_router.post("/{notification_id}/read", summary="标记已读")
+async def mark_read(notification_id: uuid.UUID, ctx: MemberDep, session: SessionDep) -> dict:
+    n = await session.get(Notification, notification_id)
+    if n is None or n.workspace_id != ctx.ws_id or n.user_id != ctx.user.id:
+        raise AppError(ErrorCode.NOTIFICATION_NOT_FOUND, "通知不存在")
+    if n.read_at is None:
+        n.read_at = datetime.now(timezone.utc)
+    return ok({"read_at": n.read_at})
+
+
+@notify_router.post("/read-all", summary="全部标记已读")
+async def mark_all_read(ctx: MemberDep, session: SessionDep) -> dict:
+    from sqlalchemy import update as sa_update
+
+    result = await session.execute(
+        sa_update(Notification)
+        .where(
+            Notification.workspace_id == ctx.ws_id,
+            Notification.user_id == ctx.user.id,
+            Notification.read_at.is_(None),
+        )
+        .values(read_at=datetime.now(timezone.utc))
+    )
+    return ok({"marked": result.rowcount})
+
+
+__all__ = ["notify_router", "router"]

backend/app/domain/ticket.py

@@ -0,0 +1,138 @@
+"""工单状态机与权限判定（纯函数，不碰 ORM 以外的世界）。
+
+状态机：
+
+    open ──accept──▶ accepted ──start──▶ in_progress ──submit──▶ pending_verify
+                                          ▲                          │
+                                          └────────reject────────────┘
+                                                                     │approve
+                                                                     ▼
+                                                                   done
+
+`canceled` 可以从任何未终结状态进入；`done` 与 `canceled` 是终态。
+
+两条刻意的规则：
+
+1. **驳回不是打回 open，而是回 in_progress**。办理人已经在做了，让他重新"认领"一次纯属折腾；
+   驳回次数单独计数（`reject_count`），用来提示"这事返工过"。
+2. **谁也不能替别人验收**，除了管理员/所有者。提需求的人验收是这套流程的核心信任点，
+   如果办理人能自己点"通过"，工单就退化成"打个勾"。
+"""
+
+from __future__ import annotations
+
+import uuid
+
+from ..errors import AppError, ErrorCode
+
+TICKET_TRANSITIONS: dict[str, tuple[str, ...]] = {
+    "open": ("accepted", "canceled"),
+    "accepted": ("in_progress", "canceled"),
+    "in_progress": ("pending_verify", "canceled"),
+    "pending_verify": ("done", "in_progress", "canceled"),
+    "done": (),
+    "canceled": (),
+}
+
+# 动作 → (允许的前置状态, 目标状态)；cancel 特殊处理
+ACTION_RULES: dict[str, tuple[tuple[str, ...], str]] = {
+    "accept": (("open",), "accepted"),
+    "start": (("accepted",), "in_progress"),
+    "submit": (("in_progress",), "pending_verify"),
+    "approve": (("pending_verify",), "done"),
+    "reject": (("pending_verify",), "in_progress"),
+}
+
+CANCELABLE = ("open", "accepted", "in_progress", "pending_verify")
+
+NOW_FIELD: dict[str, str] = {
+    "accepted": "accepted_at",
+    "pending_verify": "submitted_at",
+    "done": "verified_at",
+}
+
+
+def ensure_transition(current: str, target: str) -> None:
+    allowed = TICKET_TRANSITIONS.get(current, ())
+    if target not in allowed:
+        raise AppError(
+            ErrorCode.TICKET_STATUS_INVALID,
+            f"工单当前是「{current}」，不能直接变成「{target}」",
+            details={"from": current, "to": target, "allowed": list(allowed)},
+        )
+
+
+def target_status(action: str, current: str) -> str:
+    """动作 → 目标状态，顺带校验前置状态。"""
+    if action == "cancel":
+        if current not in CANCELABLE:
+            raise AppError(
+                ErrorCode.TICKET_STATUS_INVALID,
+                f"「{current}」的工单不能取消",
+                details={"from": current, "allowed": list(CANCELABLE)},
+            )
+        return "canceled"
+
+    if action not in ACTION_RULES:
+        raise AppError(ErrorCode.BAD_REQUEST, f"未知动作：{action}")
+
+    allowed_from, target = ACTION_RULES[action]
+    if current not in allowed_from:
+        raise AppError(
+            ErrorCode.TICKET_STATUS_INVALID,
+            f"工单当前是「{current}」，不能执行「{action}」",
+            # allowed 回答"现在还能去哪儿"，allowed_from 回答"这个动作要什么前置状态"：
+            # 客户端两种都要用得上，别让它自己拼。
+            details={
+                "from": current,
+                "action": action,
+                "allowed": list(TICKET_TRANSITIONS.get(current, ())),
+                "allowed_from": list(allowed_from),
+            },
+        )
+    return target
+
+
+def ensure_can_act(
+    *,
+    action: str,
+    actor_id: uuid.UUID,
+    role: str,
+    assignee_id: uuid.UUID | None,
+    requester_id: uuid.UUID | None,
+) -> None:
+    """谁能做这个动作。
+
+    管理员/所有者可以代任何人操作（人休假、离职时的兜底），
+    普通成员只能动自己的位置上的事。
+    """
+    is_admin = role in ("owner", "admin")
+    is_assignee = assignee_id is not None and actor_id == assignee_id
+    is_requester = requester_id is not None and actor_id == requester_id
+
+    if action in ("accept", "start", "submit"):
+        allowed = is_assignee or is_admin
+        hint = "只有被指派的人才能操作这条工单"
+    elif action in ("approve", "reject"):
+        allowed = is_requester or is_admin
+        hint = "只有提出这条工单的人（或管理员）才能验收"
+    elif action == "cancel":
+        allowed = is_requester or is_assignee or is_admin
+        hint = "只有提出人、办理人或管理员能取消"
+    else:
+        allowed = is_admin or is_requester
+        hint = "权限不足"
+
+    if not allowed:
+        raise AppError(ErrorCode.TICKET_FORBIDDEN, hint, details={"action": action})
+
+
+def returned_task_status(ticket_status: str) -> str | None:
+    """工单状态变化后，原任务该回到什么状态；None 表示不动。
+
+    闭环回流只发生在验收通过那一刻——这是"事情真的办完了"的唯一信号。
+    取消不回填：取消意味着不办了，原任务继续躺在 blocked 里等人处理才对。
+    """
+    if ticket_status == "done":
+        return "todo"
+    return None

backend/app/errors.py

@@ -40,6 +40,15 @@ class ErrorCode(IntEnum):
     ITEM_NOT_FOUND = 5001
     ITEM_STATUS_INVALID = 5002
     TAG_NOT_FOUND = 5003
+    LINK_NOT_FOUND = 5004
+    LINK_CONFLICT = 5005
+
+    # 6xxx 工单域
+    TICKET_NOT_FOUND = 6001
+    TICKET_ALREADY_EXISTS = 6002
+    TICKET_STATUS_INVALID = 6003
+    TICKET_FORBIDDEN = 6004
+    NOTIFICATION_NOT_FOUND = 6005
 
 
 # 默认 HTTP 状态：业务码自己带语义，HTTP 只是传输层
@@ -67,6 +76,13 @@ _HTTP_STATUS: dict[ErrorCode, int] = {
     ErrorCode.ITEM_NOT_FOUND: 404,
     ErrorCode.ITEM_STATUS_INVALID: 409,
     ErrorCode.TAG_NOT_FOUND: 404,
+    ErrorCode.LINK_NOT_FOUND: 404,
+    ErrorCode.LINK_CONFLICT: 409,
+    ErrorCode.TICKET_NOT_FOUND: 404,
+    ErrorCode.TICKET_ALREADY_EXISTS: 409,
+    ErrorCode.TICKET_STATUS_INVALID: 409,
+    ErrorCode.TICKET_FORBIDDEN: 403,
+    ErrorCode.NOTIFICATION_NOT_FOUND: 404,
 }
 
 

backend/app/main.py

@@ -12,7 +12,7 @@ from fastapi.responses import JSONResponse
 from sqlalchemy import text
 from starlette.exceptions import HTTPException as StarletteHTTPException
 
-from .api import auth, items, workspaces
+from .api import auth, items, tickets, workspaces
 from .config import get_settings
 from .db import SessionFactory, current_app_role, engine
 from .errors import AppError, ErrorCode, http_status_for
@@ -102,6 +102,8 @@ app.include_router(auth.router, prefix=settings.api_prefix)
 app.include_router(workspaces.router, prefix=settings.api_prefix)
 app.include_router(items.router, prefix=settings.api_prefix)
 app.include_router(items.tag_router, prefix=settings.api_prefix)
+app.include_router(tickets.router, prefix=settings.api_prefix)
+app.include_router(tickets.notify_router, prefix=settings.api_prefix)
 
 
 @app.get("/health", tags=["运维"], summary="存活探针")

backend/app/models.py

@@ -133,9 +133,10 @@ class AuditLog(Base):
 
 # 需要 RLS 的租户表（迁移与测试共用这一份清单，避免漏加）
 #
-# 任务域的表在 models_task.py —— 它 import 本模块的 Base，所以只能在文件末尾反向 import，
+# 任务域的表在 models_task.py、工单域在 models_ticket.py —— 它们 import 本模块的 Base，
-# 否则循环导入。
+# 所以只能在文件末尾反向 import，否则循环导入。
 from .models_task import TASK_TENANT_TABLES  # noqa: E402
+from .models_ticket import TICKET_TENANT_TABLES  # noqa: E402
 
 TENANT_TABLES: tuple[str, ...] = (
     "workspace",
@@ -143,4 +144,5 @@ TENANT_TABLES: tuple[str, ...] = (
     "invitation",
     "audit_log",
     *TASK_TENANT_TABLES,
+    *TICKET_TENANT_TABLES,
 )

backend/app/models_ticket.py

@@ -0,0 +1,145 @@
+"""工单域模型（M2）：ticket_ext / item_link / notification。
+
+三条设计约束（对应 PRODUCT-PLAN 的 D1/D2）：
+
+1. **工单不是新表，是 `item` 的一个 kind**。标题、正文、截止、标签、评论、事件流
+   全部复用，M1 写的每一处代码对工单依然成立；`ticket_ext` 只补"谁请谁、验没验收"。
+2. **升级不是搬家**。原任务留在原地并进入 `blocked`（它确实被卡住了），工单是**新建**的一条
+   item，通过 `TicketExt.parent_item_id` 指回去。工单验收通过时把原任务从 blocked 拉回 todo，
+   这就是"闭环回流"——任务那条时间线完整保留，历史不被搬来搬去。
+3. **通知只记录"发生了什么"，不做已读以外的状态机**。通知看错、漏看是常态，
+   真正的状态永远以 item 为准，通知只是提醒，不能变成第二份真相。
+"""
+
+from __future__ import annotations
+
+import uuid
+from datetime import datetime
+
+from sqlalchemy import (
+    CheckConstraint,
+    DateTime,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.dialects.postgresql import UUID as PgUUID
+from sqlalchemy.orm import Mapped, mapped_column
+
+from .models import Base, Timestamps, new_uuid
+
+# 工单状态机（与任务共用 item.status 这一列，但取值集合不同）
+TICKET_STATUSES: tuple[str, ...] = (
+    "open",
+    "accepted",
+    "in_progress",
+    "pending_verify",
+    "done",
+    "canceled",
+)
+
+LINK_TYPES: tuple[str, ...] = ("parent", "related")
+
+
+class TicketExt(Base, Timestamps):
+    """工单的附加信息，与 item 一比一。"""
+
+    __tablename__ = "ticket_ext"
+    __table_args__ = (
+        Index("ix_ticket_ws_assignee", "workspace_id", "assignee_id"),
+        Index("ix_ticket_ws_requester", "workspace_id", "requester_id"),
+        CheckConstraint("reject_count >= 0", name="ck_ticket_reject_count"),
+    )
+
+    # item_id 既是主键又是外键：一条 item 最多一条 ticket_ext，天然 1:1
+    item_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="CASCADE"), primary_key=True
+    )
+    workspace_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
+    )
+    # 谁提的这件事（下单的人）
+    requester_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="RESTRICT"), nullable=False
+    )
+    # 谁在办（冗余一份，免得每次查工单都要回 item 表；两处必须一致，改指派时同时写）
+    assignee_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="RESTRICT"), nullable=False
+    )
+    # 从哪条任务升级来的；独立创建的工单为空
+    parent_item_id: Mapped[uuid.UUID | None] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="SET NULL")
+    )
+    accepted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    submitted_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    verified_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    verified_by: Mapped[uuid.UUID | None] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="SET NULL")
+    )
+    # 被驳回次数：不是为了考核，是为了让提需求的人知道"这事返工过几次"
+    reject_count: Mapped[int] = mapped_column(Integer, nullable=False, default=0)
+    # 办完的结论：写什么由办理人决定，M3 从这里长知识草稿
+    resolution_md: Mapped[str] = mapped_column(Text, nullable=False, default="")
+
+
+class ItemLink(Base, Timestamps):
+    """事项之间的关系。M2 只做两种：父子（子任务）与相关。
+
+    `from_item_id → to_item_id` 是有方向的：parent 表示 from 是 to 的子项。
+    """
+
+    __tablename__ = "item_link"
+    __table_args__ = (
+        UniqueConstraint("from_item_id", "to_item_id", "link_type", name="uq_item_link"),
+        CheckConstraint("from_item_id <> to_item_id", name="ck_item_link_not_self"),
+        CheckConstraint("link_type IN ('parent','related')", name="ck_item_link_type"),
+        Index("ix_item_link_ws_from", "workspace_id", "from_item_id"),
+        Index("ix_item_link_ws_to", "workspace_id", "to_item_id"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
+    workspace_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
+    )
+    from_item_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="CASCADE"), nullable=False
+    )
+    to_item_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="CASCADE"), nullable=False
+    )
+    link_type: Mapped[str] = mapped_column(String(16), nullable=False, default="related")
+
+
+class Notification(Base, Timestamps):
+    """站内通知。收件人维度，不是工作区里的公共消息流。"""
+
+    __tablename__ = "notification"
+    __table_args__ = (
+        Index("ix_notification_ws_user_read", "workspace_id", "user_id", "read_at"),
+        Index("ix_notification_ws_created", "workspace_id", "created_at"),
+    )
+
+    id: Mapped[uuid.UUID] = mapped_column(PgUUID(as_uuid=True), primary_key=True, default=new_uuid)
+    workspace_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("workspace.id", ondelete="CASCADE"), nullable=False
+    )
+    user_id: Mapped[uuid.UUID] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("app_user.id", ondelete="CASCADE"), nullable=False
+    )
+    kind: Mapped[str] = mapped_column(String(32), nullable=False)
+    item_id: Mapped[uuid.UUID | None] = mapped_column(
+        PgUUID(as_uuid=True), ForeignKey("item.id", ondelete="CASCADE")
+    )
+    title: Mapped[str] = mapped_column(String(120), nullable=False)
+    body: Mapped[str] = mapped_column(String(300), nullable=False, default="")
+    read_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), nullable=False, server_default=func.now()
+    )
+
+
+TICKET_TENANT_TABLES: tuple[str, ...] = ("ticket_ext", "item_link", "notification")

backend/app/serializers.py

@@ -0,0 +1,61 @@
+"""统一的 item / tag 序列化。
+
+为什么要单独一份：工单接口最初自己拼了一份 dict，忘了 `tags` 字段，
+前端列表把 `item.tags` 当数组用，一升级成工单就整页白屏。
+任何返回 item 的接口都必须用这里的 `item_out`，并用 `tags_of` 把标签一起取回来。
+"""
+
+from __future__ import annotations
+
+import uuid
+from typing import Any
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from .models_task import Item, ItemTag, Tag
+
+
+def tag_out(tag: Tag) -> dict[str, Any]:
+    return {"id": str(tag.id), "name": tag.name, "color": tag.color}
+
+
+def item_out(item: Item, tags: list[Tag] | None = None) -> dict[str, Any]:
+    return {
+        "id": str(item.id),
+        "kind": item.kind,
+        "title": item.title,
+        "body_md": item.body_md,
+        "status": item.status,
+        "priority": item.priority,
+        "creator_id": str(item.creator_id),
+        "assignee_id": str(item.assignee_id) if item.assignee_id else None,
+        "due_at": item.due_at,
+        "completed_at": item.completed_at,
+        "sort_order": item.sort_order,
+        "deleted_at": item.deleted_at,
+        "tags": [tag_out(t) for t in (tags or [])],
+        "created_at": item.created_at,
+        "updated_at": item.updated_at,
+        "version": item.version,
+    }
+
+
+async def tags_of(
+    session: AsyncSession, ws_id: uuid.UUID, item_ids: list[uuid.UUID]
+) -> dict[uuid.UUID, list[Tag]]:
+    """一次把这些事项的标签查回来（别在循环里逐条查）。"""
+    if not item_ids:
+        return {}
+    rows = (
+        await session.execute(
+            select(ItemTag.item_id, Tag)
+            .join(Tag, Tag.id == ItemTag.tag_id)
+            .where(ItemTag.workspace_id == ws_id, ItemTag.item_id.in_(item_ids))
+            .order_by(Tag.name)
+        )
+    ).all()
+    grouped: dict[uuid.UUID, list[Tag]] = {}
+    for item_id, tag in rows:
+        grouped.setdefault(item_id, []).append(tag)
+    return grouped

backend/tests/conftest.py

@@ -55,6 +55,17 @@ async def clean_database(migrated_database: None) -> AsyncIterator[None]:
         settings.resolved_migration_database_url(), poolclass=NullPool
     )
     async with owner_engine.begin() as conn:
+        # 上一次跑挂掉（Ctrl-C、进程被杀）会留下 idle in transaction 的连接，
+        # 它握着行锁，TRUNCATE 就会一直等——不是测试慢，是被锁住了。
+        # 测试库是独占的，直接踢掉除自己以外的会话，再给个 lock_timeout 兜底。
+        await conn.execute(
+            text(
+                "SELECT pg_terminate_backend(pid) FROM pg_stat_activity "
+                "WHERE datname = current_database() "
+                "AND pid <> pg_backend_pid() AND state <> 'idle'"
+            )
+        )
+        await conn.execute(text("SET lock_timeout = '15s'"))
         await conn.execute(text(f"TRUNCATE {_TABLES} CASCADE"))
     await owner_engine.dispose()
     reset_store_for_tests()

backend/tests/test_tickets.py

@@ -0,0 +1,493 @@
+"""M2 工单域测试：升级、指派、办理、验收、回流、权限、通知。
+
+重点不是"能跑通"，而是这几条容易做错的：
+
+1. 升级后**原任务必须变 blocked**（否则用户以为这事还在自己手上）；
+2. 验收通过后**原任务自动回到 todo**（闭环回流，这是整个产品的卖点）；
+3. 驳回**回 in_progress 而不是 open**，且 `reject_count` 加一；
+4. **办理人不能自己验收**（否则工单退化成打勾）；
+5. 通知发给对的人：提交 → 提需求的人；通过/驳回 → 办理人。
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+import pytest
+from httpx import AsyncClient
+
+from .conftest import create_workspace, login, unwrap
+
+V1 = "/api/v1"
+
+
+async def _member(
+    client: AsyncClient, owner: dict[str, Any], ws: dict[str, Any], role: str = "member"
+) -> dict[str, Any]:
+    """拉一个新人进工作区。"""
+    user = await login(client)
+    inv = unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/invitations", json={"role": role}, headers=owner["headers"]
+        )
+    )
+    unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/join",
+            json={"code": inv["code"]},
+            headers=user["headers"],
+        )
+    )
+    return user
+
+
+async def _task(
+    client: AsyncClient, user: dict[str, Any], ws_id: str, title: str = "整理需求"
+) -> dict:
+    return unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws_id}/items",
+            json={"title": title, "status": "todo"},
+            headers=user["headers"],
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_promote_creates_ticket_and_blocks_task(client: AsyncClient) -> None:
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "工单测试")
+    helper = await _member(client, owner, ws)
+    task = await _task(client, owner, ws["id"])
+
+    data = unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"]},
+            headers=owner["headers"],
+        )
+    )
+
+    # 原任务：还是任务，但被卡住了
+    assert data["task"]["id"] == task["id"]
+    assert data["task"]["status"] == "blocked"
+    assert data["task"]["kind"] == "task"
+
+    # 工单：新的一条 item，指回原任务
+    ticket = data["ticket"]
+    assert ticket["id"] != task["id"]
+    assert ticket["kind"] == "ticket"
+    assert ticket["status"] == "open"
+    assert ticket["ticket"]["requester_id"] == owner["user_id"]
+    assert ticket["ticket"]["assignee_id"] == helper["user_id"]
+    assert ticket["ticket"]["parent_item_id"] == task["id"]
+
+    # 工单不出现在任务列表里（两条流各看各的）
+    tasks = unwrap(
+        await client.get(f"{V1}/workspaces/{ws['id']}/items?view=all", headers=owner["headers"])
+    )
+    assert [t["id"] for t in tasks["items"]] == [task["id"]]
+
+    # 被指派人收到通知
+    notes = unwrap(
+        await client.get(f"{V1}/workspaces/{ws['id']}/notifications", headers=helper["headers"])
+    )
+    assert [n["kind"] for n in notes["notifications"]] == ["ticket_assigned"]
+    assert notes["notifications"][0]["item_id"] == ticket["id"]
+
+
+@pytest.mark.asyncio
+async def test_full_flow_reopens_task_on_approval(client: AsyncClient) -> None:
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "闭环")
+    helper = await _member(client, owner, ws)
+    task = await _task(client, owner, ws["id"], "等运维开通权限")
+
+    ticket = unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"]},
+            headers=owner["headers"],
+        )
+    )["ticket"]
+
+    base = f"{V1}/workspaces/{ws['id']}/items/{ticket['id']}/ticket"
+    assert (
+        unwrap(await client.post(f"{base}/accept", headers=helper["headers"]))["status"]
+        == "accepted"
+    )
+    assert (
+        unwrap(await client.post(f"{base}/start", headers=helper["headers"]))["status"]
+        == "in_progress"
+    )
+    submitted = unwrap(
+        await client.post(f"{base}/submit", json={"note": "权限已开通"}, headers=helper["headers"])
+    )
+    assert submitted["status"] == "pending_verify"
+
+    # 提需求的人收到验收通知
+    notes = unwrap(
+        await client.get(f"{V1}/workspaces/{ws['id']}/notifications", headers=owner["headers"])
+    )
+    assert notes["notifications"][0]["kind"] == "ticket_submitted"
+
+    done = unwrap(
+        await client.post(
+            f"{base}/verify",
+            json={"approve": True, "resolution_md": "权限已开通，验证通过"},
+            headers=owner["headers"],
+        )
+    )
+    assert done["status"] == "done"
+    assert done["ticket"]["verified_by"] == owner["user_id"]
+    assert done["ticket"]["resolution_md"] == "权限已开通，验证通过"
+
+    # 闭环回流：原任务从 blocked 回到 todo，并且能看到回流事件
+    reopened = unwrap(
+        await client.get(f"{V1}/workspaces/{ws['id']}/items/{task['id']}", headers=owner["headers"])
+    )
+    assert reopened["status"] == "todo"
+    assert reopened["completed_at"] is None
+    actions = [a["action"] for a in reopened["activities"]]
+    assert "promoted_to_ticket" in actions
+    assert "ticket_done" in actions
+
+
+@pytest.mark.asyncio
+async def test_reject_goes_back_to_in_progress_and_counts(client: AsyncClient) -> None:
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "驳回")
+    helper = await _member(client, owner, ws)
+    task = await _task(client, owner, ws["id"])
+
+    ticket = unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"]},
+            headers=owner["headers"],
+        )
+    )["ticket"]
+    base = f"{V1}/workspaces/{ws['id']}/items/{ticket['id']}/ticket"
+    await client.post(f"{base}/accept", headers=helper["headers"])
+    await client.post(f"{base}/start", headers=helper["headers"])
+    await client.post(f"{base}/submit", json={}, headers=helper["headers"])
+
+    rejected = unwrap(
+        await client.post(
+            f"{base}/verify",
+            json={"approve": False, "note": "截图里没看到生效"},
+            headers=owner["headers"],
+        )
+    )
+    assert rejected["status"] == "in_progress"
+    assert rejected["ticket"]["reject_count"] == 1
+    assert rejected["ticket"]["verified_at"] is None
+
+    # 原任务仍在 blocked（驳回不算办完）
+    still = unwrap(
+        await client.get(f"{V1}/workspaces/{ws['id']}/items/{task['id']}", headers=owner["headers"])
+    )
+    assert still["status"] == "blocked"
+
+    # 办理人收到驳回通知
+    notes = unwrap(
+        await client.get(f"{V1}/workspaces/{ws['id']}/notifications", headers=helper["headers"])
+    )
+    kinds = [n["kind"] for n in notes["notifications"]]
+    assert "ticket_rejected" in kinds
+
+
+@pytest.mark.asyncio
+async def test_assignee_cannot_self_approve(client: AsyncClient) -> None:
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "越权")
+    helper = await _member(client, owner, ws)
+    task = await _task(client, owner, ws["id"])
+    ticket = unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"]},
+            headers=owner["headers"],
+        )
+    )["ticket"]
+    base = f"{V1}/workspaces/{ws['id']}/items/{ticket['id']}/ticket"
+    await client.post(f"{base}/accept", headers=helper["headers"])
+    await client.post(f"{base}/start", headers=helper["headers"])
+    await client.post(f"{base}/submit", json={}, headers=helper["headers"])
+
+    resp = await client.post(f"{base}/verify", json={"approve": True}, headers=helper["headers"])
+    assert resp.json()["code"] == 6004  # TICKET_FORBIDDEN
+
+
+@pytest.mark.asyncio
+async def test_illegal_transition_reports_allowed(client: AsyncClient) -> None:
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "状态机")
+    helper = await _member(client, owner, ws)
+    task = await _task(client, owner, ws["id"])
+    ticket = unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"]},
+            headers=owner["headers"],
+        )
+    )["ticket"]
+
+    resp = await client.post(
+        f"{V1}/workspaces/{ws['id']}/items/{ticket['id']}/ticket/submit",
+        json={},
+        headers=helper["headers"],
+    )
+    body = resp.json()
+    assert body["code"] == 6003
+    assert body["data"]["from"] == "open"
+    # allowed 回答"现在还能去哪儿"，allowed_from 回答"这个动作要什么前置状态"
+    assert body["data"]["allowed"] == ["accepted", "canceled"]
+    assert body["data"]["allowed_from"] == ["in_progress"]
+
+
+@pytest.mark.asyncio
+async def test_promote_requires_member_assignee_and_no_double(client: AsyncClient) -> None:
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "升级校验")
+    task = await _task(client, owner, ws["id"])
+    stranger = await login(client)  # 不在工作区
+
+    bad = await client.post(
+        f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+        json={"assignee_id": stranger["user_id"]},
+        headers=owner["headers"],
+    )
+    assert bad.json()["code"] == 1001
+
+    helper = await _member(client, owner, ws)
+    unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"]},
+            headers=owner["headers"],
+        )
+    )
+    twice = await client.post(
+        f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+        json={"assignee_id": helper["user_id"]},
+        headers=owner["headers"],
+    )
+    assert twice.json()["code"] == 6002
+
+
+@pytest.mark.asyncio
+async def test_ticket_list_scopes(client: AsyncClient) -> None:
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "列表")
+    helper = await _member(client, owner, ws)
+    task = await _task(client, owner, ws["id"], "等设计稿")
+    unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"]},
+            headers=owner["headers"],
+        )
+    )
+
+    mine_as_helper = unwrap(
+        await client.get(
+            f"{V1}/workspaces/{ws['id']}/tickets?role=assigned", headers=helper["headers"]
+        )
+    )
+    assert len(mine_as_helper["tickets"]) == 1
+    assert mine_as_helper["tickets"][0]["title"] == "等设计稿"
+
+    requested = unwrap(
+        await client.get(
+            f"{V1}/workspaces/{ws['id']}/tickets?role=requested", headers=owner["headers"]
+        )
+    )
+    assert len(requested["tickets"]) == 1
+
+    # 只按状态过滤
+    empty = unwrap(
+        await client.get(
+            f"{V1}/workspaces/{ws['id']}/tickets?role=assigned&status=done",
+            headers=helper["headers"],
+        )
+    )
+    assert empty["tickets"] == []
+
+
+@pytest.mark.asyncio
+async def test_cross_tenant_ticket_invisible(client: AsyncClient) -> None:
+    alice = await login(client)
+    bob = await login(client)
+    ws_a = await create_workspace(client, alice, "A 公司")
+    ws_b = await create_workspace(client, bob, "B 公司")
+
+    helper = await _member(client, alice, ws_a)
+    task = await _task(client, alice, ws_a["id"])
+    ticket = unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws_a['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"]},
+            headers=alice["headers"],
+        )
+    )["ticket"]
+
+    # 用 B 工作区去访问 A 的工单：一律"不存在"，不泄漏存在性
+    detail = await client.get(
+        f"{V1}/workspaces/{ws_b['id']}/items/{ticket['id']}", headers=bob["headers"]
+    )
+    assert detail.json()["code"] in (5001, 3003)
+    act = await client.post(
+        f"{V1}/workspaces/{ws_b['id']}/items/{ticket['id']}/ticket/accept", headers=bob["headers"]
+    )
+    assert act.json()["code"] in (5001, 3003)
+
+    # A 的工单列表在 B 里是空的
+    listing = unwrap(
+        await client.get(f"{V1}/workspaces/{ws_b['id']}/tickets?role=all", headers=bob["headers"])
+    )
+    assert listing["tickets"] == []
+
+
+@pytest.mark.asyncio
+async def test_links_parent_child_and_related(client: AsyncClient) -> None:
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "关系")
+    parent = await _task(client, owner, ws["id"], "上线新版")
+    child = await _task(client, owner, ws["id"], "写发布说明")
+    other = await _task(client, owner, ws["id"], "相关：监控告警")
+
+    unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{parent['id']}/links",
+            json={"to_item_id": child["id"], "link_type": "parent"},
+            headers=owner["headers"],
+        )
+    )
+    unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{parent['id']}/links",
+            json={"to_item_id": other["id"], "link_type": "related"},
+            headers=owner["headers"],
+        )
+    )
+
+    links = unwrap(
+        await client.get(
+            f"{V1}/workspaces/{ws['id']}/items/{parent['id']}/links", headers=owner["headers"]
+        )
+    )
+    assert [c["item"]["id"] for c in links["children"]] == [child["id"]]
+    assert [r["item"]["id"] for r in links["related"]] == [other["id"]]
+    assert links["parents"] == []
+
+    # 从子项看，父项在 parents 里
+    from_child = unwrap(
+        await client.get(
+            f"{V1}/workspaces/{ws['id']}/items/{child['id']}/links", headers=owner["headers"]
+        )
+    )
+    assert [p["item"]["id"] for p in from_child["parents"]] == [parent["id"]]
+
+    # 重复挂同一条关系 → 冲突
+    dup = await client.post(
+        f"{V1}/workspaces/{ws['id']}/items/{parent['id']}/links",
+        json={"to_item_id": child["id"], "link_type": "parent"},
+        headers=owner["headers"],
+    )
+    assert dup.json()["code"] == 5005
+
+
+@pytest.mark.asyncio
+async def test_notifications_read_flow(client: AsyncClient) -> None:
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "通知")
+    helper = await _member(client, owner, ws)
+    task = await _task(client, owner, ws["id"])
+    unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"]},
+            headers=owner["headers"],
+        )
+    )
+
+    count = unwrap(
+        await client.get(
+            f"{V1}/workspaces/{ws['id']}/notifications/unread-count", headers=helper["headers"]
+        )
+    )
+    assert count["unread"] == 1
+
+    notes = unwrap(
+        await client.get(f"{V1}/workspaces/{ws['id']}/notifications", headers=helper["headers"])
+    )
+    note_id = notes["notifications"][0]["id"]
+    unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/notifications/{note_id}/read", headers=helper["headers"]
+        )
+    )
+    after = unwrap(
+        await client.get(
+            f"{V1}/workspaces/{ws['id']}/notifications/unread-count", headers=helper["headers"]
+        )
+    )
+    assert after["unread"] == 0
+
+    # 别人的通知看不到（同工作区的另一个人）
+    other = await _member(client, owner, ws)
+    stranger_notes = unwrap(
+        await client.get(
+            f"{V1}/workspaces/{ws['id']}/notifications?unread_only=false", headers=other["headers"]
+        )
+    )
+    assert stranger_notes["notifications"] == []
+
+
+@pytest.mark.asyncio
+async def test_ticket_payload_always_has_tags(client: AsyncClient) -> None:
+    """回归：工单接口曾经自己拼 item 字典，漏了 `tags`，
+    前端 `item.tags.map(...)` 直接白屏。任何返回 item 的接口都得带 tags 数组。"""
+    owner = await login(client)
+    ws = await create_workspace(client, owner, "标签回归")
+    helper = await _member(client, owner, ws)
+    task = await _task(client, owner, ws["id"], "带标签的任务")
+
+    tagged = unwrap(
+        await client.patch(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}",
+            json={"tag_names": ["权限", "运维"], "version": task["version"]},
+            headers=owner["headers"],
+        )
+    )
+
+    data = unwrap(
+        await client.post(
+            f"{V1}/workspaces/{ws['id']}/items/{task['id']}/promote",
+            json={"assignee_id": helper["user_id"], "version": tagged["version"]},
+            headers=owner["headers"],
+        )
+    )
+    # 工单是独立的一条事项，标签不自动继承；关键是字段必须在、而且必须是数组
+    assert isinstance(data["task"]["tags"], list)
+    assert isinstance(data["ticket"]["tags"], list)
+    assert sorted(t["name"] for t in tagged["tags"]) == ["权限", "运维"]
+
+    # 列表与详情也要一致
+    listing = unwrap(
+        await client.get(f"{V1}/workspaces/{ws['id']}/tickets?role=all", headers=owner["headers"])
+    )
+    assert isinstance(listing["tickets"][0]["tags"], list)
+
+    # 详情接口（任务与工单走同一个入口）都必须带 tags
+    task_detail = unwrap(
+        await client.get(f"{V1}/workspaces/{ws['id']}/items/{task['id']}", headers=owner["headers"])
+    )
+    assert sorted(t["name"] for t in task_detail["tags"]) == ["权限", "运维"]
+    ticket_detail = unwrap(
+        await client.get(
+            f"{V1}/workspaces/{ws['id']}/items/{data['ticket']['id']}", headers=owner["headers"]
+        )
+    )
+    assert isinstance(ticket_detail["tags"], list)