feat(generation): align validated SSE adapter with backend tasks

backend/packages/app/src/windup_app/web/api/generation.py

@@ -29,10 +29,12 @@
 from windup_common.result import Response
 from windup_framework.db import get_session
 
+from windup_app.server.character.model import Character
 from windup_app.server.orchestrator.model import (
     ActionType,
     GenerationTask,
 )
+from windup_app.server.project.model import Project
 
 logger = logging.getLogger("windup.generation.api")
 
@@ -54,23 +56,34 @@ class _EventBus:
     """任务进度内存发布-订阅。"""
 
     def __init__(self) -> None:
-        self._queues: dict[str, list[asyncio.Queue]] = defaultdict(list)
+        self._queues: dict[tuple[int, int], list[asyncio.Queue]] = defaultdict(list)
 
-    async def subscribe(self, task_id: int) -> asyncio.Queue:
+    async def subscribe(self, project_id: int, task_id: int) -> asyncio.Queue:
         queue: asyncio.Queue = asyncio.Queue()
-        self._queues[str(task_id)].append(queue)
+        self._queues[(project_id, task_id)].append(queue)
         return queue
 
-    async def unsubscribe(self, task_id: int, queue: asyncio.Queue) -> None:
-        key = str(task_id)
+    async def unsubscribe(
+        self,
+        project_id: int,
+        task_id: int,
+        queue: asyncio.Queue,
+    ) -> None:
+        key = (project_id, task_id)
         subs = self._queues.get(key)
         if subs and queue in subs:
             subs.remove(queue)
             if not subs:
                 del self._queues[key]
 
-    def publish(self, task_id: int, event: str, data: dict) -> None:
-        for queue in self._queues.get(str(task_id), []):
+    def publish(
+        self,
+        project_id: int,
+        task_id: int,
+        event: str,
+        data: dict,
+    ) -> None:
+        for queue in self._queues.get((project_id, task_id), []):
             queue.put_nowait((event, data))
 
 
@@ -86,8 +99,7 @@ def publish(self, task_id: int, event: str, data: dict) -> None:
 class CharacterImageGenerateRequest(BaseModel):
     """提交角色图片生成任务。"""
 
-    user_id: int = Field(gt=0)
-    project_id: int | None = None
+    project_id: int = Field(gt=0)
     reference_image_url: str | None = None
     prompt: str = ""
     negative_prompt: str = ""
@@ -99,8 +111,7 @@ class CharacterImageGenerateRequest(BaseModel):
 class CharacterActionGenerateRequest(BaseModel):
     """提交角色动作生成任务。"""
 
-    user_id: int = Field(gt=0)
-    project_id: int | None = None
+    project_id: int = Field(gt=0)
     character_id: int = Field(gt=0)
     action_type: ActionType
     custom_prompt: str | None = None
