feat(sse): SSE 推送替代前端轮询 - #133
Closed
xiaocheny214 wants to merge 79 commits into
Closed
Conversation
Describe the backend package layout (common/framework/app) and the server domain module split: user, project, asset, character with action/character_template/wearable subdomains, generation and media. All modules currently define abstract interfaces only.
Add the shared response/exception kernel in windup_common: Response/ListResponse unified bodies (HTTP always 200, business code in body), BizException base, ModelException for LLM calls, and the BizCode/ModelErrorType enums. Add the app-level global exception handlers that convert these into Response.fail.
Add windup_framework infrastructure abstractions: SQLAlchemy db base/session, config loaders (database/provider/storage), LLM provider protocols (chat/image/video) and Kodo object storage.
Add abstract service interfaces and domain models for the server modules: user, project, asset, character (with action/character_template/wearable subdomains), generation and media. Drop the now-obsolete .gitkeep placeholders.
Drop the placeholder health router and its registration in create_app; the web layer now revolves around the global exception handlers.
Expose the windup CLI entrypoint (windup_app.bootstrap.app:main) and add python-multipart for form/file uploads; refresh uv.lock accordingly.
Point the sub-entity references at the nested character subpackages (character.action / character_template / wearable) instead of the old top-level paths.
Remove unused imports flagged by ruff F401 in generation/interface.py and user/model.py.
将 validate-branch 与 validate-commits 从 backend.yml 拆到独立的 naming.yml。命名规范门禁不局限于 backend 范畴,应作为仓库级 CI 单独维护(PR review:@minorcell 建议)。 backend.yml 仅保留 lint-and-test;naming.yml 用独立 concurrency group 避免与 backend 共用 group 互相取消。
Add SqlAlchemyProjectService inheriting ProjectService; implements create/get/list/delete plus project_name_exists against SQLAlchemy. - interface: reconcile ProjectService to session-per-call; session is passed per request via get_session, create_project takes **fields. - service: stateless impl as a module-level `service` singleton; flush only, commit/rollback handled by get_session; list ordered by id desc. - model: id uses BigInteger().with_variant(Integer, "sqlite") so the in-memory SQLite test DB autoincrements; Postgres unchanged. - api/bootstrap: import service from the service module, register project_router, add a print_banner startup stub.
Add SQLite in-memory test fixtures and project module tests. - conftest: SQLite in-memory fixtures (StaticPool, single connection) with no Docker Postgres dependency; get_session override for the API TestClient. - test_project_service: 14 unit tests for create/exists/get/list/delete covering the unique constraint, pagination, id-desc ordering and user filter. - test_project_api: 9 end-to-end tests via TestClient covering CRUD and the unified response contract (HTTP 200, code in body, 400/404).
Reflect character/media/generation/user module redesigns in the split document.
Replace strategy pattern with typed submit methods.
- model: add CharacterImageOutput/CharacterActionOutput/CharacterActionFrame
as typed task results, remove generic GenerationResult
- interface: split submit into generate_character_image and
generate_character_action with typed inputs
- document SSE streaming flow (GET /generation/tasks/{id}/stream)
replacing frontend polling
Remove OAuth abstract methods and models from UserService. Commented-out methods: get_oauth_authorize_url, login_by_oauth, bind_oauth, get_oauth_bindings. Related imports OAuthCallbackInput and UserOAuth removed from interface. The OAuth region is preserved as a placeholder for future extension.
Add ObjectStorageMediaService implementing MediaService. - service: upload to Kodo via KodoStorage adapter, auto-generated object keys with category prefix, no user filenames in keys - app: register media router (POST /media/upload) in create_app
refactor:asset character media
ci: split naming gate into its own workflow
feat:project module
跨层契约(common,无内部依赖):ActionType/GenRoute/CharacterCard/ActionSpec/AssetPackageRef。ai_engine 与 app 均依赖此。Refs #53。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
SufyVideoProvider(kling i2v,OpenAI 兼容异步 /videos)+ OnnxU2NetMatteProvider(onnxruntime 直跑 u2netp,替代 rembg 在 3.12 无轮子的 numba 链)+ 能力接口(Image/Video/MatteProvider)。同时声明 framework 代码实际 import 的依赖(langchain-openai/onnxruntime/qiniu/pillow/numpy)——补齐 #64 机器审指出的缺依赖。Refs #53 #35。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
填充空骨架:slicing(抽帧 + 周期/一次性选帧)、strategy(ROUTE_MATRIX 分流 + VideoFrameStrategy 图生视频)、prompt(walk/jump/attack/idle 提示词库)、postprocess(像素化 + 脚线对齐 + 打包 + 逐帧时长)、generator 串联。MVP:root_motion 暂不做(character_data.frames 无该字段,见 #63);graph/ 留占位非本 PR。Refs #53 #35 #21。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
- Remove dead storage module (empty __init__.py and kodo.py) from framework - Implement actual Qiniu Kodo SDK upload in ObjectStorageMediaService - Move MediaCategory to windup_common.enums.media, remove duplicate from media/model - Update media __init__ to import MediaCategory from common
Move qiniu import inside upload() so module collection does not fail when qiniu is not installed (e.g. CI test runs).
按 MVP 边界(与作者对齐):CharacterGeneratorPort.generate(card, action, master, progress) → GeneratedAction(frames/durations/fps);母版由 server 以 bytes 传入,不再经 ArtifactStore;去掉 sprite sheet 打包与存储引用(上传/落库/导出在 server/export 侧)。补 VideoFrameStrategy.derive 离线端到端测试(monkeypatch 抽帧,不解码 mp4/不联网)。Refs #53。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
refactor(media): implement Qiniu upload and deduplicate MediaCategory
backend-dev 的 ActionType.JUMP 值是大写 "JUMP",与其它小写成员不一致,且 executor 按值映射到 ai_engine 的 JUMP="jump" 会失败;改为 "jump"。(原 #24 另加一个 JUMP="jump" 导致枚举'JUMP already defined'、pytest 挂——改成修正现有值,不重复定义。)另 ROUTE_MATRIX 把 idle 从空桩 PROC_IDLE 改走 VIDEO_I2V(build_idle_prompt 躯干呼吸)。现 walk/idle/attack/jump 四动作全走视频 i2v。Refs #53 #35。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
… sprite_sample_url
feat(generation): 启用 jump 动作 + idle 走 i2v(四动作全跑通)
idle: 提示词原写 "chest and shoulders rise and fall",直接让肩膀起伏, i2v 出的是耸肩不是呼吸。改为只让胸腔/躯干在慢节律里起伏、肩膀锁平同高, 并去掉亚阈值幅度词(barely/faintly),过 prompt_lint。 jump: 重甲带剑角色的跳跃强动作先验压不住——站立母版让 i2v 只举剑摆造型、 不腾空(同 attack 用站立母版会抡过头顶的老坑)。按 master_prep 既有铁律 "母版姿态决定动作,提示词只能微调",给 jump 在 MASTER_POSES 补极限蓄力 半蹲母版姿态;runtime 的 prepare_master 仍只补顶部留白,此项为纯 spec。 prompt_lint 通过。on-screen 效果复验待 qnaigc 视频源站从 522 恢复后补跑。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
fix(ai_engine): idle 去耸肩、jump 补蓄力母版压站姿先验
该包职责是**生成任务的编排调度**(建 PENDING 记录 → 后台 executor 驱动 → 查询状态), 实际 AI 生成在 ai_engine;原名 generation 易与"生成本身"混淆,更名为 orchestrator。 - git mv server/generation → server/orchestrator(rename 保留历史) - 全部 windup_app.server.generation → .orchestrator(bootstrap / web api / 包内自引用 / 测试) - 分层契约不变(约束在 windup_app.server 层级,不牵子包名) 门禁:ruff / lint-imports(2 kept) / 编排测试(3 passed) 全绿。类名(GenerationTask 等) 与 web 路由不动(仍描述生成任务,改动最小)。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…n-orchestrator refactor(app): generation 包更名为 orchestrator(生成任务编排/调度)
- Add name varchar(20) column to windup_character ORM model - Add name field to CharacterCreate, CharacterUpdate, CharacterOut schemas
feat(character): add name field to character model and API schemas
- Add backend/Dockerfile with multi-stage build (uv + Python 3.12) - Add docker-compose.yml with backend and PostgreSQL services - Add db/init.sql for automatic database table initialization - Add .env.example with configuration template - PostgreSQL configured with port 7856 and secure password
…browser 三处让部署跑不起来的问题，都在这台服务器上实测定位： 1. 构建阶段 uv sync 超时。宿主机访问 pypi.org 需 8s，构建容器内默认超时会在 下载大包（uvloop）时 "operation timed out" 直接失败。改走国内镜像源并把 UV_HTTP_TIMEOUT 拉到 180s。 2. 容器起来即反复重启，报 "exec /app/.venv/bin/uvicorn: no such file or directory"。 文件其实存在，报的是它 shebang 指向的解释器——uv 装出来的 venv 里 shebang 与 .pth 都是绝对路径，builder 在 /build、runtime 在 /app，跨路径拷贝后解释器与 workspace 包全部失效。把 builder 的 WORKDIR 也改成 /app 即可。 3. 七牛上传 TLS 握手超时、媒体上传请求挂死。宿主机网卡 MTU 1480，而 compose 自建网络不继承 daemon 的 mtu 设置、默认仍是 1500，大包被丢。显式给网络设 1450 后，up-z0.qiniup.com 从握手超时 14s 变为 1.0s，上传恢复正常。 4. 浏览器跨域被全部拦下：OPTIONS 预检返回 405、响应无 access-control-* 头， 后端日志里连请求都看不到。挂上 CORSMiddleware，允许来源用 WINDUP_CORS_ORIGINS 覆盖，并放行 Vercel 预览域名。 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
…imiting - Add User ORM model and service implementation - Add JWT dual token (access + refresh) authentication - Add email verification via Resend - Add Redis-based rate limiting middleware - Add auth middleware with whitelist - Add /auth/* API endpoints (register, login, send-code, refresh, logout, me, change-password) - Add unit tests for user service (25 tests passing) - Move Redis client to db package for better organization - Add dependencies: redis, resend, passlib[bcrypt], pydantic[email]
- Add auth_client fixture with valid JWT token - Update test_project_api.py to use auth_client - Fix CI failures caused by auth middleware blocking unauthenticated requests
- Rate limit middleware now skips checks when Redis is not available - Prevents CI failures in environments without Redis - Logs warning when Redis connection fails
fix(deploy): 修好容器构建与浏览器可达性（服务器实测定位）
feat:user module
视频下载是"钱已经花完"之后的最后一步,却是单次读取、无重试、不校验长度:
`client.get(url).raise_for_status().content`。2026-08-05 实测同一角色连续两单
都死在这里,各烧掉一次生成费用:
peer closed connection without sending complete message body
(received 720450 bytes, expected 929531)
重试对成品 URL 的 GET 是幂等的、不再计费——代价是一次重下,不重试的代价是一次重新生成。
另校验 Content-Length:截断不一定抛异常,放过去的坏视频要到出帧环节才暴露成"解码失败",
很难回溯到下载这一步;分块传输无该头时跳过校验。
回归测试用 httpx MockTransport,不联网。已用旧实现做控制样本验证:
三条断言(重试/拒绝截断/包装最终原因)在修复前确实失败。
Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
fix(providers): 视频成品下载加重试与长度校验（Refs #129）
- generation.py 新增 _EventBus 内存发布订阅 + GET /tasks/{task_id}/stream 端点
- task_repo.update_status/update_result 后自动推送完整 task 数据到 EventBus
- bootstrap 延迟绑定 EventBus 到 task_repo
- SSE 端点支持终态检查：已终态任务立即推送并关闭连接
- 8 个单元测试覆盖 EventBus 核心逻辑
|
The latest updates on your projects. Learn more about Vercel for GitHub. |
![fennoai[bot]](https://avatars.githubusercontent.com/in/1924301?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
I found a few high-impact issues in the new generation/SSE path: task execution can race the transaction that creates the task, task reads are not actually scoped by project_id, and the in-memory SSE bus is written from worker threads through non-thread-safe asyncio queues.
| task = generation_service.generate_character_image( | ||
| session, user_id=body.user_id, project_id=body.project_id, input=input_data, | ||
| ) | ||
| threading.Thread( |
Contributor
There was a problem hiding this comment.
Starting the worker thread here races the request transaction. generate_character_image() only flush()es the new row; get_session() commits after the endpoint returns. The worker opens its own session immediately, so on PostgreSQL it can run update_status() before the task row is visible, making that write a no-op. For fast failures or image generation paths, later writes can be lost too, leaving a submitted task stuck in pending. Commit before dispatching, or enqueue the work after the dependency transaction has committed.
| ) | ||
| # 后台线程自开 session 跑生成(经项目约束 → ai_engine)。调度器由 bootstrap 注入 | ||
| # app.state,web 不静态依赖 ai_engine(满足入口层门禁)。 | ||
| threading.Thread( |
Contributor
There was a problem hiding this comment.
This has the same transaction race as the image endpoint: the action worker can start before the request-scoped session commits the newly created task row. Its first task_repo.update_status() uses a separate session and returns if session.get() cannot see the uncommitted row, so clients may never receive the running update and quick failures/completions can be dropped. Dispatch the thread only after the create transaction is committed.
| def get_task( | ||
| self, session: Session, project_id: int, task_id: int, | ||
| ) -> GenerationTask | None: | ||
| return task_repo.get_task(session, task_id) |
Contributor
There was a problem hiding this comment.
project_id is accepted by the public GET /generation/tasks/{task_id} and SSE endpoints, but this method ignores it and reads by task id alone. That means any authenticated caller who can guess or observe a task id can fetch another project's task payload/result despite passing an unrelated project_id; the stream endpoint also subscribes by bare task id after this lookup. Please query by both task_id and project_id (and ideally the authenticated user) before returning or subscribing.
|
|
||
| def publish(self, task_id: int, event: str, data: dict) -> None: | ||
| for queue in self._queues.get(str(task_id), []): | ||
| queue.put_nowait((event, data)) |
Contributor
There was a problem hiding this comment.
publish() is called from the background threading.Thread executors, but these subscriber queues are asyncio.Queues owned by the request event loop. asyncio.Queue/its waiters are not thread-safe; calling put_nowait() from the worker thread can fail to wake the SSE coroutine or corrupt the queue state under load. Use the loop captured at subscription time with loop.call_soon_threadsafe(queue.put_nowait, ...), or move publication onto a thread-safe queue/anyio channel.
Member
|
Member
|
@xiaocheny214 此外，这个修改应该和前端的同学同步下，可以 艾特 到人的。一起来 review 下； |
This was referenced Aug 6, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
概述
内嵌 EventBus 到 generation 模块，通过 SSE 实时推送任务状态，替代前端 2s 间隔轮询。
改动
generation.py新增_EventBus内存发布订阅 +GET /generation/tasks/{task_id}/streamSSE 端点task_repo.py状态变更（update_status/update_result）后自动推送完整 task 数据到 EventBusbootstrap/app.py延迟绑定 EventBus 到 task_repo关联