diff --git a/alembic/versions/f9f4eaef2e6f_task_claims_and_task_display_fields.py b/alembic/versions/f9f4eaef2e6f_task_claims_and_task_display_fields.py
new file mode 100644
index 0000000..13550d8
--- /dev/null
+++ b/alembic/versions/f9f4eaef2e6f_task_claims_and_task_display_fields.py
@@ -0,0 +1,59 @@
+"""task claims and task display fields
+
+Revision ID: f9f4eaef2e6f
+Revises: 0007_user_classification
+Create Date: 2026-08-25 18:23:09.123825
+
+仅包含本次「任务中心 + 扫码接单」改动：tasks 增加展示/流转字段 + 新增 task_claims 流水表。
+（autogenerate 曾误带 training_* 删除与 users 列改动，已剔除，勿并入。）
+"""
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+
+revision = 'f9f4eaef2e6f'
+down_revision = '0007_user_classification'
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ---- task_claims 接单流水表 ----
+    op.create_table('task_claims',
+    sa.Column('id', sa.String(), nullable=False),
+    sa.Column('task_id', sa.String(), nullable=False),
+    sa.Column('claimer_user_id', sa.String(), nullable=True),
+    sa.Column('claimer_name', sa.String(), nullable=False, server_default=''),
+    sa.Column('claim_source', sa.String(), nullable=False, server_default='scan'),
+    sa.Column('claimed_at', sa.String(), nullable=False, server_default=''),
+    sa.Column('status', sa.String(), nullable=False, server_default='claimed'),
+    sa.Column('created_at', sa.String(), nullable=False, server_default=''),
+    sa.Column('updated_at', sa.String(), nullable=False, server_default=''),
+    sa.ForeignKeyConstraint(['claimer_user_id'], ['users.id'], ondelete='SET NULL'),
+    sa.ForeignKeyConstraint(['task_id'], ['tasks.id'], ondelete='CASCADE'),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_index(op.f('ix_task_claims_task_id'), 'task_claims', ['task_id'], unique=False)
+
+    # ---- tasks 增加展示/流转字段 ----
+    op.add_column('tasks', sa.Column('task_code', sa.String(), nullable=True))
+    op.add_column('tasks', sa.Column('tags', sa.String(), nullable=True))
+    op.add_column('tasks', sa.Column('display_priority', sa.Integer(), nullable=True))
+    op.add_column('tasks', sa.Column('claimed_by', sa.String(), nullable=True))
+    op.add_column('tasks', sa.Column('claimed_at', sa.String(), nullable=True))
+    op.add_column('tasks', sa.Column('doing_at', sa.String(), nullable=True))
+    op.create_index(op.f('ix_tasks_task_code'), 'tasks', ['task_code'], unique=False)
+
+
+def downgrade() -> None:
+    op.drop_index(op.f('ix_tasks_task_code'), table_name='tasks')
+    op.drop_column('tasks', 'doing_at')
+    op.drop_column('tasks', 'claimed_at')
+    op.drop_column('tasks', 'claimed_by')
+    op.drop_column('tasks', 'display_priority')
+    op.drop_column('tasks', 'tags')
+    op.drop_column('tasks', 'task_code')
+    op.drop_index(op.f('ix_task_claims_task_id'), table_name='task_claims')
+    op.drop_table('task_claims')
diff --git a/app/api/routers/rbac_opc.py b/app/api/routers/rbac_opc.py
index d54a359..6271787 100644
--- a/app/api/routers/rbac_opc.py
+++ b/app/api/routers/rbac_opc.py
@@ -10,7 +10,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request
 from pydantic import BaseModel
 
 from ..dependencies import get_db
-from ..schemas.opc import BidRequest, ProfileUpdate, FinanceRecordCreate
+from ..schemas.opc import BidRequest, ProfileUpdate, FinanceRecordCreate, TaskClaimRequest
 from ...rbac import require_roles, write_audit
 from ...infrastructure.repositories import Database, new_id, utcnow_iso
 from ...infrastructure.models import FinanceRecord
@@ -132,6 +132,54 @@ async def opc_grab(
     return updated
 
 
+@router.post("/tasks/grab-by-code", summary="扫码/按短码接单")
+async def opc_grab_by_code(
+    req: TaskClaimRequest,
+    request: Request,
+    db: Database = Depends(get_db),
+    actor: dict = Depends(require_roles("opc_member")),
+):
+    from ...services.task_service import TaskService
+
+    updated = await TaskService(db).claim(
+        (req.task_code or "").strip(), actor, source="scan",
+    )
+    await write_audit(db, action="task.claim", resource="task",
+                resource_id=updated["id"], detail=actor.get("username"),
+                user=actor, request=request)
+    return updated
+
+
+@router.post("/tasks/{task_id}/doing", summary="开始做单")
+async def opc_task_doing(
+    task_id: str,
+    request: Request,
+    db: Database = Depends(get_db),
+    actor: dict = Depends(require_roles("opc_member")),
+):
+    from ...services.task_service import TaskService
+
+    updated = await TaskService(db).start_doing(task_id, actor)
+    await write_audit(db, action="task.doing", resource="task", resource_id=task_id,
+                user=actor, request=request)
+    return updated
+
+
+@router.post("/tasks/{task_id}/complete", summary="完成做单")
+async def opc_task_complete(
+    task_id: str,
+    request: Request,
+    db: Database = Depends(get_db),
+    actor: dict = Depends(require_roles("opc_member")),
+):
+    from ...services.task_service import TaskService
+
+    updated = await TaskService(db).complete(task_id, actor)
+    await write_audit(db, action="task.complete", resource="task", resource_id=task_id,
+                user=actor, request=request)
+    return updated
+
+
 @router.post("/tasks/{task_id}/bid", summary="OPC 投标")
 async def opc_bid(
     task_id: str,
diff --git a/app/api/routers/rbac_operator.py b/app/api/routers/rbac_operator.py
index 9630632..22d7787 100644
--- a/app/api/routers/rbac_operator.py
+++ b/app/api/routers/rbac_operator.py
@@ -11,7 +11,7 @@ from fastapi import APIRouter, Depends, HTTPException, Request, Response
 from pydantic import BaseModel
 
 from ..dependencies import get_db
-from ..schemas.operator import TaskCreateRequest, TaskStatusRequest, ProviderCreateRequest, ProviderUpdateRequest, ContentCreateRequest, ContentStatusRequest, ConfigUpdateRequest, ComputePingResponse, ComputeProvisionRequest, ComputeProvisionResponse, CourseCreateRequest, CourseStatusRequest, ActivityStatusRequest, BookingUpdateRequest, TestCreateRequest, TestStatusRequest
+from ..schemas.operator import TaskCreateRequest, TaskUpdateRequest, TaskStatusRequest, ProviderCreateRequest, ProviderUpdateRequest, ContentCreateRequest, ContentStatusRequest, ConfigUpdateRequest, ComputePingResponse, ComputeProvisionRequest, ComputeProvisionResponse, CourseCreateRequest, CourseStatusRequest, ActivityCreateRequest, ActivityStatusRequest, BookingUpdateRequest, TestCreateRequest, TestStatusRequest
 from ...rbac import require_permission, require_roles, write_audit
 from ...infrastructure.repositories import Database
 from ...services import compute_client
@@ -36,12 +36,41 @@ async def create_task(
     db: Database = Depends(get_db),
     actor: dict = Depends(require_permission("action:task.manage")),
 ):
-    task = await db.tasks.create(req.model_dump(exclude_none=True))
+    fields = req.model_dump(exclude_none=True)
+    if not fields.get("task_code"):
+        fields["task_code"] = _gen_task_code()
+    task = await db.tasks.create(fields)
     await write_audit(db, action="task.create", resource="task", resource_id=task["id"],
                 detail=task["title"], user=actor, request=request)
     return task
 
 
+@router.patch("/tasks/{task_id}", summary="更新任务（增改）")
+async def update_task(
+    task_id: str,
+    req: TaskUpdateRequest,
+    request: Request,
+    db: Database = Depends(get_db),
+    actor: dict = Depends(require_permission("action:task.manage")),
+):
+    if await db.tasks.get(task_id) is None:
+        raise HTTPException(status_code=404, detail="Task not found")
+    task = await db.tasks.update(task_id, req.model_dump(exclude_none=True))
+    await write_audit(db, action="task.update", resource="task", resource_id=task_id,
+                detail=(req.title or ""), user=actor, request=request)
+    return task
+
+
+def _gen_task_code() -> str:
+    """生成便于扫码展示的短码：TK-YYYYMMDD-XXXX（基于时间戳短采样）。"""
+    import time as _t
+    from datetime import datetime as _dt
+
+    stamp = _dt.now().strftime("%Y%m%d")
+    rand = f"{int(_t.time()) % 1000000:06d}"
+    return f"TK-{stamp}-{rand}"
+
+
 @router.post("/tasks/{task_id}/status", summary="更新任务状态")
 async def set_task_status(
     task_id: str,
@@ -326,6 +355,19 @@ async def list_activities(
     return training.list_activities(status=status)
 
 
+@router.post("/activities", summary="创建活动")
+async def create_activity(
+    req: ActivityCreateRequest,
+    request: Request,
+    db: Database = Depends(get_db),
+    actor: dict = Depends(require_permission("action:activity.manage")),
+):
+    activity = training.create_activity(req.model_dump(exclude_none=True))
+    await write_audit(db, action="activity.create", resource="activity", resource_id=activity["id"],
+                detail=activity["title"], user=actor, request=request)
+    return activity
+
+
 @router.post("/activities/{event_id}/status", summary="更新活动状态")
 async def set_activity_status(
     event_id: str,
@@ -400,3 +442,25 @@ async def set_test_status(
     if item is None:
         raise HTTPException(status_code=404, detail="Test not found")
     return item
+
+
+# ── 培训业务 · 调研 / 政策 / 流程日志（落库结果只读）─────────────────────────
+@router.get("/surveys", summary="调研结果列表")
+async def list_surveys(
+    _u: dict = Depends(require_roles("operator")),
+):
+    return training.list_surveys()
+
+
+@router.get("/policies", summary="政策测评结果列表")
+async def list_policies(
+    _u: dict = Depends(require_roles("operator")),
+):
+    return training.list_policies()
+
+
+@router.get("/plans", summary="流程启动日志列表")
+async def list_plans(
+    _u: dict = Depends(require_roles("operator")),
+):
+    return training.list_plans()
diff --git a/app/api/schemas/opc.py b/app/api/schemas/opc.py
index a4663d0..5957a1e 100644
--- a/app/api/schemas/opc.py
+++ b/app/api/schemas/opc.py
@@ -5,6 +5,11 @@ class BidRequest(BaseModel):
     plan: str = ""
 
 
+class TaskClaimRequest(BaseModel):
+    task_code: str = ""
+    task_id: str = ""
+
+
 
 class ProfileUpdate(BaseModel):
     nickname: str | None = None
diff --git a/app/api/schemas/operator.py b/app/api/schemas/operator.py
index 8dcc4f5..22a1b88 100644
--- a/app/api/schemas/operator.py
+++ b/app/api/schemas/operator.py
@@ -5,16 +5,33 @@ class TaskCreateRequest(BaseModel):
     category: str = ""
     sub_category: str = ""
     description: str = ""
+    tags: str = ""
     mode: str = "grab"
     budget_min: int = 0
     budget_max: int = 0
     deadline: str = ""
+    task_code: str = ""
+    display_priority: int = 0
     publisher_name: str = ""
 
 
+class TaskUpdateRequest(BaseModel):
+    title: str | None = None
+    category: str | None = None
+    sub_category: str | None = None
+    description: str | None = None
+    tags: str | None = None
+    mode: str | None = None
+    budget_min: int | None = None
+    budget_max: int | None = None
+    deadline: str | None = None
+    task_code: str | None = None
+    display_priority: int | None = None
+    publisher_name: str | None = None
+
 
 class TaskStatusRequest(BaseModel):
-    status: str  # review/published/completed/cancelled
+    status: str  # review/published/completed/cancelled/claimed/doing
 
 
 
@@ -81,6 +98,22 @@ class CourseStatusRequest(BaseModel):
     status: str  # draft/published/ended/offline
 
 
+class ActivityCreateRequest(BaseModel):
+    title: str
+    type: str | None = None          # free/salon
+    mode: str | None = None          # online/offline
+    status: str | None = None        # open/invite/pending/done/ended
+    subtitle: str | None = None
+    desc: str | None = None
+    location: str | None = None
+    host: str | None = None
+    image: str | None = None
+    link: str | None = None
+    start_at: str | None = None
+    duration_min: int | None = None
+    capacity: int | None = None
+
+
 class ActivityStatusRequest(BaseModel):
     status: str  # open/invite/pending/done/ended
 
diff --git a/app/infrastructure/models.py b/app/infrastructure/models.py
index fea619b..3ad0819 100644
--- a/app/infrastructure/models.py
+++ b/app/infrastructure/models.py
@@ -196,26 +196,57 @@ class Agent(Base):
 # ── 运营端业务模型（平台运营/任务/服务商/内容/配置）────────────────────────
 
 class Task(Base):
-    """平台任务（运营端可审核/下架；后续各端口接入）。"""
+    """平台任务（运营端可审核/下架；后续各端口接入）。
+
+    状态机（扩展支持扫码接单）：draft/pending/review/published/claimed/doing/completed/cancelled。
+    扫码接单流转：published → claimed(claimed_by/claimed_at) → doing(doing_at) → completed。
+    历史接单流水见 TaskClaim；claim 记录 claimed_by/claimed_at/doing_at。
+    """
 
     __tablename__ = "tasks"
 
     id: Mapped[str] = mapped_column(String, primary_key=True)
+    task_code: Mapped[str] = mapped_column(String, default="", index=True)  # 展示/扫码短码(唯一)
     title: Mapped[str] = mapped_column(String, default="")
     category: Mapped[str] = mapped_column(String, default="")   # 设计创意/技术开发/...
     sub_category: Mapped[str] = mapped_column(String, default="")
     description: Mapped[str] = mapped_column(Text, default="")
+    tags: Mapped[str] = mapped_column(String, default="")   # 逗号分隔展示标签
     mode: Mapped[str] = mapped_column(String, default="grab")   # grab/bid/designated/dispatch
     budget_min: Mapped[int] = mapped_column(Integer, default=0)
     budget_max: Mapped[int] = mapped_column(Integer, default=0)
     deadline: Mapped[str] = mapped_column(String, default="")
-    status: Mapped[str] = mapped_column(String, default="draft")  # draft/pending/review/published/completed/cancelled
+    display_priority: Mapped[int] = mapped_column(Integer, default=0)  # 大屏展示排序(越大越前)
+    status: Mapped[str] = mapped_column(String, default="draft")  # draft/pending/review/published/claimed/doing/completed/cancelled
+    claimed_by: Mapped[str | None] = mapped_column(ForeignKey("users.id"), nullable=True)
+    claimed_at: Mapped[str] = mapped_column(String, default="")
+    doing_at: Mapped[str] = mapped_column(String, default="")
     publisher_org_id: Mapped[str | None] = mapped_column(ForeignKey("organizations.id"), nullable=True)
     publisher_name: Mapped[str] = mapped_column(String, default="")
     created_at: Mapped[str] = mapped_column(String, default="")
     updated_at: Mapped[str] = mapped_column(String, default="")
 
 
+class TaskClaim(Base):
+    """任务接单流水（扫码/抢单/投标中标），rich 记录任务状态流转历史。"""
+
+    __tablename__ = "task_claims"
+
+    id: Mapped[str] = mapped_column(String, primary_key=True)
+    task_id: Mapped[str] = mapped_column(
+        ForeignKey("tasks.id", ondelete="CASCADE"), nullable=False, index=True,
+    )
+    claimer_user_id: Mapped[str] = mapped_column(
+        ForeignKey("users.id", ondelete="SET NULL"), nullable=True,
+    )
+    claimer_name: Mapped[str] = mapped_column(String, default="")
+    claim_source: Mapped[str] = mapped_column(String, default="scan")  # scan/grab/bid/dispatch
+    claimed_at: Mapped[str] = mapped_column(String, default="")
+    status: Mapped[str] = mapped_column(String, default="claimed")  # claimed/doing/completed/withdrawn
+    created_at: Mapped[str] = mapped_column(String, default="")
+    updated_at: Mapped[str] = mapped_column(String, default="")
+
+
 class ServiceProvider(Base):
     """服务商（运营端审核/评级/考核）。"""
 
diff --git a/app/infrastructure/repositories.py b/app/infrastructure/repositories.py
index 5077b5e..d32b520 100644
--- a/app/infrastructure/repositories.py
+++ b/app/infrastructure/repositories.py
@@ -46,6 +46,7 @@ from .models import (
     SubsidyApplication,
     SystemConfig,
     Task,
+    TaskClaim,
     TrainingEnrollment,
     User,
     UserIdentity,
@@ -895,10 +896,14 @@ class TaskRepository:
 
     def _to_dict(self, t: Task) -> dict:
         return {
-            "id": t.id, "title": t.title, "category": t.category,
-            "sub_category": t.sub_category, "description": t.description,
+            "id": t.id, "task_code": t.task_code, "title": t.title,
+            "category": t.category, "sub_category": t.sub_category,
+            "description": t.description, "tags": t.tags,
             "mode": t.mode, "budget_min": t.budget_min, "budget_max": t.budget_max,
-            "deadline": t.deadline, "status": t.status,
+            "deadline": t.deadline, "display_priority": t.display_priority,
+            "status": t.status,
+            "claimed_by": t.claimed_by, "claimed_at": t.claimed_at,
+            "doing_at": t.doing_at,
             "publisher_org_id": t.publisher_org_id, "publisher_name": t.publisher_name,
             "created_at": t.created_at, "updated_at": t.updated_at,
         }
@@ -909,17 +914,37 @@ class TaskRepository:
             q = q.where(Task.status == status)
         return [self._to_dict(t) for t in await self.session.scalars(q)]
 
+    async def list_published(self) -> list[dict]:
+        """大屏/广场展示：published + claimed/doing 均可上墙，按 display_priority 降序。"""
+        q = (
+            select(Task)
+            .where(Task.status.in_(["published", "claimed", "doing"]))
+            .order_by(Task.display_priority.desc(), Task.updated_at.desc())
+        )
+        return [self._to_dict(t) for t in await self.session.scalars(q)]
+
     async def get(self, task_id: str) -> dict | None:
         t = await self.session.get(Task, task_id)
         return self._to_dict(t) if t else None
 
+    async def get_by_code(self, task_code: str) -> dict | None:
+        if not task_code:
+            return None
+        q = select(Task).where(Task.task_code == task_code)
+        t = (await self.session.scalars(q)).first()
+        return self._to_dict(t) if t else None
+
     async def create(self, fields: dict) -> dict:
         now = utcnow_iso()
-        t = Task(id=new_id("task"), title=fields.get("title", ""),
+        t = Task(id=new_id("task"), task_code=fields.get("task_code", ""),
+                 title=fields.get("title", ""),
                  category=fields.get("category", ""), sub_category=fields.get("sub_category", ""),
-                 description=fields.get("description", ""), mode=fields.get("mode", "grab"),
+                 description=fields.get("description", ""),
+                 tags=fields.get("tags", ""), mode=fields.get("mode", "grab"),
                  budget_min=fields.get("budget_min", 0), budget_max=fields.get("budget_max", 0),
-                 deadline=fields.get("deadline", ""), status=fields.get("status", "draft"),
+                 deadline=fields.get("deadline", ""),
+                 display_priority=fields.get("display_priority", 0),
+                 status=fields.get("status", "draft"),
                  publisher_org_id=fields.get("publisher_org_id"),
                  publisher_name=fields.get("publisher_name", ""),
                  created_at=now, updated_at=now)
@@ -927,6 +952,21 @@ class TaskRepository:
         await self.session.commit()
         return self._to_dict(t)
 
+    async def update(self, task_id: str, fields: dict) -> dict | None:
+        t = await self.session.get(Task, task_id)
+        if t is None:
+            return None
+        for key in (
+            "task_code", "title", "category", "sub_category", "description",
+            "tags", "mode", "budget_min", "budget_max", "deadline",
+            "display_priority", "publisher_name",
+        ):
+            if key in fields:
+                setattr(t, key, fields.get(key))
+        t.updated_at = utcnow_iso()
+        await self.session.commit()
+        return self._to_dict(t)
+
     async def set_status(self, task_id: str, status: str) -> dict | None:
         t = await self.session.get(Task, task_id)
         if t is None:
@@ -936,6 +976,70 @@ class TaskRepository:
         await self.session.commit()
         return self._to_dict(t)
 
+    async def claim(self, task_id: str, user_id: str) -> dict | None:
+        """published → claimed：记录 claimed_by / claimed_at。"""
+        t = await self.session.get(Task, task_id)
+        if t is None:
+            return None
+        t.status = "claimed"
+        t.claimed_by = user_id
+        t.claimed_at = t.claimed_at or utcnow_iso()
+        t.updated_at = utcnow_iso()
+        await self.session.commit()
+        return self._to_dict(t)
+
+    async def set_doing(self, task_id: str) -> dict | None:
+        """claimed → doing：记录 doing_at。"""
+        t = await self.session.get(Task, task_id)
+        if t is None:
+            return None
+        t.status = "doing"
+        t.doing_at = t.doing_at or utcnow_iso()
+        t.updated_at = utcnow_iso()
+        await self.session.commit()
+        return self._to_dict(t)
+
+
+class TaskClaimRepository:
+    def __init__(self, session: AsyncSession):
+        self.session = session
+
+    def _to_dict(self, c: TaskClaim) -> dict:
+        return {
+            "id": c.id, "task_id": c.task_id,
+            "claimer_user_id": c.claimer_user_id, "claimer_name": c.claimer_name,
+            "claim_source": c.claim_source, "claimed_at": c.claimed_at,
+            "status": c.status, "created_at": c.created_at, "updated_at": c.updated_at,
+        }
+
+    async def create(self, task_id: str, user_id: str, name: str,
+                     source: str = "scan") -> dict:
+        now = utcnow_iso()
+        c = TaskClaim(id=new_id("claim"), task_id=task_id,
+                      claimer_user_id=user_id, claimer_name=name,
+                      claim_source=source, claimed_at=now,
+                      status="claimed", created_at=now, updated_at=now)
+        self.session.add(c)
+        await self.session.commit()
+        return self._to_dict(c)
+
+    async def list_by_task(self, task_id: str) -> list[dict]:
+        q = select(TaskClaim).where(TaskClaim.task_id == task_id)
+        return [self._to_dict(c) for c in await self.session.scalars(q)]
+
+    async def list_by_claimer(self, user_id: str) -> list[dict]:
+        q = select(TaskClaim).where(TaskClaim.claimer_user_id == user_id)
+        return [self._to_dict(c) for c in await self.session.scalars(q)]
+
+    async def set_status(self, claim_id: str, status: str) -> dict | None:
+        c = await self.session.get(TaskClaim, claim_id)
+        if c is None:
+            return None
+        c.status = status
+        c.updated_at = utcnow_iso()
+        await self.session.commit()
+        return self._to_dict(c)
+
 
 # ---------------------------------------------------------------------------
 # 服务商
@@ -1916,6 +2020,7 @@ class Database:
         self.regions = RegionRepository(self.session)
         self.audit = AuditRepository(self.session)
         self.tasks = TaskRepository(self.session)
+        self.task_claims = TaskClaimRepository(self.session)
         self.providers = ProviderRepository(self.session)
         self.content = ContentRepository(self.session)
         self.config = ConfigRepository(self.session)
diff --git a/app/infrastructure/seed.py b/app/infrastructure/seed.py
index df40097..c38ebf3 100644
--- a/app/infrastructure/seed.py
+++ b/app/infrastructure/seed.py
@@ -355,21 +355,34 @@ async def _seed_operator_business(session: AsyncSession, now: str) -> None:
     if await session.scalar(select(Task.id).limit(1)) is not None:
         return
     session.add_all([
-        Task(id="task_001", title="电商小程序首页 UI 设计", category="设计创意",
-             sub_category="UI设计", description="为电商小程序设计首页 UI，含 5 个页面",
-             mode="bid", budget_min=2000, budget_max=5000, deadline="2026-09-01",
+        Task(id="task_001", task_code="TK-001-UI", title="电商小程序首页 UI 设计",
+             category="设计创意", sub_category="UI设计",
+             description="为电商小程序设计首页 UI，含 5 个页面",
+             tags="UI,设计", mode="bid", budget_min=2000, budget_max=5000,
+             deadline="2026-09-01", display_priority=10,
              status="published", publisher_org_id="o_ent_001",
              publisher_name="云南某某甲方企业", created_at=now, updated_at=now),
-        Task(id="task_002", title="品牌文案写作（月度）", category="文案写作",
-             sub_category="内容写作", description="每月 8 篇品牌公众号文案",
-             mode="designated", budget_min=3000, budget_max=6000, deadline="2026-12-31",
+        Task(id="task_002", task_code="TK-002-CT", title="品牌文案写作（月度）",
+             category="文案写作", sub_category="内容写作",
+             description="每月 8 篇品牌公众号文案", tags="文案,内容",
+             mode="designated", budget_min=3000, budget_max=6000,
+             deadline="2026-12-31", display_priority=5,
              status="review", publisher_org_id="o_ent_001",
              publisher_name="云南某某甲方企业", created_at=now, updated_at=now),
-        Task(id="task_003", title="小程序前后端开发", category="技术开发",
-             sub_category="小程序开发", description="点餐小程序全栈开发，3 期交付",
-             mode="bid", budget_min=30000, budget_max=60000, deadline="2026-10-15",
+        Task(id="task_003", task_code="TK-003-DEV", title="小程序前后端开发",
+             category="技术开发", sub_category="小程序开发",
+             description="点餐小程序全栈开发，3 期交付", tags="开发",
+             mode="bid", budget_min=30000, budget_max=60000,
+             deadline="2026-10-15", display_priority=20,
              status="pending", publisher_org_id="o_ent_001",
              publisher_name="云南某某甲方企业", created_at=now, updated_at=now),
+        Task(id="task_004", task_code="TK-004-GRAB", title="短视频剪辑（扫码即领）",
+             category="技术开发", sub_category="视频剪辑",
+             description="为园区企业制作 3 条产品短视频，扫码即可接单，2 周内交付。",
+             tags="短视频,剪辑", mode="grab", budget_min=1500, budget_max=3000,
+             deadline="2026-11-30", display_priority=30,
+             status="published", publisher_org_id="o_ent_001",
+             publisher_name="园区运营方", created_at=now, updated_at=now),
     ])
     session.add_all([
         ServiceProvider(id="prov_001", name="云超财税服务所", category="财税服务",
diff --git a/app/park/routers.py b/app/park/routers.py
index e54cc78..e7b9671 100644
--- a/app/park/routers.py
+++ b/app/park/routers.py
@@ -11,12 +11,16 @@ import json
 import logging
 import os
 import time
+from datetime import datetime
 from pathlib import Path
 
-from fastapi import APIRouter, File, Header, HTTPException, Request, UploadFile
+from fastapi import APIRouter, Depends, File, Header, HTTPException, Request, UploadFile
 from fastapi.responses import FileResponse, JSONResponse, StreamingResponse
 from pydantic import BaseModel, Field
+from sqlalchemy.ext.asyncio import AsyncSession
 
+from ..infrastructure.db import get_session
+from ..infrastructure.repositories import TaskRepository
 from . import park_config, tenants
 from .auth import create_device_token, create_token, parse_device_token, parse_token, require_tenant
 from .config import settings
@@ -447,6 +451,41 @@ async def dashboard_overview(authorization: str | None = Header(None), tenant_id
     return get_engine(tid).snapshot()
 
 
+@router.get("/api/tasks", summary="大屏任务展示（平台全局已发布任务）")
+async def park_tasks(
+    authorization: str | None = Header(None),
+    tenant_id: str | None = None,
+    session: AsyncSession = Depends(get_session),
+):
+    """返回系统任务中心的上墙任务（published/claimed/doing），供大屏轮播卡片。
+
+    大屏卡片含任务 ID、task_code 与二维码（scan_payload）。任务为平台全局实体，
+    不受园区隔离；此处按园区身份校验后返回。
+    """
+    await _resolve_tenant(authorization, tenant_id)
+    repo = TaskRepository(session)
+    items = await repo.list_published()
+    now_label = datetime.now().strftime("%Y%m%d")
+
+    def payload(t: dict) -> dict:
+        return {
+            "id": t["id"],
+            "task_code": t["task_code"] or f"TK-{now_label}-{t['id'][-5:]}",
+            "scan_payload": t["task_code"] or f"TK-{now_label}-{t['id'][-5:]}",
+            "title": t["title"],
+            "category": t["category"],
+            "summary": (t["description"] or "")[:160],
+            "budget_min": t["budget_min"],
+            "budget_max": t["budget_max"],
+            "status": t["status"],
+            "claimed_by": t["claimed_by"],
+            "claimed_at": t["claimed_at"],
+            "tags": [tag for tag in (t.get("tags") or "").split(",") if tag],
+        }
+
+    return {"items": [payload(t) for t in items]}
+
+
 @router.get("/api/park/zones")
 async def park_zones(authorization: str | None = Header(None), tenant_id: str | None = None):
     tid = await _resolve_tenant(authorization, tenant_id)
@@ -500,8 +539,10 @@ def _gen_code() -> str:
 
 
 def _publish_bind(device_id: str, payload: dict) -> None:
-    """未绑定大屏仅订阅 bind 频道；绑定成功后经此频道通知并下发 token。"""
-    hub.publish(f"opc/display/bind/{device_id}", payload, qos=1)
+    """未绑定大屏仅订阅 bind 频道；绑定成功后经此频道通知并下发 token。
+    retain=True：绑定/解绑状态持久，屏幕即使延迟连接，订阅时也会立即收到最新状态（避免
+    绑定成功但屏幕 MQTT 尚未订阅导致丢失 bound 事件、无法自动进入）。"""
+    hub.publish(f"opc/display/bind/{device_id}", payload, qos=1, retain=True)
 
 
 @router.post("/api/devices/register")
@@ -545,6 +586,9 @@ async def bind_screen_by_code(tid: str, body: BindByCodeBody):
 async def device_unbind(device_id: str, authorization: str | None = Header(None), tenant_id: str | None = None):
     await _resolve_tenant(authorization, tenant_id)
     ok = await tenants.unbind_device(device_id)
+    if ok:
+        # 通知大屏退出到待绑定页（大屏订阅 bind/<device> 频道，收到 unbound 即清理本地并回绑定页）
+        _publish_bind(device_id, {"event": "unbound"})
     return {"ok": ok, "device_id": device_id}
 
 
diff --git a/app/services/task_service.py b/app/services/task_service.py
index d7fa862..2eb81d6 100644
--- a/app/services/task_service.py
+++ b/app/services/task_service.py
@@ -18,11 +18,49 @@ class TaskService:
         self.db = db
 
     async def grab(self, task_id: str, actor: dict) -> dict:
-        """抢单：仅 published + grab 模式可抢，抢后置 in_progress。"""
+        """抢单（兼容旧入口）：published + grab → claimed，记录接单流水。"""
+        return await self.claim_by_id(task_id, actor, source="grab")
+
+    async def claim(self, task_code: str, actor: dict, source: str = "scan") -> dict:
+        """扫码/按短码接单：解析 task_code 后领单。"""
+        task = await self.db.tasks.get_by_code(task_code)
+        if task is None:
+            raise HTTPException(status_code=404, detail="任务不存在")
+        return await self.claim_by_id(task["id"], actor, source=source)
+
+    async def claim_by_id(self, task_id: str, actor: dict, source: str) -> dict:
+        """published + grab 模式 → claimed（claimed_by/claimed_at + TaskClaim 流水）。"""
         task = await self.db.tasks.get(task_id)
-        if task is None or task["status"] != "published" or task["mode"] != "grab":
-            raise HTTPException(status_code=400, detail="任务不可抢单")
-        return await self.db.tasks.set_status(task_id, "in_progress")
+        if task is None:
+            raise HTTPException(status_code=404, detail="任务不存在")
+        if task["status"] != "published":
+            raise HTTPException(status_code=400, detail="任务不可接单")
+        if task["mode"] != "grab":
+            raise HTTPException(status_code=400, detail="任务不支持扫码接单")
+        updated = await self.db.tasks.claim(task_id, actor["id"])
+        await self.db.task_claims.create(
+            task_id, actor["id"],
+            actor.get("nickname") or actor.get("username", ""), source,
+        )
+        return updated
+
+    async def start_doing(self, task_id: str, actor: dict) -> dict:
+        """claimed → doing。"""
+        task = await self.db.tasks.get(task_id)
+        if task is None:
+            raise HTTPException(status_code=404, detail="任务不存在")
+        if task["status"] != "claimed":
+            raise HTTPException(status_code=400, detail="任务未接单，无法开始")
+        return await self.db.tasks.set_doing(task_id)
+
+    async def complete(self, task_id: str, actor: dict) -> dict:
+        """doing → completed。"""
+        task = await self.db.tasks.get(task_id)
+        if task is None:
+            raise HTTPException(status_code=404, detail="任务不存在")
+        if task["status"] != "doing":
+            raise HTTPException(status_code=400, detail="任务未在进行中，无法完成")
+        return await self.db.tasks.set_status(task_id, "completed")
 
     async def bid(self, task_id: str, actor: dict, quote: int, plan: str) -> dict:
         """投标：仅 published + bid 模式可投。"""
diff --git a/app/services/training_admin_bridge.py b/app/services/training_admin_bridge.py
index 166ba8a..9956ad4 100644
--- a/app/services/training_admin_bridge.py
+++ b/app/services/training_admin_bridge.py
@@ -9,6 +9,7 @@
 """
 from __future__ import annotations
 
+import json
 from datetime import datetime, timezone
 
 from app.training import db as tdb
@@ -52,6 +53,16 @@ def list_activities(status: str | None = None) -> list[dict]:
     return rows
 
 
+def create_activity(data: dict) -> dict:
+    row = dict(data)
+    row["id"] = tdb.gen_id("E-")
+    row.setdefault("mode", "offline")
+    row.setdefault("status", "open")
+    row.setdefault("duration_min", 90)
+    tdb.insert("events", row)
+    return tdb.fetch_by_id("events", row["id"])
+
+
 def set_activity_status(event_id: str, status: str) -> dict | None:
     if tdb.fetch_by_id("events", event_id) is None:
         return None
@@ -136,3 +147,68 @@ def create_test(data: dict) -> dict:
     }
     tdb.insert("tests", row)
     return pack_test(tdb.fetch_by_id("tests", row["id"]))
+
+
+def _parse_json(raw):
+    if raw is None:
+        return None
+    if isinstance(raw, (dict, list)):
+        return raw
+    try:
+        return json.loads(raw)
+    except (TypeError, ValueError):
+        return raw
+
+
+# ── 调研（survey_logs 表）─────────────────────────────────────────────────
+
+def list_surveys() -> list[dict]:
+    rows = tdb.list_all("survey_logs", "created_at DESC")
+    return [
+        {
+            "id": r.get("id"),
+            "username": r.get("username"),
+            "source": r.get("source"),
+            "answers": _parse_json(r.get("answers")),
+            "created_at": r.get("created_at"),
+        }
+        for r in rows
+    ]
+
+
+# ── 政策（policy_logs 表）─────────────────────────────────────────────────
+
+def list_policies() -> list[dict]:
+    rows = tdb.list_all("policy_logs", "created_at DESC")
+    return [
+        {
+            "id": r.get("id"),
+            "username": r.get("username"),
+            "answers": _parse_json(r.get("answers")),
+            "policies_count": r.get("policies_count"),
+            "subsidies_count": r.get("subsidies_count"),
+            "loans_count": r.get("loans_count"),
+            "summary": r.get("summary"),
+            "created_at": r.get("created_at"),
+        }
+        for r in rows
+    ]
+
+
+# ── 流程（plan_logs 表）───────────────────────────────────────────────────
+
+def list_plans() -> list[dict]:
+    rows = tdb.list_all("plan_logs", "created_at DESC")
+    return [
+        {
+            "id": r.get("id"),
+            "username": r.get("username"),
+            "region": r.get("region"),
+            "status": r.get("status"),
+            "need_park": r.get("need_park"),
+            "has_staff": r.get("has_staff"),
+            "steps_count": r.get("steps_count"),
+            "created_at": r.get("created_at"),
+        }
+        for r in rows
+    ]
diff --git a/app/training/main.py b/app/training/main.py
index cf5c5aa..d1d1d73 100644
--- a/app/training/main.py
+++ b/app/training/main.py
@@ -6,6 +6,7 @@
 import asyncio
 import json
 import os
+import secrets
 import time
 from typing import Optional
 from fastapi import FastAPI, Header, Request, HTTPException, UploadFile, File
@@ -18,6 +19,12 @@ from .opc_engine import SECTIONS, AXIS_NAMES, ADAPT_LABELS, PROFILES
 
 from contextlib import asynccontextmanager
 
+# 平台域（任务/OPC 身份）访问：复用唯一总库 + 四层 Repository
+from ..infrastructure.db import AsyncSessionLocal
+from ..infrastructure.repositories import Database
+from ..services.task_service import TaskService
+from ..api.routers.auth import _ensure_opc_identity
+
 
 @asynccontextmanager
 async def lifespan(app):
@@ -202,6 +209,62 @@ async def login(req: Request):
     return {"ok": True, "token": make_token(username), "username": username, "name": acct.get("name") or username}
 
 
+# ============================= 扫码接单（任务中心）=============================
+@app.post("/api/tasks/claim-by-code")
+async def tasks_claim_by_code(request: Request):
+    """扫码接单：task_code → 领单（published → claimed，记录 claimed_by/流水）。"""
+    body = await request.json()
+    code = str(body.get("task_code") or body.get("code") or "").strip()
+    if not code:
+        raise HTTPException(status_code=400, detail="缺少任务码")
+    auth = request.headers.get("Authorization", "")
+    async with AsyncSessionLocal() as session:
+        pdb = Database(session=session)
+        payload = require_auth(auth)
+        username = payload.get("username") or payload.get("sub") or ""
+        acct = await asyncio.to_thread(db.fetch_one, "accounts", username=username)
+        if not acct:
+            raise HTTPException(status_code=401, detail="请先登录")
+        key = (acct.get("phone") or "").strip() or username
+        user = await pdb.users.get_by_username(key)
+        if user is None:
+            user = await pdb.users.create(
+                key, password=secrets.token_hex(16),
+                phone=(acct.get("phone") or "").strip() or "",
+                nickname=(acct.get("name") or "").strip() or "",
+                role="opc_member", source="mini_program", auth_type="phone",
+            )
+        identity = await _ensure_opc_identity(pdb, user["id"])
+        actor = dict(user)
+        actor["id"] = user["id"]
+        actor["username"] = user["username"]
+        actor["nickname"] = user.get("nickname") or user.get("username")
+        actor["port"] = "opc"
+        actor["role"] = identity.get("role") or "opc_member"
+        task = await TaskService(pdb).claim(code, actor, source="scan")
+        return {"ok": True, "task": task}
+
+
+@app.get("/api/tasks/my")
+async def tasks_my_tasks(request: Request):
+    """我的接单（扫码接单领取的任务）。"""
+    auth = request.headers.get("Authorization", "")
+    async with AsyncSessionLocal() as session:
+        pdb = Database(session=session)
+        payload = require_auth(auth)
+        username = payload.get("username") or payload.get("sub") or ""
+        acct = await asyncio.to_thread(db.fetch_one, "accounts", username=username)
+        if not acct:
+            raise HTTPException(status_code=401, detail="请先登录")
+        key = (acct.get("phone") or "").strip() or username
+        user = await pdb.users.get_by_username(key)
+        if user is None:
+            return {"items": []}
+        items = await pdb.tasks.list(status="claimed") + await pdb.tasks.list(status="doing")
+        mine = [t for t in items if t.get("claimed_by") == user["id"]]
+        return {"items": mine}
+
+
 @app.post("/api/auth/send-code")
 async def send_code(req: Request):
     b = await req.json()
diff --git a/tests/test_task_claim.py b/tests/test_task_claim.py
new file mode 100644
index 0000000..712b03c
--- /dev/null
+++ b/tests/test_task_claim.py
@@ -0,0 +1,92 @@
+# -*- coding: utf-8 -*-
+"""任务中心·扫码接单：TaskService 状态机 + Repository 访问（自包含内存库）。"""
+from __future__ import annotations
+
+from contextlib import asynccontextmanager
+
+import pytest
+from fastapi import HTTPException
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from app.infrastructure.models import Base  # noqa: F401  (注册表到 Base)
+from app.infrastructure.repositories import Database
+from app.services.task_service import TaskService
+
+
+@asynccontextmanager
+async def _db():
+    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
+    async with engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)
+    Session = async_sessionmaker(engine, expire_on_commit=False)
+    async with Session() as session:
+        yield Database(session=session)
+    await engine.dispose()
+
+
+def _grab_published(**kw):
+    return dict(
+        task_code="TK-004-GRAB", title="短视频剪辑", mode="grab",
+        status="published", category="技术开发", budget_min=1500, budget_max=3000,
+        display_priority=30, **kw,
+    )
+
+
+async def test_claim_marks_claimed_and_records_claim():
+    """published + grab → claimed（claimed_by/claimed_at），并写 task_claims 流水。"""
+    async with _db() as db:
+        t = await db.tasks.create(_grab_published())
+        actor = {"id": "u_opc_1", "username": "opc01", "nickname": "阿健"}
+        res = await TaskService(db).claim(t["task_code"], actor, source="scan")
+        assert res["status"] == "claimed"
+        assert res["claimed_by"] == "u_opc_1"
+        assert res["claimed_at"]
+        claims = await db.task_claims.list_by_task(t["id"])
+        assert claims and claims[0]["claim_source"] == "scan"
+
+
+async def test_claim_disallowed_non_grab_mode():
+    """bid 模式不可扫码接单。"""
+    async with _db() as db:
+        t = await db.tasks.create(_grab_published(title="UI", mode="bid"))
+        with pytest.raises(HTTPException):
+            await TaskService(db).claim(t["task_code"], {"id": "u1"}, "scan")
+
+
+async def test_claim_unknown_code_404():
+    async with _db() as db:
+        with pytest.raises(HTTPException):
+            await TaskService(db).claim("TK-NOT-EXIST", {"id": "u1"}, "scan")
+
+
+async def test_claim_already_claimed_rejected():
+    """已 claim 的任务再 claim 应报错。"""
+    async with _db() as db:
+        t = await db.tasks.create(_grab_published())
+        await TaskService(db).claim(t["task_code"], {"id": "u1"}, "scan")
+        with pytest.raises(HTTPException):
+            await TaskService(db).claim(t["task_code"], {"id": "u2"}, "scan")
+
+
+async def test_doing_then_complete():
+    """claimed → doing → completed。"""
+    async with _db() as db:
+        t = await db.tasks.create(_grab_published())
+        actor = {"id": "u_opc_1", "username": "opc01", "nickname": "阿健"}
+        await TaskService(db).claim(t["task_code"], actor, source="scan")
+        doing = await TaskService(db).start_doing(t["id"], actor)
+        assert doing["status"] == "doing" and doing["doing_at"]
+        done = await TaskService(db).complete(t["id"], actor)
+        assert done["status"] == "completed"
+
+
+async def test_list_published_returns_display_fields():
+    """大屏取数：list_published 返回新字段且按 display_priority 降序。"""
+    async with _db() as db:
+        await db.tasks.create(_grab_published(task_code="TK-LOW", display_priority=1))
+        await db.tasks.create(_grab_published(task_code="TK-HIGH", display_priority=99))
+        items = await db.tasks.list_published()
+        codes = [t["task_code"] for t in items]
+        assert "TK-LOW" in codes and "TK-HIGH" in codes
+        assert codes.index("TK-HIGH") < codes.index("TK-LOW")
+        assert all("task_code" in t and "claimed_by" in t for t in items)
diff --git a/tests/test_training_persist.py b/tests/test_training_persist.py
new file mode 100644
index 0000000..b4bfa26
--- /dev/null
+++ b/tests/test_training_persist.py
@@ -0,0 +1,119 @@
+# -*- coding: utf-8 -*-
+"""培训业务落库 + 管理端打通 自动化验证。
+
+覆盖用户侧（小程序/网页共用）测评/调研/政策/流程流程「真实写库」，以及
+运营端 /admin/* 管理面能否读到这些落库结果、能否创建活动。
+
+环境说明：
+- /api/* 由 app.training 子应用提供（生产经 dispatcher /api 前缀路由），
+  lifespan 为空操作，建表由 alembic/seed 非运行态完成 —— 这里把训练库指向临时
+  文件并执行 SCHEMA，实现完全隔离。
+- 平台库（app.main）在测试中预置为空（Database.initialize 不在运行时建表/种子，
+  平台 RBAC 测试本身依赖外部迁移+种子，非本次范围）。因此管理端验证分两层：
+  1) 桥接函数 list_* / create_activity 直接对同一训练库读回，证明「落库→管理可见」；
+  2) 匿名访问 /admin/* 应 401/403 而非 404，证明路由已注册且受 RBAC 门卫。
+"""
+from __future__ import annotations
+
+import pytest
+from fastapi.testclient import TestClient
+
+from app.services import training_admin_bridge as training
+from app.training import db as tdb
+from app.training.main import app as training_app
+
+
+@pytest.fixture()
+def training_db(tmp_path, monkeypatch, client):
+    """训练库指向临时文件并建表；client 为 conftest 平台端（仅用于路由门卫校验）。"""
+    tmp_db = tmp_path / "train.db"
+    monkeypatch.setattr(tdb, "DB_PATH", str(tmp_db))
+    conn = tdb.get_conn()
+    conn.executescript(tdb.SCHEMA)
+    conn.commit()
+    conn.close()
+    with TestClient(training_app) as mp:
+        yield mp, client
+
+
+def _assert_physical_row(table: str, rid: str):
+    row = tdb.fetch_by_id(table, rid)
+    assert row is not None, f"{table} 未落库"
+
+
+def test_opc_test_persisted(training_db):
+    mp, _ = training_db
+    calc = mp.post("/api/tests/opc/calculate", json={"version": "quick", "answers": {}})
+    assert calc.status_code == 200, calc.text
+    r = calc.json()["result"]
+    report = mp.post("/api/tests", json={
+        "typeCode": r["typeCode"], "persona": (r["persona"] or {}).get("name", ""),
+        "adaptIndex": r["adaptIndex"], "adaptLevel": (r["adaptLevel"] or {}).get("level", ""),
+        "tracks": [t["code"] for t in (r["tracks"] or [])], "version": "quick", "username": "u_test",
+    })
+    assert report.status_code == 200, report.text
+    tid = report.json()["id"]
+    _assert_physical_row("tests", tid)
+    # 管理面（桥接）可读回
+    assert any(x.get("id") == tid for x in training.list_tests())
+
+
+def test_survey_persisted(training_db):
+    mp, _ = training_db
+    res = mp.post("/api/survey/submit", json={"username": "u_survey", "source": "mp", "answers": {"q1": "A"}})
+    assert res.status_code == 200, res.text
+    sid = res.json()["id"]
+    _assert_physical_row("survey_logs", sid)
+    assert any(x.get("id") == sid for x in training.list_surveys())
+
+
+def test_policy_persisted(training_db):
+    mp, _ = training_db
+    calc = mp.post("/api/policy/calculate", json={
+        "answers": {"status": "base", "region": "yn", "entity": "individual",
+                    "capital": "self", "industry": ["tourism"]}})
+    assert calc.status_code == 200, calc.text
+    rr = calc.json()["result"]
+    res = mp.post("/api/policy-logs", json={
+        "username": "u_pol", "answers": {"status": "base"},
+        "policiesCount": len(rr["policies"]), "subsidiesCount": len(rr["subsidies"]),
+        "loansCount": len(rr["loans"]), "summary": "测试摘要",
+    })
+    assert res.status_code == 200, res.text
+    pid = res.json()["id"]
+    _assert_physical_row("policy_logs", pid)
+    assert any(x.get("id") == pid for x in training.list_policies())
+
+
+def test_plan_persisted(training_db):
+    mp, _ = training_db
+    gen = mp.post("/api/plan/generate", json={"needPark": True, "needRegister": False, "hasStaff": False})
+    assert gen.status_code == 200, gen.text
+    steps = gen.json()["plan"]
+    res = mp.post("/api/plan-logs", json={
+        "username": "u_plan", "region": "yn", "status": "base",
+        "needPark": True, "hasStaff": False, "stepsCount": len(steps),
+    })
+    assert res.status_code == 200, res.text
+    lid = res.json()["id"]
+    _assert_physical_row("plan_logs", lid)
+    assert any(x.get("id") == lid for x in training.list_plans())
+
+
+def test_admin_create_activity(training_db):
+    _mp, _ = training_db
+    item = training.create_activity({
+        "title": "新增公益课", "type": "free", "mode": "online",
+        "start_at": "2026-09-20T19:30:00+08:00", "capacity": 50,
+    })
+    assert item["id"]
+    _assert_physical_row("events", item["id"])
+    assert any(x.get("id") == item["id"] for x in training.list_activities())
+
+
+def test_admin_routes_gated(training_db):
+    # /admin/surveys|policies|plans 应已注册且受 RBAC 门卫（匿名非 404）
+    _mp, admin = training_db
+    for path in ("/admin/surveys", "/admin/policies", "/admin/plans", "/admin/activities"):
+        res = admin.get(path)
+        assert res.status_code in (401, 403), f"{path} 未注册或无门卫: {res.status_code}"