@@ -115,7 +126,6 @@ class GenerationTaskOut(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
-    user_id: int
     project_id: int | None = None
     task_type: str
     status: str
@@ -131,7 +141,6 @@ def _task_to_out(task: GenerationTask) -> GenerationTaskOut:
         result_dict = dataclasses.asdict(task.result)
     return GenerationTaskOut(
         id=task.id,
-        user_id=task.user_id,
         project_id=task.project_id,
         task_type=task.task_type.value,
         status=task.status.value,
@@ -146,15 +155,32 @@ def _task_to_out(task: GenerationTask) -> GenerationTaskOut:
 # ══════════════════════════════════════════════════════════════════════════════
 
 
-def _validate_project_size(session: Session, project_id: int | None, width: int, height: int) -> None:
-    """校验输入尺寸与项目约束是否一致;不一致则抛异常。"""
-    if project_id is None:
-        return
-    from windup_app.server.project.service import SqlAlchemyProjectService
+def _get_project_or_raise(
+    session: Session,
+    project_id: int,
+    user_id: int,
+) -> Project:
+    """校验项目存在且属于 token 对应用户。"""
+    project = session.get(Project, project_id)
+    if project is None or project.user_id != user_id:
+        raise BizException("项目不存在", code=BizCode.NOT_FOUND)
+    return project
+
 
-    project = SqlAlchemyProjectService().get_project(session, project_id)
-    if project is None:
-        return
+def _get_character_or_raise(
+    session: Session,
+    character_id: int,
+    project_id: int,
+) -> Character:
+    """校验角色存在且属于本次生成所指定的项目。"""
+    character = session.get(Character, character_id)
+    if character is None or character.project_id != project_id:
+        raise BizException("角色不存在", code=BizCode.NOT_FOUND)
+    return character
+
+
+def _validate_project_size(project: Project, width: int, height: int) -> None:
+    """校验输入尺寸与项目约束是否一致;不一致则抛异常。"""
     if width != project.sprite_width or height != project.sprite_height:
         raise BizException(
             f"输入尺寸 {width}×{height} 与项目约束 {project.sprite_width}×{project.sprite_height} 不一致",
@@ -169,7 +195,9 @@ def submit_image_generation(
     session: Session = Depends(get_session),
 ) -> Response[GenerationTaskOut]:
     """提交角色图片生成任务:建 PENDING 记录立即返回,实际图生图后台跑。"""
-    _validate_project_size(session, body.project_id, body.width, body.height)
+    user_id = request.state.current_user.id
+    project = _get_project_or_raise(session, body.project_id, user_id)
+    _validate_project_size(project, body.width, body.height)
     # TODO: service.create_image_task + background_tasks.add_task
     raise BizException("接口待实现", code=BizCode.BAD_REQUEST)
 
@@ -181,6 +209,9 @@ def submit_action_generation(
     session: Session = Depends(get_session),
 ) -> Response[GenerationTaskOut]:
     """提交角色动作生成任务:建 PENDING 记录立即返回,实际生成后台跑。"""
+    user_id = request.state.current_user.id
+    _get_project_or_raise(session, body.project_id, user_id)
+    _get_character_or_raise(session, body.character_id, body.project_id)
     # TODO: service.create_action_task + background_tasks.add_task
     raise BizException("接口待实现", code=BizCode.BAD_REQUEST)
 
@@ -189,10 +220,13 @@ def submit_action_generation(
 def get_task(
     task_id: int,
     project_id: int = Query(..., gt=0),
+    request: Request = None,
     session: Session = Depends(get_session),
 ) -> Response[GenerationTaskOut]:
     """查询生成任务状态与结果。"""
-    # TODO: service.get_task
+    user_id = request.state.current_user.id
+    _get_project_or_raise(session, project_id, user_id)
+    # TODO: service.get_task，并校验任务属于 project_id
     raise BizException("接口待实现", code=BizCode.BAD_REQUEST)
 
 
@@ -212,8 +246,10 @@ async def stream_task(
 
     若客户端订阅时任务已处于终态,立即推送终态事件并关闭连接。
     """
-    # TODO: 检查任务初始状态,若已终态立即推送
-    queue = await event_bus.subscribe(task_id)
+    user_id = request.state.current_user.id
+    _get_project_or_raise(session, project_id, user_id)
+    # TODO: 检查任务属于 project_id 及初始状态,若已终态立即推送
+    queue = await event_bus.subscribe(project_id, task_id)
     logger.debug("SSE 订阅: task_id=%d", task_id)
 
     async def _event_generator():
@@ -224,7 +260,8 @@ async def _event_generator():
                     break
                 try:
                     event, data = await asyncio.wait_for(
-                        queue.get(), timeout=_HEARTBEAT_TIMEOUT,
+                        queue.get(),
+                        timeout=_HEARTBEAT_TIMEOUT,
                     )
                     payload = json.dumps(data, ensure_ascii=False)
                     yield f"event: {event}\ndata: {payload}\n\n"
@@ -234,7 +271,7 @@ async def _event_generator():
                 except asyncio.TimeoutError:
                     yield ": heartbeat\n\n"
         finally:
-            await event_bus.unsubscribe(task_id, queue)
+            await event_bus.unsubscribe(project_id, task_id, queue)
             logger.debug("SSE 取消订阅: task_id=%d", task_id)
 
     return StreamingResponse(

backend/packages/app/src/windup_app/web/api/project.py

@@ -39,7 +39,6 @@ class ProjectOut(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
-    user_id: int
     workflow_id: int | None
     project_name: str
     character_perspective: int
@@ -64,15 +63,17 @@ def create_project(
     ):
         logger.warning(
             "[WINDUP] 创建拒绝-名称重复 | user_id=%s project_name=%s",
-            user_id, body.project_name,
+            user_id,
+            body.project_name,
         )
         raise BizException("项目名称已存在", code=BizCode.BAD_REQUEST)
     try:
         project = service.create_project(session, user_id=user_id, **body.model_dump())
     except IntegrityError:
         logger.warning(
             "[WINDUP] 创建拒绝-并发冲突 | user_id=%s project_name=%s",
-            user_id, body.project_name,
+            user_id,
+            body.project_name,
         )
         session.rollback()
         raise BizException("项目名称已存在", code=BizCode.BAD_REQUEST) from None

backend/tests/test_generation_api.py

@@ -0,0 +1,149 @@
+"""生成任务 API 的认证与资源归属测试。"""
+
+import asyncio
+
+from windup_app.web.api.generation import GenerationTaskOut, _EventBus
+
+
+def _create_project(auth_client, name: str = "生成项目") -> dict:
+    return auth_client.post(
+        "/projects",
+        json={
+            "project_name": name,
+            "character_perspective": 1,
+            "directional_movement": 2,
+            "sprite_width": 64,
+            "sprite_height": 64,
+        },
+    ).json()["data"]
+
+
+def _create_character(auth_client, project_id: int) -> dict:
+    return auth_client.post(
+        "/characters",
+        json={
+            "project_id": project_id,
+            "workflow_run_id": 1,
+            "name": "勇者",
+        },
+    ).json()["data"]
+
+
+def _image_payload(project_id: int, **overrides) -> dict:
+    payload = {
+        "project_id": project_id,
+        "prompt": "像素风勇者",
+        "width": 64,
+        "height": 64,
+    }
+    payload.update(overrides)
+    return payload
+
+
+def _action_payload(project_id: int, character_id: int, **overrides) -> dict:
+    payload = {
+        "project_id": project_id,
+        "character_id": character_id,
+        "action_type": "walk",
+    }
+    payload.update(overrides)
+    return payload
+
+
+def test_image_generation_uses_token_user_without_body_user_id(auth_client):
+    project = _create_project(auth_client)
+
+    response = auth_client.post(
+        "/generation/image",
+        json=_image_payload(project["id"]),
+    )
+
+    assert response.json()["code"] == 400
+    assert response.json()["message"] == "接口待实现"
+
+
+def test_spoofed_body_user_id_cannot_access_other_users_project(
+    auth_client,
+    auth_client_b,
+):
+    project = _create_project(auth_client)
+
+    response = auth_client_b.post(
+        "/generation/image",
+        json=_image_payload(project["id"], user_id=1),
+    )
+
+    assert response.json()["code"] == 404
+    assert response.json()["message"] == "项目不存在"
+
+
+def test_action_generation_uses_token_user_without_body_user_id(auth_client):
+    project = _create_project(auth_client)
+    character = _create_character(auth_client, project["id"])
+
+    response = auth_client.post(
+        "/generation/action",
+        json=_action_payload(project["id"], character["id"]),
+    )
+
+    assert response.json()["code"] == 400
+    assert response.json()["message"] == "接口待实现"
+
+
+def test_action_character_must_belong_to_requested_project(auth_client):
+    first_project = _create_project(auth_client, "项目一")
+    second_project = _create_project(auth_client, "项目二")
+    character = _create_character(auth_client, first_project["id"])
+
+    response = auth_client.post(
+        "/generation/action",
+        json=_action_payload(second_project["id"], character["id"]),
+    )
+
+    assert response.json()["code"] == 404
+    assert response.json()["message"] == "角色不存在"
+
+
+def test_task_query_checks_project_ownership(auth_client, auth_client_b):
+    project = _create_project(auth_client)
+
+    response = auth_client_b.get(
+        "/generation/tasks/1",
+        params={"project_id": project["id"]},
+    )
+
+    assert response.json()["code"] == 404
+    assert response.json()["message"] == "项目不存在"
+
+
+def test_task_stream_checks_project_ownership(auth_client, auth_client_b):
+    project = _create_project(auth_client)
+
+    response = auth_client_b.get(
+        "/generation/tasks/1/stream",
+        params={"project_id": project["id"]},
+    )
+
+    assert response.json()["code"] == 404
+    assert response.json()["message"] == "项目不存在"
+
+
+def test_event_bus_isolates_same_task_id_between_projects():
+    async def scenario():
+        bus = _EventBus()
+        first_queue = await bus.subscribe(1, 9)
+        second_queue = await bus.subscribe(2, 9)
+
+        bus.publish(1, 9, "progress", {"status": "running"})
+
+        assert first_queue.get_nowait() == (
+            "progress",
+            {"status": "running"},
+        )
+        assert second_queue.empty()
+
+    asyncio.run(scenario())
+
+
+def test_generation_response_contract_does_not_expose_user_id():
+    assert "user_id" not in GenerationTaskOut.model_fields